Configuration

Internal Access in the Same namespace for Plain

When the message producer or consumer applications are in same namespace as the Message Bus service then they can access the Kafka cluster using the Bootstrap Kubernetes service object name and port.

Run the following command to test the standalone producer. Here the project namespace is sr and instance is quick.

$kubectl -n sr run kafka-producer-plain -ti \
--image=<STRIMZI_KAFKA_IMAGE_NAME> \
--rm=true --restart=Never \
-- bin/kafka-console-producer.sh \
--bootstrap-server sr-quick-messaging-kafka-bootstrap:9092 \
--topic ora-test-topic

Type a few lines of text and each ENTER sends a message to Kafka broker. Type CTRL-C to quit.

Run the following command to test the standalone consumer. Here the project namespace is sr and instance is quick.

$kubectl -n sr run kafka-consumer-plain -ti \
--image=<STRIMZI_KAFKA_IMAGE_NAME> \
--rm=true --restart=Never \
-- bin/kafka-console-consumer.sh \
--bootstrap-server sr-quick-messaging-kafka-bootstrap:9092 \
--group ora-uim-consumer-test --isolation-level read_committed \
--topic  ora-test-topic --from-beginning

You get responses after the validation is successful.

Internal Access in a Different namespace for Plain

When the massage producer or consumer applications are in different namespace than the Message Bus service then they can access the Kafka cluster using the bootstrap service name and port but need to suffix <namespace>.svc.cluster.local to the service name.

See "Internal access - same namespace - plain" section on running the standalone console test producer and consumer pods for testing. Replace the bootstrap-server url with sr-quick-messaging-kafka-bootstrap.sr.svc.cluster.local, where the namespace is sr and instance is quick.

Internal Access in the Same namespace for Authentication

When the message producer or consumer applications are in same namespace as the Message Bus service then they can access the Kafka cluster using the bootstrap Kubernetes service object name and port.

Create a test client pod definition.

1. Copy the following YAML content into the bastion host (or worker node) as mb-test-client-deployment.yaml file.
2. Update the hostAliases section according to your OAuth service environment.
3. Update the STRIMZI_KAFKA_IMAGE_NAME.
4. Update the OAUTH Endpoint, Client Id and Secret.
5. Update the OAUTH Endpoint, Client Id, Client Secret, Scope, Audience, and anything else that are applicable to your client configuration

apiVersion: apps/v1
kind: Deployment
metadata:
  name:  mb-test-auth-client-deployment
  labels:
    app:  mb-test-auth-client
spec:
  replicas: 1
  selector:
    matchLabels:
      app:  mb-test-auth-client
  template:
    metadata:
      labels:
        app:  mb-test-auth-client
    spec:
#     <Uncomment below and replace with your bootstrap and brokers DNS names>  
      #hostAliases:
      #- ip: <LOADBALANCER_IP>
      #  hostnames:
      #  - "<OHS_HOSTNAME>"
      containers:
      - name:  mb-test-client
        image: <STRIMZI_KAFKA_IMAGE_NAME>
        command:
        - "tail"
        - "-f"
        - "/dev/null"
        imagePullPolicy: IfNotPresent
        env:
        - name: OAUTH_TOKEN_ENDPOINT_URI
          value: <Update the OAUTH_TOKEN_ENDPOINT_URI>
        - name: OAUTH_CLIENT_ID
          value: <Update the OAUTH_CLIENT_ID>
        - name: OAUTH_CLIENT_SECRET
          value: <Update the OAUTH_CLIENT_SECRET>
       # - name: OAUTH_SCOPE
       #   value: <Uncomment and update OAUTH_SCOPE>
       # - name: OAUTH_AUDIENCE
       #   value: <Uncomment and update OAUTH_AUDIENCE>
        ports:
        - containerPort: 9090
          name: http
          protocol: TCP

Create the authentication properties in a file (mb_test_client.properties).

sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
security.protocol=SASL_PLAINTEXT
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

Run the test client container and provide authentication properties

#Apply the test client pod definition in the namespace (say "sr").
$kubectl apply -f mb-test-client-deployment.yaml -n sr
 
#Get the newly created pod name
$kubectl get pod -n sr | grep mb-test-auth-client-deployment
 
