Chapter 6. Deploying AMQ Streams using installation artifacts Having prepared your environment for a deployment of AMQ Streams , you can deploy AMQ Streams to an OpenShift cluster. You can use the deployment files provided with the release artifacts. Use the deployment files to create the Kafka cluster . Optionally, you can deploy the following Kafka components according to your requirements: Kafka Connect Kafka MirrorMaker Kafka Bridge AMQ Streams is based on Strimzi 0.28.x. You can deploy AMQ Streams 2.1 on OpenShift 4.6 to 4.10. Note To run the commands in this guide, your cluster user must have the rights to manage role-based access control (RBAC) and CRDs. 6.1. Create the Kafka cluster To be able to manage a Kafka cluster with the Cluster Operator, you must deploy it as a Kafka resource. AMQ Streams provides example deployment files to do this. You can use these files to deploy the Topic Operator and User Operator at the same time. If you haven't deployed a Kafka cluster as a Kafka resource, you can't use the Cluster Operator to manage it. This applies, for example, to a Kafka cluster running outside of OpenShift. But you can deploy and use the Topic Operator and User Operator as standalone components. Note The Cluster Operator can watch one, multiple, or all namespaces in an OpenShift cluster. The Topic Operator and User Operator watch for KafkaTopics and KafkaUsers in the single namespace of the Kafka cluster deployment. Deploying a Kafka cluster with the Topic Operator and User Operator Perform these deployment steps if you want to use the Topic Operator and User Operator with a Kafka cluster managed by AMQ Streams. Deploy the Cluster Operator Use the Cluster Operator to deploy the: Kafka cluster Topic Operator User Operator Deploying a standalone Topic Operator and User Operator Perform these deployment steps if you want to use the Topic Operator and User Operator with a Kafka cluster that is not managed by AMQ Streams. Deploy the standalone Topic Operator Deploy the standalone User Operator 6.1.1. Deploying the Cluster Operator The Cluster Operator is responsible for deploying and managing Apache Kafka clusters within an OpenShift cluster. The procedures in this section describe how to deploy the Cluster Operator to watch one of the following: A single namespace Multiple namespaces All namespaces 6.1.1.1. Watch options for a Cluster Operator deployment When the Cluster Operator is running, it starts to watch for updates of Kafka resources. You can choose to deploy the Cluster Operator to watch Kafka resources from: A single namespace (the same namespace containing the Cluster Operator) Multiple namespaces All namespaces Note AMQ Streams provides example YAML files to make the deployment process easier. The Cluster Operator watches for changes to the following resources: Kafka for the Kafka cluster. KafkaConnect for the Kafka Connect cluster. KafkaConnector for creating and managing connectors in a Kafka Connect cluster. KafkaMirrorMaker for the Kafka MirrorMaker instance. KafkaMirrorMaker2 for the Kafka MirrorMaker 2.0 instance. KafkaBridge for the Kafka Bridge instance. KafkaRebalance for the Cruise Control optimization requests. When one of these resources is created in the OpenShift cluster, the operator gets the cluster description from the resource and starts creating a new cluster for the resource by creating the necessary OpenShift resources, such as StatefulSets, Services and ConfigMaps. Each time a Kafka resource is updated, the operator performs corresponding updates on the OpenShift resources that make up the cluster for the resource. Resources are either patched or deleted, and then recreated in order to make the cluster for the resource reflect the desired state of the cluster. This operation might cause a rolling update that might lead to service disruption. When a resource is deleted, the operator undeploys the cluster and deletes all related OpenShift resources. 6.1.1.2. Deploying the Cluster Operator to watch a single namespace This procedure shows how to deploy the Cluster Operator to watch AMQ Streams resources in a single namespace in your OpenShift cluster. Prerequisites This procedure requires use of an OpenShift user account which is able to create CustomResourceDefinitions , ClusterRoles and ClusterRoleBindings . Use of Role Base Access Control (RBAC) in the OpenShift cluster usually means that permission to create, edit, and delete these resources is limited to OpenShift cluster administrators, such as system:admin . Procedure Edit the AMQ Streams installation files to use the namespace the Cluster Operator is going to be installed into. For example, in this procedure the Cluster Operator is installed into the namespace <my_cluster_operator_namespace> . On Linux, use: On MacOS, use: Deploy the Cluster Operator: oc create -f install/cluster-operator -n <my_cluster_operator_namespace> Check the status of the deployment: oc get deployments -n <my_cluster_operator_namespace> Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1 READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . 6.1.1.3. Deploying the Cluster Operator to watch multiple namespaces This procedure shows how to deploy the Cluster Operator to watch AMQ Streams resources across multiple namespaces in your OpenShift cluster. Prerequisites This procedure requires use of an OpenShift user account which is able to create CustomResourceDefinitions , ClusterRoles and ClusterRoleBindings . Use of Role Base Access Control (RBAC) in the OpenShift cluster usually means that permission to create, edit, and delete these resources is limited to OpenShift cluster administrators, such as system:admin . Procedure Edit the AMQ Streams installation files to use the namespace the Cluster Operator is going to be installed into. For example, in this procedure the Cluster Operator is installed into the namespace <my_cluster_operator_namespace> . On Linux, use: On MacOS, use: Edit the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to add a list of all the namespaces the Cluster Operator will watch to the STRIMZI_NAMESPACE environment variable. For example, in this procedure the Cluster Operator will watch the namespaces watched-namespace-1 , watched-namespace-2 , watched-namespace-3 . apiVersion: apps/v1 kind: Deployment spec: # ... template: spec: serviceAccountName: strimzi-cluster-operator containers: - name: strimzi-cluster-operator image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.1.0 imagePullPolicy: IfNotPresent env: - name: STRIMZI_NAMESPACE value: watched-namespace-1,watched-namespace-2,watched-namespace-3 For each namespace listed, install the RoleBindings . In this example, we replace watched-namespace in these commands with the namespaces listed in the step, repeating them for watched-namespace-1 , watched-namespace-2 , watched-namespace-3 : oc create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace> oc create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace> Deploy the Cluster Operator: oc create -f install/cluster-operator -n <my_cluster_operator_namespace> Check the status of the deployment: oc get deployments -n <my_cluster_operator_namespace> Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1 READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . 6.1.1.4. Deploying the Cluster Operator to watch all namespaces This procedure shows how to deploy the Cluster Operator to watch AMQ Streams resources across all namespaces in your OpenShift cluster. When running in this mode, the Cluster Operator automatically manages clusters in any new namespaces that are created. Prerequisites This procedure requires use of an OpenShift user account which is able to create CustomResourceDefinitions , ClusterRoles and ClusterRoleBindings . Use of Role Base Access Control (RBAC) in the OpenShift cluster usually means that permission to create, edit, and delete these resources is limited to OpenShift cluster administrators, such as system:admin . Procedure Edit the AMQ Streams installation files to use the namespace the Cluster Operator is going to be installed into. For example, in this procedure the Cluster Operator is installed into the namespace <my_cluster_operator_namespace> . On Linux, use: On MacOS, use: Edit the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file to set the value of the STRIMZI_NAMESPACE environment variable to * . apiVersion: apps/v1 kind: Deployment spec: # ... template: spec: # ... serviceAccountName: strimzi-cluster-operator containers: - name: strimzi-cluster-operator image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.1.0 imagePullPolicy: IfNotPresent env: - name: STRIMZI_NAMESPACE value: "*" # ... Create ClusterRoleBindings that grant cluster-wide access for all namespaces to the Cluster Operator. oc create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount <my_cluster_operator_namespace> :strimzi-cluster-operator oc create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount <my_cluster_operator_namespace> :strimzi-cluster-operator Replace <my_cluster_operator_namespace> with the namespace you want to install the Cluster Operator into. Deploy the Cluster Operator to your OpenShift cluster. oc create -f install/cluster-operator -n <my_cluster_operator_namespace> Check the status of the deployment: oc get deployments -n <my_cluster_operator_namespace> Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1 READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . 6.1.2. Deploying Kafka Apache Kafka is an open-source distributed publish-subscribe messaging system for fault-tolerant real-time data feeds. The procedures in this section describe the following: How to use the Cluster Operator to deploy: An ephemeral or persistent Kafka cluster The Topic Operator and User Operator by configuring the Kafka custom resource: Topic Operator User Operator Alternative standalone deployment procedures for the Topic Operator and User Operator: Deploy the standalone Topic Operator Deploy the standalone User Operator When installing Kafka, AMQ Streams also installs a ZooKeeper cluster and adds the necessary configuration to connect Kafka with ZooKeeper. 6.1.2.1. Deploying the Kafka cluster This procedure shows how to deploy a Kafka cluster to your OpenShift cluster using the Cluster Operator. The deployment uses a YAML file to provide the specification to create a Kafka resource. AMQ Streams provides example configuration files . For a Kafka deployment, the following examples are provided: kafka-persistent.yaml Deploys a persistent cluster with three ZooKeeper and three Kafka nodes. kafka-jbod.yaml Deploys a persistent cluster with three ZooKeeper and three Kafka nodes (each using multiple persistent volumes). kafka-persistent-single.yaml Deploys a persistent cluster with a single ZooKeeper node and a single Kafka node. kafka-ephemeral.yaml Deploys an ephemeral cluster with three ZooKeeper and three Kafka nodes. kafka-ephemeral-single.yaml Deploys an ephemeral cluster with three ZooKeeper nodes and a single Kafka node. In this procedure, we use the examples for an ephemeral and persistent Kafka cluster deployment. Ephemeral cluster In general, an ephemeral (or temporary) Kafka cluster is suitable for development and testing purposes, not for production. This deployment uses emptyDir volumes for storing broker information (for ZooKeeper) and topics or partitions (for Kafka). Using an emptyDir volume means that its content is strictly related to the pod life cycle and is deleted when the pod goes down. Persistent cluster A persistent Kafka cluster uses PersistentVolumes to store ZooKeeper and Kafka data. The PersistentVolume is acquired using a PersistentVolumeClaim to make it independent of the actual type of the PersistentVolume . For example, it can use Amazon EBS volumes in Amazon AWS deployments without any changes in the YAML files. The PersistentVolumeClaim can use a StorageClass to trigger automatic volume provisioning. The example YAML files specify the latest supported Kafka version, and configuration for its supported log message format version and inter-broker protocol version. The inter.broker.protocol.version property for the Kafka config must be the version supported by the specified Kafka version ( spec.kafka.version ). The property represents the version of Kafka protocol used in a Kafka cluster. From Kafka 3.0.0, when the inter.broker.protocol.version is set to 3.0 or higher, the log.message.format.version option is ignored and doesn't need to be set. An update to the inter.broker.protocol.version is required when upgrading Kafka . The example clusters are named my-cluster by default. The cluster name is defined by the name of the resource and cannot be changed after the cluster has been deployed. To change the cluster name before you deploy the cluster, edit the Kafka.metadata.name property of the Kafka resource in the relevant YAML file. Default cluster name and specified Kafka versions apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: kafka: version: 3.1.0 #... config: #... log.message.format.version: "3.1" inter.broker.protocol.version: "3.1" # ... Prerequisites The Cluster Operator must be deployed. Procedure Create and deploy an ephemeral or persistent cluster. For development or testing, you might prefer to use an ephemeral cluster. You can use a persistent cluster in any situation. To create and deploy an ephemeral cluster: oc apply -f examples/kafka/kafka-ephemeral.yaml To create and deploy a persistent cluster: oc apply -f examples/kafka/kafka-persistent.yaml Check the status of the deployment: oc get pods -n <my_cluster_operator_namespace> Output shows the pod names and readiness NAME READY STATUS RESTARTS my-cluster-entity-operator 3/3 Running 0 my-cluster-kafka-0 1/1 Running 0 my-cluster-kafka-1 1/1 Running 0 my-cluster-kafka-2 1/1 Running 0 my-cluster-zookeeper-0 1/1 Running 0 my-cluster-zookeeper-1 1/1 Running 0 my-cluster-zookeeper-2 1/1 Running 0 my-cluster is the name of the Kafka cluster. With the default deployment, you install an Entity Operator cluster, 3 Kafka pods, and 3 ZooKeeper pods. READY shows the number of replicas that are ready/expected. The deployment is successful when the STATUS shows as Running . Additional resources Kafka cluster configuration 6.1.2.2. Deploying the Topic Operator using the Cluster Operator This procedure describes how to deploy the Topic Operator using the Cluster Operator. You configure the entityOperator property of the Kafka resource to include the topicOperator . By default, the Topic Operator watches for KafkaTopics in the namespace of the Kafka cluster deployment. If you want to use the Topic Operator with a Kafka cluster that is not managed by AMQ Streams, you must deploy the Topic Operator as a standalone component . For more information about configuring the entityOperator and topicOperator properties, see Configuring the Entity Operator . Prerequisites The Cluster Operator must be deployed. Procedure Edit the entityOperator properties of the Kafka resource to include topicOperator : apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: #... entityOperator: topicOperator: {} userOperator: {} Configure the Topic Operator spec using the properties described in EntityTopicOperatorSpec schema reference . Use an empty object ( {} ) if you want all properties to use their default values. Create or update the resource: Use oc apply : oc apply -f <your-file> 6.1.2.3. Deploying the User Operator using the Cluster Operator This procedure describes how to deploy the User Operator using the Cluster Operator. You configure the entityOperator property of the Kafka resource to include the userOperator . By default, the User Operator watches for KafkaUsers in the namespace of the Kafka cluster deployment. If you want to use the User Operator with a Kafka cluster that is not managed by AMQ Streams, you must deploy the User Operator as a standalone component . For more information about configuring the entityOperator and userOperator properties, see Configuring the Entity Operator . Prerequisites The Cluster Operator must be deployed. Procedure Edit the entityOperator properties of the Kafka resource to include userOperator : apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: #... entityOperator: topicOperator: {} userOperator: {} Configure the User Operator spec using the properties described in EntityUserOperatorSpec schema reference . Use an empty object ( {} ) if you want all properties to use their default values. Create or update the resource: oc apply -f <your-file> 6.1.3. Alternative standalone deployment options for AMQ Streams Operators You can perform a standalone deployment of the Topic Operator and User Operator. Consider a standalone deployment of these operators if you are using a Kafka cluster that is not managed by the Cluster Operator. You deploy the operators to OpenShift. Kafka can be running outside of OpenShift. For example, you might be using a Kafka as a managed service. You adjust the deployment configuration for the standalone operator to match the address of your Kafka cluster. 6.1.3.1. Deploying the standalone Topic Operator This procedure shows how to deploy the Topic Operator as a standalone component for topic management. You can use a standalone Topic Operator with a Kafka cluster that is not managed by the Cluster Operator. A standalone deployment can operate with any Kafka cluster. Standalone deployment files are provided with AMQ Streams. Use the 05-Deployment-strimzi-topic-operator.yaml deployment file to deploy the Topic Operator. Add or set the environment variables needed to make a connection to a Kafka cluster. Prerequisites You are running a Kafka cluster for the Topic Operator to connect to. As long as the standalone Topic Operator is correctly configured for connection, the Kafka cluster can be running on a bare-metal environment, a virtual machine, or as a managed cloud application service. Procedure Edit the env properties in the install/topic-operator/05-Deployment-strimzi-topic-operator.yaml standalone deployment file. Example standalone Topic Operator deployment configuration apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-topic-operator labels: app: strimzi spec: # ... template: # ... spec: # ... containers: - name: strimzi-topic-operator # ... env: - name: STRIMZI_NAMESPACE 1 valueFrom: fieldRef: fieldPath: metadata.namespace - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2 value: my-kafka-bootstrap-address:9092 - name: STRIMZI_RESOURCE_LABELS 3 value: "strimzi.io/cluster=my-cluster" - name: STRIMZI_ZOOKEEPER_CONNECT 4 value: my-cluster-zookeeper-client:2181 - name: STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 5 value: "18000" - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6 value: "120000" - name: STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS 7 value: "6" - name: STRIMZI_LOG_LEVEL 8 value: INFO - name: STRIMZI_TLS_ENABLED 9 value: "false" - name: STRIMZI_JAVA_OPTS 10 value: "-Xmx=512M -Xms=256M" - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 11 value: "-Djavax.net.debug=verbose -DpropertyName=value" - name: STRIMZI_PUBLIC_CA 12 value: "false" - name: STRIMZI_TLS_AUTH_ENABLED 13 value: "false" - name: STRIMZI_SASL_ENABLED 14 value: "false" - name: STRIMZI_SASL_USERNAME 15 value: "admin" - name: STRIMZI_SASL_PASSWORD 16 value: "password" - name: STRIMZI_SASL_MECHANISM 17 value: "scram-sha-512" - name: STRIMZI_SECURITY_PROTOCOL 18 value: "SSL" 1 The OpenShift namespace for the Topic Operator to watch for KafkaTopic resources. Specify the namespace of the Kafka cluster. 2 The host and port pair of the bootstrap broker address to discover and connect to all brokers in the Kafka cluster. Use a comma-separated list to specify two or three broker addresses in case a server is down. 3 The label selector to identify the KafkaTopic resources managed by the Topic Operator. 4 The host and port pair of the address to connect to the ZooKeeper cluster. This must be the same ZooKeeper cluster that your Kafka cluster is using. 5 The ZooKeeper session timeout, in milliseconds. The default is 18000 (18 seconds). 6 The interval between periodic reconciliations, in milliseconds. The default is 120000 (2 minutes). 7 The number of attempts at getting topic metadata from Kafka. The time between each attempt is defined as an exponential backoff. Consider increasing this value when topic creation takes more time due to the number of partitions or replicas. The default is 6 attempts. 8 The level for printing logging messages. You can set the level to ERROR , WARNING , INFO , DEBUG , or TRACE . 9 Enables TLS support for encrypted communication with the Kafka brokers. 10 (Optional) The Java options used by the JVM running the Topic Operator. 11 (Optional) The debugging ( -D ) options set for the Topic Operator. 12 (Optional) Skips the generation of trust store certificates if TLS is enabled through STRIMZI_TLS_ENABLED . If this environment variable is enabled, the brokers must use a public trusted certificate authority for their TLS certificates. The default is false . 13 (Optional) Generates key store certificates for mutual TLS authentication. Setting this to false disables client authentication with TLS to the Kafka brokers. The default is true . 14 (Optional) Enables SASL support for client authentication when connecting to Kafka brokers. The default is false . 15 (Optional) The SASL username for client authentication. Mandatory only if SASL is enabled through STRIMZI_SASL_ENABLED . 16 (Optional) The SASL password for client authentication. Mandatory only if SASL is enabled through STRIMZI_SASL_ENABLED . 17 (Optional) The SASL mechanism for client authentication. Mandatory only if SASL is enabled through STRIMZI_SASL_ENABLED . You can set the value to plain , scram-sha-256 , or scram-sha-512 . 18 (Optional) The security protocol used for communication with Kafka brokers. The default value is "PLAINTEXT". You can set the value to PLAINTEXT , SSL , SASL_PLAINTEXT , or SASL_SSL . If you want to connect to Kafka brokers that are using certificates from a public certificate authority, set STRIMZI_PUBLIC_CA to true . Set this property to true , for example, if you are using Amazon AWS MSK service. If you enabled TLS with the STRIMZI_TLS_ENABLED environment variable, specify the keystore and truststore used to authenticate connection to the Kafka cluster. Example TLS configuration # .... env: - name: STRIMZI_TRUSTSTORE_LOCATION 1 value: "/path/to/truststore.p12" - name: STRIMZI_TRUSTSTORE_PASSWORD 2 value: " TRUSTSTORE-PASSWORD " - name: STRIMZI_KEYSTORE_LOCATION 3 value: "/path/to/keystore.p12" - name: STRIMZI_KEYSTORE_PASSWORD 4 value: " KEYSTORE-PASSWORD " # ... 1 The truststore contains the public keys of the Certificate Authorities used to sign the Kafka and ZooKeeper server certificates. 2 The password for accessing the truststore. 3 The keystore contains the private key for TLS client authentication. 4 The password for accessing the keystore. Deploy the Topic Operator. oc create -f install/topic-operator Check the status of the deployment: oc get deployments Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE strimzi-topic-operator 1/1 1 1 READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . 6.1.3.2. Deploying the standalone User Operator This procedure shows how to deploy the User Operator as a standalone component for user management. You can use a standalone User Operator with a Kafka cluster that is not managed by the Cluster Operator. A standalone deployment can operate with any Kafka cluster. Standalone deployment files are provided with AMQ Streams. Use the 05-Deployment-strimzi-user-operator.yaml deployment file to deploy the User Operator. Add or set the environment variables needed to make a connection to a Kafka cluster. Prerequisites You are running a Kafka cluster for the User Operator to connect to. As long as the standalone User Operator is correctly configured for connection, the Kafka cluster can be running on a bare-metal environment, a virtual machine, or as a managed cloud application service. Procedure Edit the following env properties in the install/user-operator/05-Deployment-strimzi-user-operator.yaml standalone deployment file. Example standalone User Operator deployment configuration apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-user-operator labels: app: strimzi spec: # ... template: # ... spec: # ... containers: - name: strimzi-user-operator # ... env: - name: STRIMZI_NAMESPACE 1 valueFrom: fieldRef: fieldPath: metadata.namespace - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2 value: my-kafka-bootstrap-address:9092 - name: STRIMZI_CA_CERT_NAME 3 value: my-cluster-clients-ca-cert - name: STRIMZI_CA_KEY_NAME 4 value: my-cluster-clients-ca - name: STRIMZI_LABELS 5 value: "strimzi.io/cluster=my-cluster" - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6 value: "120000" - name: STRIMZI_LOG_LEVEL 7 value: INFO - name: STRIMZI_GC_LOG_ENABLED 8 value: "true" - name: STRIMZI_CA_VALIDITY 9 value: "365" - name: STRIMZI_CA_RENEWAL 10 value: "30" - name: STRIMZI_JAVA_OPTS 11 value: "-Xmx=512M -Xms=256M" - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 12 value: "-Djavax.net.debug=verbose -DpropertyName=value" - name: STRIMZI_SECRET_PREFIX 13 value: "kafka-" - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 14 value: "true" 1 The OpenShift namespace for the User Operator to watch for KafkaUser resources. Only one namespace can be specified. 2 The host and port pair of the bootstrap broker address to discover and connect to all brokers in the Kafka cluster. Use a comma-separated list to specify two or three broker addresses in case a server is down. 3 The OpenShift Secret that contains the public key ( ca.crt ) value of the Certificate Authority that signs new user certificates for TLS client authentication. 4 The OpenShift Secret that contains the private key ( ca.key ) value of the Certificate Authority that signs new user certificates for TLS client authentication. 5 The label selector used to identify the KafkaUser resources managed by the User Operator. 6 The interval between periodic reconciliations, in milliseconds. The default is 120000 (2 minutes). 7 The level for printing logging messages. You can set the level to ERROR , WARNING , INFO , DEBUG , or TRACE . 8 Enables garbage collection (GC) logging. The default is true . 9 The validity period for the Certificate Authority. The default is 365 days. 10 The renewal period for the Certificate Authority. The renewal period is measured backwards from the expiry date of the current certificate. The default is 30 days to initiate certificate renewal before the old certificates expire. 11 (Optional) The Java options used by the JVM running the User Operator 12 (Optional) The debugging ( -D ) options set for the User Operator 13 (Optional) Prefix for the names of OpenShift secrets created by the User Operator. 14 (Optional) Indicates whether the Kafka cluster supports management of authorization ACL rules using the Kafka Admin API. When set to false , the User Operator will reject all resources with simple authorization ACL rules. This helps to avoid unnecessary exceptions in the Kafka cluster logs. The default is true . If you are using TLS to connect to the Kafka cluster, specify the secrets used to authenticate connection. Otherwise, go to the step. Example TLS configuration # .... env: - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 1 value: my-cluster-cluster-ca-cert - name: STRIMZI_EO_KEY_SECRET_NAME 2 value: my-cluster-entity-operator-certs # ..." 1 The OpenShift Secret that contains the public key ( ca.crt ) value of the Certificate Authority that signs Kafka broker certificates for TLS client authentication. 2 The OpenShift Secret that contains the keystore ( entity-operator.p12 ) with the private key and certificate for TLS authentication against the Kafka cluster. The Secret must also contain the password ( entity-operator.password ) for accessing the keystore. Deploy the User Operator. oc create -f install/user-operator Check the status of the deployment: oc get deployments Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE strimzi-user-operator 1/1 1 1 READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . 6.2. Deploy Kafka Connect Kafka Connect is a tool for streaming data between Apache Kafka and external systems. In AMQ Streams, Kafka Connect is deployed in distributed mode. Kafka Connect can also work in standalone mode, but this is not supported by AMQ Streams. Using the concept of connectors , Kafka Connect provides a framework for moving large amounts of data into and out of your Kafka cluster while maintaining scalability and reliability. Kafka Connect is typically used to integrate Kafka with external databases and storage and messaging systems. The procedures in this section show how to: Deploy a Kafka Connect cluster using a KafkaConnect resource Run multiple Kafka Connect instances Create a Kafka Connect image containing the connectors you need to make your connection Create and manage connectors using a KafkaConnector resource or the Kafka Connect REST API Deploy a KafkaConnector resource to Kafka Connect Note The term connector is used interchangeably to mean a connector instance running within a Kafka Connect cluster, or a connector class. In this guide, the term connector is used when the meaning is clear from the context. 6.2.1. Deploying Kafka Connect to your OpenShift cluster This procedure shows how to deploy a Kafka Connect cluster to your OpenShift cluster using the Cluster Operator. A Kafka Connect cluster is implemented as a Deployment with a configurable number of nodes (also called workers ) that distribute the workload of connectors as tasks so that the message flow is highly scalable and reliable. The deployment uses a YAML file to provide the specification to create a KafkaConnect resource. AMQ Streams provides example configuration files . In this procedure, we use the following example file: examples/connect/kafka-connect.yaml Prerequisites The Cluster Operator must be deployed. Running Kafka cluster. Procedure Deploy Kafka Connect to your OpenShift cluster. Use the examples/connect/kafka-connect.yaml file to deploy Kafka Connect. oc apply -f examples/connect/kafka-connect.yaml Check the status of the deployment: oc get deployments -n <my_cluster_operator_namespace> Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE my-connect-cluster-connect 1/1 1 1 my-connect-cluster is the name of the Kafka Connect cluster. READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . Additional resources Kafka Connect cluster configuration 6.2.2. Kafka Connect configuration for multiple instances If you are running multiple instances of Kafka Connect, you have to change the default configuration of the following config properties: apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect spec: # ... config: group.id: connect-cluster 1 offset.storage.topic: connect-cluster-offsets 2 config.storage.topic: connect-cluster-configs 3 status.storage.topic: connect-cluster-status 4 # ... # ... 1 The Kafka Connect cluster ID within Kafka. 2 Kafka topic that stores connector offsets. 3 Kafka topic that stores connector and task status configurations. 4 Kafka topic that stores connector and task status updates. Note Values for the three topics must be the same for all Kafka Connect instances with the same group.id . Unless you change the default settings, each Kafka Connect instance connecting to the same Kafka cluster is deployed with the same values. What happens, in effect, is all instances are coupled to run in a cluster and use the same topics. If multiple Kafka Connect clusters try to use the same topics, Kafka Connect will not work as expected and generate errors. If you wish to run multiple Kafka Connect instances, change the values of these properties for each instance. 6.2.3. Extending Kafka Connect with connector plugins The AMQ Streams container images for Kafka Connect include two built-in file connectors for moving file-based data into and out of your Kafka cluster. Table 6.1. File connectors File Connector Description FileStreamSourceConnector Transfers data to your Kafka cluster from a file (the source). FileStreamSinkConnector Transfers data from your Kafka cluster to a file (the sink). The procedures in this section show how to add your own connector classes to connector images by: Creating a new container image automatically using AMQ Streams Creating a container image from the Kafka Connect base image (manually or using continuous integration) Important You create the configuration for connectors directly using the Kafka Connect REST API or KafkaConnector custom resources . 6.2.3.1. Creating a new container image automatically using AMQ Streams This procedure shows how to configure Kafka Connect so that AMQ Streams automatically builds a new container image with additional connectors. You define the connector plugins using the .spec.build.plugins property of the KafkaConnect custom resource. AMQ Streams will automatically download and add the connector plugins into a new container image. The container is pushed into the container repository specified in .spec.build.output and automatically used in the Kafka Connect deployment. Prerequisites The Cluster Operator must be deployed. A container registry. You need to provide your own container registry where images can be pushed to, stored, and pulled from. AMQ Streams supports private container registries as well as public registries such as Quay or Docker Hub . Procedure Configure the KafkaConnect custom resource by specifying the container registry in .spec.build.output , and additional connectors in .spec.build.plugins : apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec: 1 #... build: output: 2 type: docker image: my-registry.io/my-org/my-connect-cluster:latest pushSecret: my-registry-credentials plugins: 3 - name: debezium-postgres-connector artifacts: - type: tgz url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03 - name: camel-telegram artifacts: - type: tgz url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479 #... 1 The specification for the Kafka Connect cluster . 2 (Required) Configuration of the container registry where new images are pushed. 3 (Required) List of connector plugins and their artifacts to add to the new container image. Each plugin must be configured with at least one artifact . Create or update the resource: Wait for the new container image to build, and for the Kafka Connect cluster to be deployed. Use the Kafka Connect REST API or the KafkaConnector custom resources to use the connector plugins you added. Additional resources See the Using Strimzi guide for more information on: Kafka Connect Build schema reference 6.2.3.2. Creating a Docker image from the Kafka Connect base image This procedure shows how to create a custom image and add it to the /opt/kafka/plugins directory. You can use the Kafka container image on Red Hat Ecosystem Catalog as a base image for creating your own custom image with additional connector plugins. At startup, the AMQ Streams version of Kafka Connect loads any third-party connector plugins contained in the /opt/kafka/plugins directory. Prerequisites The Cluster Operator must be deployed. Procedure Create a new Dockerfile using registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 as the base image: Example plug-in file Note This example uses the Debezium connectors for MongoDB, MySQL, and PostgreSQL. Debezium running in Kafka Connect looks the same as any other Kafka Connect task. Build the container image. Push your custom image to your container registry. Point to the new container image. You can either: Edit the KafkaConnect.spec.image property of the KafkaConnect custom resource. If set, this property overrides the STRIMZI_KAFKA_CONNECT_IMAGES variable in the Cluster Operator. apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec: 1 #... image: my-new-container-image 2 config: 3 #... 1 The specification for the Kafka Connect cluster . 2 The docker image for the pods. 3 Configuration of the Kafka Connect workers (not connectors). or In the install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml file, edit the STRIMZI_KAFKA_CONNECT_IMAGES variable to point to the new container image, and then reinstall the Cluster Operator. Additional resources Container image configuration and the KafkaConnect.spec.image property Cluster Operator configuration and the STRIMZI_KAFKA_CONNECT_IMAGES variable 6.2.4. Creating and managing connectors When you have created a container image for your connector plug-in, you need to create a connector instance in your Kafka Connect cluster. You can then configure, monitor, and manage a running connector instance. A connector is an instance of a particular connector class that knows how to communicate with the relevant external system in terms of messages. Connectors are available for many external systems, or you can create your own. You can create source and sink types of connector. Source connector A source connector is a runtime entity that fetches data from an external system and feeds it to Kafka as messages. Sink connector A sink connector is a runtime entity that fetches messages from Kafka topics and feeds them to an external system. 6.2.4.1. APIs for creating and managing connectors AMQ Streams provides two APIs for creating and managing connectors: KafkaConnector custom resources (referred to as KafkaConnectors) The Kafka Connect REST API Using the APIs, you can: Check the status of a connector instance Reconfigure a running connector Increase or decrease the number of connector tasks for a connector instance Restart connectors Restart connector tasks, including failed tasks Pause a connector instance Resume a previously paused connector instance Delete a connector instance KafkaConnector custom resources KafkaConnectors allow you to create and manage connector instances for Kafka Connect in an OpenShift-native way, so an HTTP client such as cURL is not required. Like other Kafka resources, you declare a connector's desired state in a KafkaConnector YAML file that is deployed to your OpenShift cluster to create the connector instance. KafkaConnector resources must be deployed to the same namespace as the Kafka Connect cluster they link to. You manage a running connector instance by updating its corresponding KafkaConnector resource, and then applying the updates. You remove a connector by deleting its corresponding KafkaConnector . To ensure compatibility with earlier versions of AMQ Streams, KafkaConnectors are disabled by default. To enable KafkaConnectors for a Kafka Connect cluster, you set the strimzi.io/use-connector-resources annotation to true in the KafkaConnect resource. For instructions, see Configuring Kafka Connect . When KafkaConnectors are enabled, the Cluster Operator begins to watch for them. It updates the configurations of running connector instances to match the configurations defined in their KafkaConnectors. AMQ Streams provides an example KafkaConnector configuration file, which you can use to create and manage a FileStreamSourceConnector and a FileStreamSinkConnector . Note You can restart a connector or restart a connector task by annotating a KafkaConnector resource. Kafka Connect API The operations supported by the Kafka Connect REST API are described in the Apache Kafka Connect API documentation . Switching from using the Kafka Connect API to using KafkaConnectors You can switch from using the Kafka Connect API to using KafkaConnectors to manage your connectors. To make the switch, do the following in the order shown: Deploy KafkaConnector resources with the configuration to create your connector instances. Enable KafkaConnectors in your Kafka Connect configuration by setting the strimzi.io/use-connector-resources annotation to true . Warning If you enable KafkaConnectors before creating the resources, you will delete all your connectors. To switch from using KafkaConnectors to using the Kafka Connect API, first remove the annotation that enables the KafkaConnectors from your Kafka Connect configuration. Otherwise, manual changes made directly using the Kafka Connect REST API are reverted by the Cluster Operator. 6.2.4.2. Deploying example KafkaConnector resources Use KafkaConnectors with Kafka Connect to stream data to and from a Kafka cluster. AMQ Streams provides example configuration files . In this procedure, we use the following example file: examples/connect/source-connector.yaml . The file is used to create the following connector instances: A FileStreamSourceConnector instance that reads each line from the Kafka license file (the source) and writes the data as messages to a single Kafka topic. A FileStreamSinkConnector instance that reads messages from the Kafka topic and writes the messages to a temporary file (the sink). Note In a production environment, you prepare container images containing your desired Kafka Connect connectors, as described in Section 6.2.3, "Extending Kafka Connect with connector plugins" . The FileStreamSourceConnector and FileStreamSinkConnector are provided as examples. Running these connectors in containers as described here is unlikely to be suitable for production use cases. Prerequisites A Kafka Connect deployment KafkaConnectors are enabled in the Kafka Connect deployment The Cluster Operator is running Procedure Edit the examples/connect/source-connector.yaml file: apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-source-connector 1 labels: strimzi.io/cluster: my-connect-cluster 2 spec: class: org.apache.kafka.connect.file.FileStreamSourceConnector 3 tasksMax: 2 4 config: 5 file: "/opt/kafka/LICENSE" 6 topic: my-topic 7 # ... 1 Name of the KafkaConnector resource, which is used as the name of the connector. Use any name that is valid for an OpenShift resource. 2 Name of the Kafka Connect cluster to create the connector instance in. Connectors must be deployed to the same namespace as the Kafka Connect cluster they link to. 3 Full name or alias of the connector class. This should be present in the image being used by the Kafka Connect cluster. 4 Maximum number of Kafka Connect Tasks that the connector can create. 5 Connector configuration as key-value pairs. 6 This example source connector configuration reads data from the /opt/kafka/LICENSE file. 7 Kafka topic to publish the source data to. Create the source KafkaConnector in your OpenShift cluster: oc apply -f examples/connect/source-connector.yaml Create an examples/connect/sink-connector.yaml file: touch examples/connect/sink-connector.yaml Paste the following YAML into the sink-connector.yaml file: apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-sink-connector labels: strimzi.io/cluster: my-connect spec: class: org.apache.kafka.connect.file.FileStreamSinkConnector 1 tasksMax: 2 config: 2 file: "/tmp/my-file" 3 topics: my-topic 4 1 Full name or alias of the connector class. This should be present in the image being used by the Kafka Connect cluster. 2 Connector configuration as key-value pairs. 3 Temporary file to publish the source data to. 4 Kafka topic to read the source data from. Create the sink KafkaConnector in your OpenShift cluster: oc apply -f examples/connect/sink-connector.yaml Check that the connector resources were created: oc get kctr --selector strimzi.io/cluster= MY-CONNECT-CLUSTER -o name my-source-connector my-sink-connector Replace MY-CONNECT-CLUSTER with your Kafka Connect cluster. In the container, execute kafka-console-consumer.sh to read the messages that were written to the topic by the source connector: oc exec MY-CLUSTER -kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server MY-CLUSTER -kafka-bootstrap. NAMESPACE .svc:9092 --topic my-topic --from-beginning Source and sink connector configuration options The connector configuration is defined in the spec.config property of the KafkaConnector resource. The FileStreamSourceConnector and FileStreamSinkConnector classes support the same configuration options as the Kafka Connect REST API. Other connectors support different configuration options. Table 6.2. Configuration options for the FileStreamSource connector class Name Type Default value Description file String Null Source file to write messages to. If not specified, the standard input is used. topic List Null The Kafka topic to publish data to. Table 6.3. Configuration options for FileStreamSinkConnector class Name Type Default value Description file String Null Destination file to write messages to. If not specified, the standard output is used. topics List Null One or more Kafka topics to read data from. topics.regex String Null A regular expression matching one or more Kafka topics to read data from. 6.2.4.3. Performing a restart of a Kafka connector This procedure describes how to manually trigger a restart of a Kafka connector by using an OpenShift annotation. Prerequisites The Cluster Operator is running. Procedure Find the name of the KafkaConnector custom resource that controls the Kafka connector you want to restart: oc get KafkaConnector To restart the connector, annotate the KafkaConnector resource in OpenShift. For example, using oc annotate : oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart=true Wait for the reconciliation to occur (every two minutes by default). The Kafka connector is restarted, as long as the annotation was detected by the reconciliation process. When Kafka Connect accepts the restart request, the annotation is removed from the KafkaConnector custom resource. 6.2.4.4. Performing a restart of a Kafka connector task This procedure describes how to manually trigger a restart of a Kafka connector task by using an OpenShift annotation. Prerequisites The Cluster Operator is running. Procedure Find the name of the KafkaConnector custom resource that controls the Kafka connector task you want to restart: oc get KafkaConnector Find the ID of the task to be restarted from the KafkaConnector custom resource. Task IDs are non-negative integers, starting from 0. oc describe KafkaConnector KAFKACONNECTOR-NAME To restart the connector task, annotate the KafkaConnector resource in OpenShift. For example, using oc annotate to restart task 0: oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart-task=0 Wait for the reconciliation to occur (every two minutes by default). The Kafka connector task is restarted, as long as the annotation was detected by the reconciliation process. When Kafka Connect accepts the restart request, the annotation is removed from the KafkaConnector custom resource. 6.2.4.5. Exposing the Kafka Connect API Use the Kafka Connect REST API as an alternative to using KafkaConnector resources to manage connectors. The Kafka Connect REST API is available as a service running on <connect_cluster_name> -connect-api:8083 , where <connect_cluster_name> is the name of your Kafka Connect cluster. The service is created when you create a Kafka Connect instance. Note The strimzi.io/use-connector-resources annotation enables KafkaConnectors. If you applied the annotation to your KafkaConnect resource configuration, you need to remove it to use the Kafka Connect API. Otherwise, manual changes made directly using the Kafka Connect REST API are reverted by the Cluster Operator. You can add the connector configuration as a JSON object. Example curl request to add connector configuration curl -X POST \ http://my-connect-cluster-connect-api:8083/connectors \ -H 'Content-Type: application/json' \ -d '{ "name": "my-source-connector", "config": { "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector", "file": "/opt/kafka/LICENSE", "topic":"my-topic", "tasksMax": "4", "type": "source" } }' The API is only accessible within the OpenShift cluster. If you want to make the Kafka Connect API accessible to applications running outside of the OpenShift cluster, you can expose it manually by creating one of the following features: LoadBalancer or NodePort type services Ingress resources OpenShift routes Note The connection is insecure, so allow external access advisedly. If you decide to create services, use the labels from the selector of the <connect_cluster_name> -connect-api service to configure the pods to which the service will route the traffic: Selector configuration for the service # ... selector: strimzi.io/cluster: my-connect-cluster 1 strimzi.io/kind: KafkaConnect strimzi.io/name: my-connect-cluster-connect 2 #... 1 Name of the Kafka Connect custom resource in your OpenShift cluster. 2 Name of the Kafka Connect deployment created by the Cluster Operator. You must also create a NetworkPolicy that allows HTTP requests from external clients. Example NetworkPolicy to allow requests to the Kafka Connect API apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: my-custom-connect-network-policy spec: ingress: - from: - podSelector: 1 matchLabels: app: my-connector-manager ports: - port: 8083 protocol: TCP podSelector: matchLabels: strimzi.io/cluster: my-connect-cluster strimzi.io/kind: KafkaConnect strimzi.io/name: my-connect-cluster-connect policyTypes: - Ingress 1 The label of the pod that is allowed to connect to the API. To add the connector configuration outside the cluster, use the URL of the resource that exposes the API in the curl command. 6.3. Deploy Kafka MirrorMaker The Cluster Operator deploys one or more Kafka MirrorMaker replicas to replicate data between Kafka clusters. This process is called mirroring to avoid confusion with the Kafka partitions replication concept. MirrorMaker consumes messages from the source cluster and republishes those messages to the target cluster. 6.3.1. Deploying Kafka MirrorMaker to your OpenShift cluster This procedure shows how to deploy a Kafka MirrorMaker cluster to your OpenShift cluster using the Cluster Operator. The deployment uses a YAML file to provide the specification to create a KafkaMirrorMaker or KafkaMirrorMaker2 resource depending on the version of MirrorMaker deployed. Important Kafka MirrorMaker 1 (referred to as just MirrorMaker in the documentation) has been deprecated in Apache Kafka 3.0.0 and will be removed in Apache Kafka 4.0.0. As a result, the KafkaMirrorMaker custom resource which is used to deploy Kafka MirrorMaker 1 has been deprecated in AMQ Streams as well. The KafkaMirrorMaker resource will be removed from AMQ Streams when we adopt Apache Kafka 4.0.0. As a replacement, use the KafkaMirrorMaker2 custom resource with the IdentityReplicationPolicy . AMQ Streams provides example configuration files . In this procedure, we use the following example files: examples/mirror-maker/kafka-mirror-maker.yaml examples/mirror-maker/kafka-mirror-maker-2.yaml Prerequisites The Cluster Operator must be deployed. Procedure Deploy Kafka MirrorMaker to your OpenShift cluster: For MirrorMaker: oc apply -f examples/mirror-maker/kafka-mirror-maker.yaml For MirrorMaker 2.0: oc apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml Check the status of the deployment: oc get deployments -n <my_cluster_operator_namespace> Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE my-mirror-maker-mirror-maker 1/1 1 1 my-mm2-cluster-mirrormaker2 1/1 1 1 my-mirror-maker is the name of the Kafka MirrorMaker cluster. my-mm2-cluster is the name of the Kafka MirrorMaker 2.0 cluster. READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . Additional resources Kafka MirrorMaker cluster configuration 6.4. Deploy Kafka Bridge The Cluster Operator deploys one or more Kafka bridge replicas to send data between Kafka clusters and clients via HTTP API. 6.4.1. Deploying Kafka Bridge to your OpenShift cluster This procedure shows how to deploy a Kafka Bridge cluster to your OpenShift cluster using the Cluster Operator. The deployment uses a YAML file to provide the specification to create a KafkaBridge resource. AMQ Streams provides example configuration files . In this procedure, we use the following example file: examples/bridge/kafka-bridge.yaml Prerequisites The Cluster Operator must be deployed. Procedure Deploy Kafka Bridge to your OpenShift cluster: oc apply -f examples/bridge/kafka-bridge.yaml Check the status of the deployment: oc get deployments -n <my_cluster_operator_namespace> Output shows the deployment name and readiness NAME READY UP-TO-DATE AVAILABLE my-bridge-bridge 1/1 1 1 my-bridge is the name of the Kafka Bridge cluster. READY shows the number of replicas that are ready/expected. The deployment is successful when the AVAILABLE output shows 1 . Additional resources Kafka Bridge cluster configuration Using the AMQ Streams Kafka Bridge 6.4.2. Exposing the Kafka Bridge service to your local machine Use port forwarding to expose the AMQ Streams Kafka Bridge service to your local machine on http://localhost:8080 . Note Port forwarding is only suitable for development and testing purposes. Procedure List the names of the pods in your OpenShift cluster: oc get pods -o name pod/kafka-consumer # ... pod/quickstart-bridge-589d78784d-9jcnr pod/strimzi-cluster-operator-76bcf9bc76-8dnfm Connect to the Kafka Bridge pod on port 8080 : oc port-forward pod/quickstart-bridge-589d78784d-9jcnr 8080:8080 & Note If port 8080 on your local machine is already in use, use an alternative HTTP port, such as 8008 . API requests are now forwarded from port 8080 on your local machine to port 8080 in the Kafka Bridge pod. 6.4.3. Accessing the Kafka Bridge outside of OpenShift After deployment, the AMQ Streams Kafka Bridge can only be accessed by applications running in the same OpenShift cluster. These applications use the <kafka_bridge_name> -bridge-service service to access the API. If you want to make the Kafka Bridge accessible to applications running outside of the OpenShift cluster, you can expose it manually by creating one of the following features: LoadBalancer or NodePort type services Ingress resources OpenShift routes If you decide to create Services, use the labels from the selector of the <kafka_bridge_name> -bridge-service service to configure the pods to which the service will route the traffic: # ... selector: strimzi.io/cluster: kafka-bridge-name 1 strimzi.io/kind: KafkaBridge #... 1 Name of the Kafka Bridge custom resource in your OpenShift cluster.