#Sample Output
#mb-test-auth-client-deployment-******-****         1/1     Running   0          98s
 
#Copy the mb_authentication.properties file into the pod
$kubectl -n sr cp mb_test_client.properties mb-test-auth-client-deployment-******-****:/home/kafka/mb_test_client.properties

Test for message bus producer client:

• Start an interactive shell process in the test client pod
• Export the environment variables needed for the authentication
• Run the console producer command.
• Enter some string messages

#Get the newly created pod name
kubectl get pod -n sr | grep mb-test-auth-client-deployment
 
#Exec into the newly created pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash
 
#Run the following test console producer
bin/kafka-console-producer.sh \
--producer.config /home/kafka/mb_test_client.properties \
--bootstrap-server sr-quick-messaging-kafka-bootstrap:9092 \
--topic ora-test-topic

Test for message bus consumer client:

• Start an interactive shell process in the test client pod
• Export the environment variables needed for the authentication
• Run the console consumer command.
• You will see the previous string messages of producer

#Get the newly created pod name
kubectl get pod -n sr | grep mb-test-auth-client-deployment
 
#Sample Output
#mb-test-auth-client-deployment-******-****         1/1     Running   0          98s
 
#Exec into the newly created pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash
 
#Run the following test console consumer
bin/kafka-console-consumer.sh \
--consumer.config /home/kafka/mb_test_client.properties \
--bootstrap-server sr-quick-messaging-kafka-bootstrap:9092 \
--topic ora-test-topic \
--from-beginning

External ingress access - SSL and Authentication

The external access to Message Bus is provided through Ingress controller (Traefik or Generic) with TLS enabled. The following must be performed in clients for testing:

• Export and import the Message Bus service (that is sr-quick-messaging-cluster-ca-cert, where sr is namespace and quick is instance) certificate into clients.
• Export and import the certificate of OAuth service into the clients.

  Note:

  This is optional and is required only if OAuth is enabled for SSL.
• Update the bootstrap and brokers DNS names with load balancer IP in the etc/hosts file of clients (that is, event producer or consumer applications).
• Update the DNS name of OAuth service with load balancer IP in /etc/hosts file of clients.

  Note:

  This is optional and is required only if the OAuth service requires DNS name to access.
• Run the producer or consumer script with SSL and Authentication details

In the following section, the external ingress access test is provided with Strimzi Kafka container. If you want to test the client code without Kubernetes cluster then you can download the Apache Kafka and perform the same.

Add Message Bus service and OAuth service certifications to trust store. See Import/export of TLS certificates section.

#Run the below command to export and import the Message Bus service certificate into the trust store (mb-cert-keystore.jks) file.
$COMMON_CNTK/scripts/export-message-bus-cert.sh -p sr -i quick -l . -k ./mb-test-client-cert-keystore.jks -a mb-cert
 
#Get the OAuth (OAM) service certificate and import into trust store (mb-test-client-cert-keystore.jks) file (Optional, needed if OAuth is SSL)
keytool -importcert -alias oauth-server -file <Path to OAuth Server certificate, the .pem file> -keystore ./mb-test-client-cert-keystore.jk --trustcacerts -noprompt

Create the following authentication properties in a file (mb_test_client.properties).

sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
ssl.endpoint.identification.algorithm=

Create a test client pod definition.

1. Copy the following YAML content into the bastion host (or worker node) as "mb-test-client-deployment.yaml" file.
2. Update the Strimzi Kafka image.
3. Update the hostAliases section according to your OAuth and Message Bus service setup. This will add entries to /etc/hosts file.
4. Update the OAuth Endpoint, Client Id, Client Secret, and Trust Store Password in env section.

apiVersion: apps/v1
kind: Deployment
metadata:
  name:  mb-test-client-deployment
  labels:
    app: mb-test-client
spec:
  replicas: 1
  selector:
    matchLabels:
      app:  mb-test-client
  template:
    metadata:
      labels:
        app:  mb-test-client
    spec:
#     <Uncomment below and replace with your bootstrap and brokers DNS names> 
#     hostAliases:
#     - ip: <Replace with your LOADBALANCER_IP>
#       hostnames:
#       - "<INSTANCE.PROJECT.messaging.broker0.uim.org>"
#       - "<INSTANCE.PROJECT.messaging.brokerN.uim.org>"
#       - "<INSTANCE.PROJECT.messaging.bootstrap.uim.org>"
#       - "<Replace with OHS_HOSTNAME>"
      containers:
      - name:  mb-test-client
        image: quay.io/strimzi/kafka:0.34.0-kafka-3.4.0
        command:
        - "tail"
        - "-f"
        - "/dev/null"
        imagePullPolicy: IfNotPresent
        env:
        - name: OAUTH_TOKEN_ENDPOINT_URI
          value: <Replace with your OAUTH_TOKEN_ENDPOINT_URI>
        - name: OAUTH_CLIENT_ID
          value: <Replace with your OAUTH_CLIENT_ID>
        - name: OAUTH_CLIENT_SECRET
          value: <Replace with your OAUTH_CLIENT_SECRET>
        #- name: OAUTH_SCOPE
        #  value: <Uncomment and replace with your OAUTH_SCOPE>
        #- name: OAUTH_AUDIENCE
        #  value: <Uncomment and replace with yours OAUTH_AUDIENCE>
        - name: KAFKA_OPTS
          value: " \
                  -Djavax.net.ssl.trustStore=/home/kafka/mb-test-client-cert-keystore.jks \
                  -Djavax.net.ssl.trustStorePassword=<Replace with your store password> \
                  -Djavax.net.ssl.trustStoreType=JKS"
        ports:
        - containerPort: 9090
          name: http
          protocol: TCP

Run the test client container and apply readiness for authentication and SSL.

#Apply the test client pod definition in the namespace (say "sr").
$kubectl apply -f mb-test-client-deployment.yaml -n sr
 
#Get the newly created pod name
$kubectl get pod -n sr | grep mb-test-client-deployment
 
#Sample Output
#mb-test-client-deployment-******-****         1/1     Running   0          98s
 
#Copy the certificate store into the newly created pod.  Replace the pod name below
kubectl -n sr cp mb-test-client-cert-keystore.jks <Replace with mb-test-client-deployment pod name>:/home/kafka/mb-test-client-cert-keystore.jks  
 
#Copy the mb_test_client.properties file into the POD
kubectl -n sr cp mb_test_client.properties <Replace with mb-test-client-deployment pod name>:/home/kafka/mb_test_client.properties

Start a shell session inside container for console test producer.

#Get the newly created pod name
$kubectl get pod -n sr | grep mb-test-auth-client-deployment
 
#Sample Output
#mb-test-auth-client-deployment-******-****         1/1     Running   0          98s
 
#Exec into the newly created pod
kubectl exec -it<Replace with mb-test-client-deployment pod name> -n sr -- bash
 
#Run the following producer command
bin/kafka-console-producer.sh \
--producer.config /home/kafka/mb_test_client.properties \
--bootstrap-server quick.sr.messaging.bootstrap.uim.org:30443 \
--topic ora-test-topic 

Start start a shell session inside container for console test consumer:

#Exec into the newly created pod
kubectl exec -it <Replace with mb-test-client-deployment pod name> -n sr -- bash
 
#Run the following producer command.  Replace the bootstrap-server url accordingly to your environment
bin/kafka-console-consumer.sh \
--consumer.config /home/kafka/mb_test_client.properties \
--bootstrap-server sthatipa.sr.messaging.bootstrap.uim.org:30443 \
--consumer-property group.id=test-client-service \
--topic ora-test-topic --from-beginning

Clean-up the newly created test pod:

kubectl delete -f mb-test-client-deployment.yaml -n sr

External node port access

The nodeport listener type allows the external access from outside of the Kubernetes cluster using the load balancer or Kubernetes worker node ip address and nodePort(port of worker node).

The Bootstrap URL is constructed with worker node IP Address and node port of bootstrap service.

Get the host port of the external bootstrap service using the following command:

$kubectl get service sr-quick-messaging-kafka-nodeport-bootstrap -o=jsonpath='{.spec.ports[0].nodePort}{"\n"}' -n sr
 
Output: 32100

Get the IP Address of the Kubernetes worker node. Replace the <NODE_NAME> in the following with your node name:

$kubectl get node <NODE_NAME> -o=jsonpath='{range .status.addresses[*]}{.type}{"\t"}{.address}{"\n"}' -n sr
 
Output:
InternalIP  100.xx.xx.142
Hostname    *********

Update the Kafka cluster Bootstrap URL as 100.xx.xx.142:32100 in the events producer and consumer applications.

To access with plain, see "Internal access - same namespace - plain" section. Replace the bootstrap URL with above constructed one.

To access with Authentication, see "Internal access - same namespace - authentication" section. Replace the bootstrap URL with above constructed one.

To access with SSL and Authentication, see "External ingress access - SSL & Authentication" section. Replace the bootstrap URL with above constructed one.

Import/export of TLS certificates

To enable TLS encrypted access, the ca-certs of Kafka cluster is needed to be extracted and imported into key store and the location of that key store is used as the producer or consumer properties in events application.

Export the ca-certs of the Kafka cluster using the following command:

$COMMON_CNTK/scripts/export-message-bus-cert.sh -p <Namespace of kafka cluster> \
-i <instance name of kafka cluster> \
-l <directory to export clustercerts temporarily> \
-k <keystore-location> \
-a <alias for cert>

For example:

$COMMON_CNTK/scripts/export-message-bus-cert.sh -p sr -i quick -l . -k ./mb-cert-keystore.jks -a mb-sr-quick-cert

The export-cluster-cert.sh script creates JKS type truststore by default in the provided key store location. If any other truststore type is created, specify that as producer or consumer property while running the clients. These exported artifacts can be used in Kafka client applications.

Using custom certificates

Custom certificates can be used while creating the Kafka cluster:

Prerequisites:

• Certificates and keys are to be in PEM format.
• Key should not be encrypted. Encrypted keys are not supported since they need user interaction for entering the passphrase during access.

Creating a custom certificate

To create a custom certificate, see "SSL Certificates".

Create Kubernetes secret

Run the following command by replacing the placeholders:

kubectl create secret generic <secret-name> --from-file=<key-file-name> --from-file=<certificate-file-name>
 For example:
kubectl create secret generic myCustomCertSecret --from-file=commonkey.pem --from-file=commoncert.pem

Update Kafka Cluster configuration

Update the customCerts configuration section in Kafka cluster's override values yaml file:

kafka-cluster:
  ## to enable custom or owned certs for tls please create a kubernetes secret with the cert and key if not already present, uncomment the below section and add respective values.
    ## please be advised that encrypted keys are not supported since they require user interaction for the passphrase
    customCerts:
      # Secret in which cert and key are present
      secretName: <secret-name created above>
      certName: <certificate file used in the secret created above>
      keyName: <key-file used in the secret created above>

Message Bus Internal Listener

The following is the configuration for internal listener type which can be commented or uncommented.

 kafka-cluster:
  listeners:
  # plain is for internal access within the same k8s cluster.
    internal:

From same namespace in cluster

This is an internal access method that is used by the message producer or consumer clients (or applications) when they are deployed in same namespace as the Message Bus service. This is enabled by default with internal listener type. To access the Message Bus, the producer or consumer applications must get the Bootstrap service URL of the Kafka cluster.

To get the Bootstrap service URL of the Kafka cluster run the following command:

kubectl get svc -n sr | grep sr-quick-messaging-kafka-bootstrap
 
sr-quick-messaging-kafka-bootstrap       ClusterIP   <clusterIP>    <none>        9091/TCP,9092/TCP

Use the sr-quick-messaging-kafka-bootstrap:9092 URL in the producer and consumer client configuration in the applications.

From another namespace in cluster

This is an internal access method which is used by the producer or consumer client applications when they are deployed in different namespace than the message-bus service. This is enabled by default with internal listener type. To access the Message Bus, the producer or consumer client applications have to get the Bootstrap service URL of the Kafka cluster and convert the URL pattern as serviceName.namespace.svc.cluster.local.

If the sr-quick-messaging-kafka-bootstrap service is hosted in sr namespace on 9092 port and the client applications from different namespace can access the Kafka cluster with Bootstrap URL as sr-quick-messaging-kafka-bootstrap.sr.svc.cluster.local:9092