Commands:
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace> /' install/cluster-operator/*RoleBinding*.yaml
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace> /' install/cluster-operator/*RoleBinding*.yaml
create -f install/cluster-operator -n <my_cluster_operator_namespace>
get deployments -n <my_cluster_operator_namespace>
NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace> /' install/cluster-operator/*RoleBinding*.yaml
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace> /' install/cluster-operator/*RoleBinding*.yaml
apiVersion: apps/v1 kind: Deployment spec: # template: spec: serviceAccountName: strimzi-cluster-operator containers: - name: strimzi-cluster-operator image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.1.0 imagePullPolicy: IfNotPresent env: - name: STRIMZI_NAMESPACE value: watched-namespace-1,watched-namespace-2,watched-namespace-3
create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace> create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>
create -f install/cluster-operator -n <my_cluster_operator_namespace>
get deployments -n <my_cluster_operator_namespace>
NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1
sed -i 's/namespace: .*/namespace: <my_cluster_operator_namespace> /' install/cluster-operator/*RoleBinding*.yaml
sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace> /' install/cluster-operator/*RoleBinding*.yaml
apiVersion: apps/v1 kind: Deployment spec: # template: spec: # serviceAccountName: strimzi-cluster-operator containers: - name: strimzi-cluster-operator image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.1.0 imagePullPolicy: IfNotPresent env: - name: STRIMZI_NAMESPACE value: "*" #
create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount <my_cluster_operator_namespace> :strimzi-cluster-operator create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount <my_cluster_operator_namespace> :strimzi-cluster-operator
create -f install/cluster-operator -n <my_cluster_operator_namespace>
get deployments -n <my_cluster_operator_namespace>
NAME READY UP-TO-DATE AVAILABLE amq-streams-cluster-operator-<version> 1/1 1 1
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: kafka: version: 3.1.0 # config: # log.message.format.version: "3.1" inter.broker.protocol.version: "3.1" #
apply -f examples/kafka/kafka-ephemeral.yaml
apply -f examples/kafka/kafka-persistent.yaml
get pods -n <my_cluster_operator_namespace>
NAME READY STATUS RESTARTS my-cluster-entity-operator 3/3 Running 0 my-cluster-kafka-0 1/1 Running 0 my-cluster-kafka-1 1/1 Running 0 my-cluster-kafka-2 1/1 Running 0 my-cluster-zookeeper-0 1/1 Running 0 my-cluster-zookeeper-1 1/1 Running 0 my-cluster-zookeeper-2 1/1 Running 0
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: # entityOperator: topicOperator: {} userOperator: {}
apply -f <your-file>
apiVersion: kafka.strimzi.io/v1beta2 kind: Kafka metadata: name: my-cluster spec: # entityOperator: topicOperator: {} userOperator: {}
apply -f <your-file>
apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-topic-operator labels: app: strimzi spec: # template: # spec: # containers: - name: strimzi-topic-operator # env: - name: STRIMZI_NAMESPACE 1 valueFrom: fieldRef: fieldPath: metadata.namespace - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2 value: my-kafka-bootstrap-address:9092 - name: STRIMZI_RESOURCE_LABELS 3 value: "strimzi.io/cluster=my-cluster" - name: STRIMZI_ZOOKEEPER_CONNECT 4 value: my-cluster-zookeeper-client:2181 - name: STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 5 value: "18000" - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6 value: "120000" - name: STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS 7 value: "6" - name: STRIMZI_LOG_LEVEL 8 value: INFO - name: STRIMZI_TLS_ENABLED 9 value: "false" - name: STRIMZI_JAVA_OPTS 10 value: "-Xmx=512M -Xms=256M" - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 11 value: "-Djavax.net.debug=verbose -DpropertyName=value" - name: STRIMZI_PUBLIC_CA 12 value: "false" - name: STRIMZI_TLS_AUTH_ENABLED 13 value: "false" - name: STRIMZI_SASL_ENABLED 14 value: "false" - name: STRIMZI_SASL_USERNAME 15 value: "admin" - name: STRIMZI_SASL_PASSWORD 16 value: "password" - name: STRIMZI_SASL_MECHANISM 17 value: "scram-sha-512" - name: STRIMZI_SECURITY_PROTOCOL 18 value: "SSL"
. env: - name: STRIMZI_TRUSTSTORE_LOCATION 1 value: "/path/to/truststore.p12" - name: STRIMZI_TRUSTSTORE_PASSWORD 2 value: " TRUSTSTORE-PASSWORD " - name: STRIMZI_KEYSTORE_LOCATION 3 value: "/path/to/keystore.p12" - name: STRIMZI_KEYSTORE_PASSWORD 4 value: " KEYSTORE-PASSWORD "
create -f install/topic-operator
get deployments
NAME READY UP-TO-DATE AVAILABLE strimzi-topic-operator 1/1 1 1
apiVersion: apps/v1 kind: Deployment metadata: name: strimzi-user-operator labels: app: strimzi spec: # template: # spec: # containers: - name: strimzi-user-operator # env: - name: STRIMZI_NAMESPACE 1 valueFrom: fieldRef: fieldPath: metadata.namespace - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2 value: my-kafka-bootstrap-address:9092 - name: STRIMZI_CA_CERT_NAME 3 value: my-cluster-clients-ca-cert - name: STRIMZI_CA_KEY_NAME 4 value: my-cluster-clients-ca - name: STRIMZI_LABELS 5 value: "strimzi.io/cluster=my-cluster" - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6 value: "120000" - name: STRIMZI_LOG_LEVEL 7 value: INFO - name: STRIMZI_GC_LOG_ENABLED 8 value: "true" - name: STRIMZI_CA_VALIDITY 9 value: "365" - name: STRIMZI_CA_RENEWAL 10 value: "30" - name: STRIMZI_JAVA_OPTS 11 value: "-Xmx=512M -Xms=256M" - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 12 value: "-Djavax.net.debug=verbose -DpropertyName=value" - name: STRIMZI_SECRET_PREFIX 13 value: "kafka-" - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 14 value: "true"
. env: - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 1 value: my-cluster-cluster-ca-cert - name: STRIMZI_EO_KEY_SECRET_NAME 2 value: my-cluster-entity-operator-certs ..."
create -f install/user-operator
get deployments
NAME READY UP-TO-DATE AVAILABLE strimzi-user-operator 1/1 1 1
apply -f examples/connect/kafka-connect.yaml
get deployments -n <my_cluster_operator_namespace>
NAME READY UP-TO-DATE AVAILABLE my-connect-cluster-connect 1/1 1 1
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect spec: # config: group.id: connect-cluster 1 offset.storage.topic: connect-cluster-offsets 2 config.storage.topic: connect-cluster-configs 3 status.storage.topic: connect-cluster-status 4 #
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec: 1 # build: output: 2 type: docker image: my-registry.io/my-org/my-connect-cluster:latest pushSecret: my-registry-credentials plugins: 3 - name: debezium-postgres-connector artifacts: - type: tgz url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03 - name: camel-telegram artifacts: - type: tgz url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479 #
oc apply -f KAFKA-CONNECT-CONFIG-FILE
FROM registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 USER root:root COPY ./ my-plugins / /opt/kafka/plugins/ USER 1001
tree ./ my-plugins / ./ my-plugins / ├── debezium-connector-mongodb │ ├── bson-3.4.2.jar │ ├── CHANGELOG.md │ ├── CONTRIBUTE.md │ ├── COPYRIGHT.txt │ ├── debezium-connector-mongodb-0.7.1.jar │ ├── debezium-core-0.7.1.jar │ ├── LICENSE.txt │ ├── mongodb-driver-3.4.2.jar │ ├── mongodb-driver-core-3.4.2.jar │ └── README.md ├── debezium-connector-mysql │ ├── CHANGELOG.md │ ├── CONTRIBUTE.md │ ├── COPYRIGHT.txt │ ├── debezium-connector-mysql-0.7.1.jar │ ├── debezium-core-0.7.1.jar │ ├── LICENSE.txt │ ├── mysql-binlog-connector-java-0.13.0.jar │ ├── mysql-connector-java-5.1.40.jar │ ├── README.md │ └── wkb-1.0.2.jar └── debezium-connector-postgres ├── CHANGELOG.md ├── CONTRIBUTE.md ├── COPYRIGHT.txt ├── debezium-connector-postgres-0.7.1.jar ├── debezium-core-0.7.1.jar ├── LICENSE.txt ├── postgresql-42.0.0.jar ├── protobuf-java-2.6.1.jar └── README.md
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnect metadata: name: my-connect-cluster spec: 1 # image: my-new-container-image 2 config: 3 #
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-source-connector 1 labels: strimzi.io/cluster: my-connect-cluster 2 spec: class: org.apache.kafka.connect.file.FileStreamSourceConnector 3 tasksMax: 2 4 config: 5 file: "/opt/kafka/LICENSE" 6 topic: my-topic 7 #
apply -f examples/connect/source-connector.yaml
touch examples/connect/sink-connector.yaml
apiVersion: kafka.strimzi.io/v1beta2 kind: KafkaConnector metadata: name: my-sink-connector labels: strimzi.io/cluster: my-connect spec: class: org.apache.kafka.connect.file.FileStreamSinkConnector 1 tasksMax: 2 config: 2 file: "/tmp/my-file" 3 topics: my-topic 4
apply -f examples/connect/sink-connector.yaml
get kctr --selector strimzi.io/cluster= MY-CONNECT-CLUSTER -o name my-source-connector my-sink-connector
exec MY-CLUSTER -kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server MY-CLUSTER -kafka-bootstrap. NAMESPACE .svc:9092 --topic my-topic --from-beginning
get KafkaConnector
annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart=true
get KafkaConnector
describe KafkaConnector KAFKACONNECTOR-NAME
annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart-task=0
curl -X POST http://my-connect-cluster-connect-api:8083/connectors -H 'Content-Type: application/json' -d '{ "name": "my-source-connector", "config": { "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector", "file": "/opt/kafka/LICENSE", "topic":"my-topic", "tasksMax": "4", "type": "source" } }'
selector: strimzi.io/cluster: my-connect-cluster 1 strimzi.io/kind: KafkaConnect strimzi.io/name: my-connect-cluster-connect 2 #
apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: my-custom-connect-network-policy spec: ingress: - from: - podSelector: 1 matchLabels: app: my-connector-manager ports: - port: 8083 protocol: TCP podSelector: matchLabels: strimzi.io/cluster: my-connect-cluster strimzi.io/kind: KafkaConnect strimzi.io/name: my-connect-cluster-connect policyTypes: - Ingress
apply -f examples/mirror-maker/kafka-mirror-maker.yaml
apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml
get deployments -n <my_cluster_operator_namespace>
NAME READY UP-TO-DATE AVAILABLE my-mirror-maker-mirror-maker 1/1 1 1 my-mm2-cluster-mirrormaker2 1/1 1 1
apply -f examples/bridge/kafka-bridge.yaml
get deployments -n <my_cluster_operator_namespace>
NAME READY UP-TO-DATE AVAILABLE my-bridge-bridge 1/1 1 1
get pods -o name pod/kafka-consumer pod/quickstart-bridge-589d78784d-9jcnr pod/strimzi-cluster-operator-76bcf9bc76-8dnfm
port-forward pod/quickstart-bridge-589d78784d-9jcnr 8080:8080 &
selector: strimzi.io/cluster: kafka-bridge-name 1 strimzi.io/kind: KafkaBridge #