Message Bus Ingress Listener

This is an external access method which is used by message producer or consumer applications when they are deployed outside of the Kubernetes cluster. This is disabled by default and must be enabled in the applications.yaml. This external access is provided through the Traefik Ingress Controller and Generic Ingress Controller to the Kafka cluster. To enable this external access, the ingress listener type configuration must be enabled in the Kafka cluster configuration yaml file.

Ingress listener type

Un-comment the ingress lister type section in applications.yaml file to expose the Message Bus Service outside of Kubernetes cluster. Ingress controller (Traefik or Generic) should be deployed in order for this ingress listener type to work and Message Bus namespace must be registered with Traefik operator. In case of Generic Ingress, set ingress.className according to your Generic Ingress Controller.

In case of Generic Ingress controller (NGINX), annotations given under the kafka-cluster.listeners.ingress.annotations tag in applications.yaml are mandatory.

# To expose the kafka-cluster to external kafka clients via ingress controller uncomment the following and modify accordingly. # Valid values are TRAEFIK, GENERIC
ingressController: "TRAEFIK"
 
#ingress:
#  #specify className field for ingressClassName of generic ingress controller.
#  #In case of nginx the default values is nginx
#  className: "nginx"
 
#provide loadbalancer port
# if TLS is enabled in global section, then loadbalancerport will be used as external port for Generic or Traefik
loadbalancerport: <loadBalancer-port>
 
kafka-cluster
  listeners:
    ingress:
      # if TLS is Disabled in global, then ingressSslPort will be used as external port.
      ingressSslPort: <LoadBalancer_SSL_Port>
      # If using Generic Ingress controller, below given annotations are mandatory for Message-Bus external access.
      # These annotations are required for nginx ingress controller in Message-Bus.
      annotations: 
        nginx.ingress.kubernetes.io/ingress.allow-http: "false"
        nginx.ingress.kubernetes.io/backend-protocol: "HTTPS"
        ingress.kubernetes.io/ssl-passthrough: "true"
        nginx.ingress.kubernetes.io/ssl-passthrough: "true"

In external producer or consumer messaging clients (or applications), the following must be done to access the Kafka cluster through Ingress controller.

• The Bootstrap server and advertised broker host names must be configured in DNS at client side.
• Import the TLS certificate and trust stores from the Kafka cluster into client configurations.
• Add required additional properties in Kafka producer or consumer client configuration.

DNS settings in client applications host

The Bootstrap server host name and advertised broker host names must be configured in /etc/hosts file in producer and consumer client applications with the Traefik or Load Balancer IP Address. Hostnames are pre-configured when deployed with ingress listener type enabled with the following pattern:

bootstrap-server:  <kafka-cluster-instance-name>.<kafka-cluser-project-name>.messaging.bootstrap.uim.org 
broker-0:          <kafka-cluster-instance-name>.<kafka-cluser-project-name>.messaging.broker0.uim.org 
broker-1:          <kafka-cluster-instance-name>.kafka-cluser-project-name>.messaging.broker1.uim.org
 
For example if a instance is quick and namesapce is sr then the hostnames will be as follows:
bootstrap-server:  quick.sr.messaging.bootstrap.uim.org 
broker-0:          quick.sr.messaging.broker0.uim.org
broker-1:          quick.sr.messaging.broker1.uim.org

#subDomainNameSeparator: "."

#Example hostnames for "-" : quick-sr-messaging-bootstrap.uim.org

Importing certificates into client applications

See the "Import/export of TLS certificates" section in “Client Access” section for exporting the ca-certs of Kafka cluster to producer or consumer applications.

Message Bus NodePort Listener

This is another external access method which is used by events producer or consumer client applications when they are deployed out-side of the Kubernetes cluster and wants to access the message-bus service without ingress controller.

Node port

The following configuration in the application yaml file allows exposing the nodeport listener type to access the Message Bus externally with tls and OAuth 2.0 Authentication.

Kafka-cluster:
  listeners:
    #To expose the kafka-cluster to external kafka clients without ingress controller, uncomment the following section and modify accordingly.
    nodeport:
      tls: true
      # if need to expose on a static nodeport, please uncomment the below nodePort key and provide values.
      nodePort: 32100
      authentication: true

When the tls is enabled the certificates of the Kafka cluster must be imported in the events producer and consumer clients to access the Kafka cluster.

See the "Import/export of TLS certificates" section in “Client Access” section for exporting the auto-generated ca-certs of Kafka cluster.

Strimzi Operator

Validate that the Strimzi operator is installed by running the following command:

$kubectl get pod -n <STRIMZI_NAMESPACE>
 
NAME                                        READY   STATUS    RESTARTS   AGE
strimzi-cluster-operator-566948f58c-sfj7c   1/1          Running    0          6m55s

Validate installed helm release for Strimzi operator by running the following command:

$helm list -n <STRIMZI_NAMESPACE>
 
NAME                    NAMESPACE          REVISION        STATUS          CHART                         APP VERSION
strimzi-operator        STRIMZI_NAMESPACE  1               deployed        strimzi-kafka-operator-0.X.0   0.X.0

Source Message Bus

The source Message Bus should be up and running (the Kafka cluster from which the topics should be replicated).

Validate the Kafka cluster is installed by running the following command:

$kubectl get pod -n sr1
   
 NAME                                                 READY   STATUS    RESTARTS   AGE
 sr1-quick1-messaging-entity-operator-5f9c688c7-2jcjg   3/3     Running   0          27h
 sr1-quick1-messaging-kafka-0                           1/1     Running   0          27h
 sr1-quick1-messaging-zookeeper-0                       1/1     Running   0          27h

Validate the persistent volume claims created for the Kafka cluster by running the following command:

$kubectl get pvc -n sr1
 
NAME                                    STATUS   VOLUME                                    CAPACITY   ACCESS MODES   STORAGECLASS   AGE
data-sr1-quick1-messaging-kafka-0       Bound    <volume>  1Gi        RWO           sc             27h
data-sr1-quick1-messaging-zookeeper-0   Bound    <volume>  1Gi        RWO           sc    

Target Message Bus

The target Message Bus should be up and running (the Kafka cluster to which the topics should be replicated).

Validate the Kafka cluster is installed by running the following command:

$kubectl get pod -n sr2
   
 NAME                                                  READY   STATUS    RESTARTS   AGE
 sr2-quick2-messaging-entity-operator-5f9c688c7-2jcjg   3/3     Running   0          27h
 sr2-quick2-messaging-kafka-0                           1/1     Running   0          27h
 sr2-quick2-messaging-zookeeper-0                       1/1     Running   0          27h

Validate the persistent volume claims created for the Kafka cluster by running the following command:

$kubectl get pvc -n <kafka target namespace>`
 
NAME                                    STATUS   VOLUME                                    CAPACITY   ACCESS MODES   STORAGECLASS   AGE
data-sr2-quick2-messaging-kafka-0       Bound    <volume>  1Gi        RWO           sc             27h
data-sr2-quick2-messaging-zookeeper-0   Bound    <volume>  1Gi        RWO           sc             27h

Installing and configuring Mirror Maker 2.0

A sample is provided at $COMMON_CNTK/samples/messaging/kafka-mirror-maker.

NotEnoughReplicasException

When you get the org.apache.kafka.common.errors.NotEnoughReplicasException: Messages are rejected since there are fewer in-sync replicas than required. The reason could be that the topics replicas is not meeting the default minInsyncReplicas value configured in the Message Bus service.

Asynchronous auto-commit of offsets failed

When you get the following error in the logs (for example: ATA Consumer). To resolve this make sure that max.polling.interval.ms is always greater than the last poll or else reduce the max.poll.records.

[Consumer clientId=consumer-ora-uim-topology-service-2, groupId=ora-uim-topology-service] Asynchronous auto-commit of offsets failed: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.. Will continue to join group.

Add these additional properties in the YAML file under the mp.messaging.connector.helidon-kafka section with override values.

mp.messaging:
  connector:
    helidon-kafka:
      # The following are default global configuration values which effects for all the consumer groups.
      max.polling.interval.ms: 300000
      max.poll.records: 500
 
  # The following are channel specific configuration values
  incoming:
    # The toInventoryChannel effects only for ora-uim-topology-service consumer group
    # uncomment and update the specific values
    #toInventoryChannel:
      #max.polling.interval.ms: 300000
      #max.poll.records: 500
 
    # The toFaultChannel effects only for ora-uim-topology-retry-service consumer group
    # Uncomment and update the specific values
    #toRetryChannel:
      #max.polling.interval.ms: 300000
      #max.poll.records: 200
  
    # The toDltChannel effects only for ora-uim-topology-dlt-service consumer group
    # uncomment and update the specific values
    #toDltChannel:
      #max.polling.interval.ms: 300000
      #max.poll.records: 100

Performance Tuning: Consumer Configurations

The following are some consumer configuration properties in message consumers which are related to performance. See https://kafka.apache.org/documentation/#consumerconfigs for all available consumer config properties.

• max.poll.records (default=500) defines the maximum number of messages that a consumer can poll at once.
• max.partition.fetch.bytes (default=1048576) defines the maximum number of bytes that the server returns in a poll for a single partition.
• max.poll.interval.ms (default=300000) defines the time a consumer must process all messages from a poll and fetch a new poll afterward. If this interval is exceeded, the consumer leaves the consumer group.
• http://heartbeat.interval.ms (default=3000) defines the frequency with which a consumer sends heartbeats.
• http://session.timeout.ms (default=10000) defines the time a consumer must send a heartbeat. If no heartbeat was received in that timeout, the member is considered dead and leaves the group.

Managing Consumer Groups

For more list of options available on the consumer groups see the apache kafka managing consumer groups section. The following sub-sections list some significant operations. See "Message Bus Client Access" for more information.

List consumer groups

#Exec into running message bus test client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash  
 
#Run the below command to list all the consumer groups
bin/kafka-consumer-groups.sh \
--command-config /home/kafka/mb_test_client.properties \
--bootstrap-server <Your Bootstrap Server URL> \
--list

Describe consumer group

#Exec into running Kafka admin client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash 
 
#Run the below command to describe specific consumer group to check topics, partitions, offsets
#Replace the command-config, bootstrap, group values accordingly
bin/kafka-consumer-groups.sh \
--command-config /home/kafka/mb_test_client.properties \
--bootstrap-server <Your Bootstrap Server URL> \
--group test-client-service \
--describe

Reset offset of a consumer group

#Exec into running Kafka admin client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash  
 
#Run the below command to reset offset for consumer group for topic to latest.  See Apache Kafka documentation for other available options.
#Replace the command-config, bootstrap, group and topic values accordingly
bin/kafka-consumer-groups.sh \
--command-config /home/kafka/mb_test_client.properties \
--bootstrap-server <Your Bootstrap Server URL> \
--group test-client-service \
--reset-offsets \
--topic ora-test-topic \
--to-latest \
--execute

Topics

For more detailed list of operations available on the topics see the "Apache Kafka Operations".The following sub-sections list some significant operations. See "Message Bus Client Access" for more information.

Create

Create a topic with three partitions and two replications.

#Exec into running Kafka admin client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash
 
 
#Run the below command to create a topic
bin/kafka-topics.sh \
 --command-config /home/kafka/mb_test_client.properties \
 --bootstrap-server <Your Bootstrap Server URL> \
 --create \
 --topic replicated-2 \
 --replication-factor 2 \
 --partitions 3

List

To list all topics:

#Exec into running Kafka admin client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash
 
 
#Run the below command to list all the topic
bin/kafka-topics.sh \
 --command-config /home/kafka/mb_test_client.properties \
 --bootstrap-server <Your Bootstrap Server URL> \
 --list

Describe

Describes the topic and its partition count, replicas factory along with leaders for the partition.

#Exec into running Kafka admin client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash
 
#Run the below command to describe the topic
bin/kafka-topics.sh \
 --command-config /home/kafka/mb_test_client.properties \
 --bootstrap-server <Your Bootstrap Server URL> \
 --topic replicated-2 \
 --describe
 
#Sample output
Topic: replicated-2      TopicId: vyalpPOmR0CtYt7Sc-gbxA           PartitionCount: 3               ReplicationFactor: 2          Configs: min.insync.replicas=1,message.format.version=3.0-IV1
 
Topic: replicated-2     Partition: 0     Leader: 1      Replicas: 1,0     Isr: 1,0
Topic: replicated-2     Partition: 1     Leader: 0      Replicas: 0,1     Isr: 0,1
Topic: replicated-2     Partition: 2     Leader: 1      Replicas: 1,0     Isr: 1,0

Alter

You can alter a topic and increase the partitions to 2.

#Exec into running Kafka admin client pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash
 
#Run the below command to alter the topic  bin/kafka-topics.sh \
 --command-config /home/kafka/mb_test_client.properties \
 --bootstrap-server <Your Bootstrap Server URL> \
 --alter \
 --topic <Your Topic Name> \
 --partitions 1

Reassignment

The partition reassignment tool can also be used to selectively move replicas of a partition to a specific set of brokers. In the following example the partitions for topic (replicated-2) are reassigned to different brokers.

See the "Message Bus Client Access"section for more information on running the message bus test pod with required configuration such as Authentication and SSL.

Create a file called custom-reassignment.json file a terminal

{"version":"1", "partitions":[{"topic":"replicated-2","partition":"0","replicas":"[0,1]"},{"topic":"replicated-2","partition":1,"replicas":"[1,2]"},{"topic":"replicated-2","partition":"2","replicas":"[0,2]"}]}

Run the following commands for reassignment:

#Copy the custom-reassignment.json file into the newly created pod under /home/kafka directory
$kubectl cp custom-reassignment.json mb-test-auth-client-deployment-*****-****:/home/kafka/custom-reassignment.json -n kafka 
 
 
#Exec into running test pod
kubectl exec -it mb-test-auth-client-deployment-******-**** -n sr -- bash  #Cd directory to /home/kafka
 
#Validate the topic ("replicated-2"
/opt/kafka/bin/kafka-topics.sh \
--command-config /home/kafka/mb_test_client.properties \
--bootstrap-server <Your Bootstrap Server URL> \
--topic replicated-2 --describe
Topic: replicated-2     TopicId: vyalpPOmR0CtYt7Sc-gbxA PartitionCount: 3       ReplicationFactor: 2    Configs: min.insync.replicas=1,message.format.version=3.0-IV1
        Topic: replicated-2     Partition: 0    Leader: 1       Replicas: 1,0   Isr: 1,0
        Topic: replicated-2     Partition: 1    Leader: 1       Replicas: 0,1   Isr: 1,0
        Topic: replicated-2     Partition: 2    Leader: 1       Replicas: 1,0   Isr: 1,0
 
#Run reassign-partitions script to reassign the partitions according to the json file
$/opt/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server dev-messaging-kafka-bootstrap:9092 --reassignment-json-file custom-reassignment.json --execute
 
Current partition replica assignment
 
{"version":1,"partitions":[{"topic":"replicated-2","partition":0,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"replicated-2","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"replicated-2","partition":2,"replicas":[1,0],"log_dirs":["any","any"]}]}
 
Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for replicated-2-0,replicated-2-1,replicated-2-2
 
#Verfify the reassignment status
$/opt/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server dev-messaging-kafka-bootstrap:9092 --reassignment-json-file custom-reassignment.json --verify
 
Status of partition reassignment:
Reassignment of partition replicated-2-0 is complete.
Reassignment of partition replicated-2-1 is complete.
Reassignment of partition replicated-2-2 is complete.
 
Clearing broker-level throttles on brokers 0,1,2
Clearing topic-level throttles on topic replicated-2
 
# Validate the partition assignments
$/opt/kafka/bin//kafka-topics.sh \
  --command-config /home/kafka/mb_test_client.properties \
  --bootstrap-server <Your Bootstrap Server URL> \
  --topic replicated-2 --describe
Topic: replicated-2     TopicId: vyalpPOmR0CtYt7Sc-gbxA PartitionCount: 3       ReplicationFactor: 2    Configs: min.insync.replicas=1,message.format.version=3.0-IV1
        Topic: replicated-2     Partition: 0    Leader: 1       Replicas: 0,1   Isr: 1,0
        Topic: replicated-2     Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2
        Topic: replicated-2     Partition: 2    Leader: 0       Replicas: 0,2   Isr: 0,2