AMQ streams (1) of openshift 4 - multiple consumers receive data from partition

《OpenShift 4.x HOL Tutorial summary 》
explain ： This article has been OpenShift 4.10 Verification in environment

AMQ Streams What is it? ？

Red Hat AMQ Stream Red hat is based on the community version Kafka Software subscription provided . It provides everything Kafka The function of , At the same time, it can be better integrated with other red hat software . stay OpenShift We use AMQ Stream Operator To build and maintain AMQ Stream Container running environment .

install AMQ Streams Environmental Science

install AMQ Streams Operator

1. establish kafka project

$ oc new-project kafka

2. Use the default configuration in kafka Install... In the project Red Hat Integration - AMQ Streams Operator, After success can be in Installed Operators see Red Hat Integration - AMQ Streams , And you can also execute the command to view pod and api-resources Resource state .

$ oc get pod -n kafka
NAME                                                   READY   STATUS    RESTARTS   AGE
amq-streams-cluster-operator-v1.4.0-59c7778c88-7bvzx   1/1     Running   0          22s
 
$ oc api-resources --api-group='kafka.strimzi.io'
NAME                 SHORTNAMES   APIGROUP           NAMESPACED   KIND
kafkabridges         kb           kafka.strimzi.io   true         KafkaBridge
kafkaconnectors      kctr         kafka.strimzi.io   true         KafkaConnector
kafkaconnects        kc           kafka.strimzi.io   true         KafkaConnect
kafkaconnects2is     kcs2i        kafka.strimzi.io   true         KafkaConnectS2I
kafkamirrormaker2s   kmm2         kafka.strimzi.io   true         KafkaMirrorMaker2
kafkamirrormakers    kmm          kafka.strimzi.io   true         KafkaMirrorMaker
kafkas               k            kafka.strimzi.io   true         Kafka
kafkatopics          kt           kafka.strimzi.io   true         KafkaTopic
kafkausers           ku           kafka.strimzi.io   true         KafkaUser

establish Kafka colony

1. After installation Red Hat Integration - AMQ Streams Operator Created in Kafka example . Use the default configuration to create Kafka Cluster instance , It contains 3 copy Kafka Instance and 3 copy Zookeeper example .
   
2. And then in “ developer ” View's “ Topology ” Page view Kafka Deployment status of resources .
   
3. Execute the command to view Kafka Resource status of the instance .

$ oc get kafka  -n kafka
NAME         DESIRED KAFKA REPLICAS   DESIRED ZK REPLICAS   READY   WARNINGS
my-cluster   3                        3                     True
 
$ oc get pod -n kafka
NAME                                                   READY   STATUS    RESTARTS   AGE
amq-streams-cluster-operator-v1.4.0-59c7778c88-7bvzx   1/1     Running   0          23m
my-cluster-entity-operator-c4cfc5695-zm5m7             3/3     Running   0          2s
my-cluster-kafka-0                                     1/2     Running   0          61s
my-cluster-kafka-1                                     1/2     Running   0          61s
my-cluster-kafka-2                                     1/2     Running   1          61s
my-cluster-zookeeper-0                                 2/2     Running   0          94s
my-cluster-zookeeper-1                                 2/2     Running   0          94s
my-cluster-zookeeper-2                                 2/2     Running   0          94s

function Topic application ：Hello World

establish Topic

1. After installation Red Hat Integration - AMQ Streams Operator Created in KafkaTopic example .
   
2. Execute the command to view my-topic The state of .

$ oc get kafkatopic my-topic -n kafka
NAME       CLUSTER      PARTITIONS   REPLICATION FACTOR   READY
my-topic   my-cluster   10           3                    True

Send and receive messages

1. According to the following, in kafka Create an application to send messages in the project hello-world-producer.

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: hello-world-producer
  name: hello-world-producer
spec:
  replicas: 1
  selector:
    matchLabels:
      app: hello-world-producer
  template:
    metadata:
      labels:
        app: hello-world-producer
    spec:
      containers:
      - name: hello-world-producer
        image: strimzici/hello-world-producer:latest
        env:
          - name: BOOTSTRAP_SERVERS
            value: my-cluster-kafka-bootstrap:9092
          - name: TOPIC
            value: my-topic
          - name: DELAY_MS
            value: "5000"
          - name: LOG_LEVEL
            value: "INFO"
          - name: MESSAGE_COUNT
            value: "5000"

2. Execute the command to view Pod Operating condition .

$ oc get pod -l app=hello-world-producer -n kafka
NAME                                   READY   STATUS    RESTARTS   AGE
hello-world-producer-f85d9f755-l77sz   1/1     Running   0          50s

3. stay OpenShift The console entry is called hello-world-producer-f85d9f755-l77sz Of Pod, Then check it out Logs.
   

4. According to the following, in kafka Create an application to send messages in the project hello-world-consumer.

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: hello-world-consumer
  name: hello-world-consumer
spec:
  replicas: 1
  selector:
    matchLabels:
      app: hello-world-consumer
  template:
    metadata:
      labels:
        app: hello-world-consumer
    spec:
      containers:
      - name: hello-world-consumer
        image: strimzici/hello-world-consumer:latest
        env:
          - name: BOOTSTRAP_SERVERS
            value: my-cluster-kafka-bootstrap:9092
          - name: TOPIC
            value: my-topic
          - name: GROUP_ID
            value: my-group
          - name: LOG_LEVEL
            value: "INFO"
          - name: MESSAGE_COUNT
            value: "5000"

5. Execute the command to view Pod Operation of the .

$ oc get pod -l app=hello-world-consumer -n kafka
NAME                                    READY   STATUS    RESTARTS   AGE
hello-world-consumer-6f9766f94c-l7wcp   1/1     Running   0          60s

6. Check out the name hello-world-consumer-6f9766f94c-l7wcp Of Pod Output log for . It can be confirmed that the test is the only one kafka-consumer Can be from number 0/1/2 One of the three partition receive data .
   
7. Carry out orders , Will run hello-world-consumer Of Pod The number increased to 2 individual .

$ oc scale deployment hello-world-consumer --replicas=2 -n kafka
 
$ oc get pod -l app=hello-world-consumer -n kafka
NAME                                    READY   STATUS    RESTARTS   AGE
hello-world-consumer-6f9766f94c-l7wcp   1/1     Running   0          4m16s
hello-world-consumer-6f9766f94c-ltdfr   1/1     Running   0          6s

8. stay OpenShift The console confirms the above 2 Run hello-world-consumer Of Pod Log . according to partition Number to confirm 1 individual Pod Only from 0-4 Number partition receive data , the other one Pod Can from 5-9 Number partition receive data .
   
   
9. Carry out orders , Will run hello-world-consumer Of Pod Again, the number drops to 1. And then again from Pod The log confirms that it is available from all partition Data received .

$ oc scale deployment hello-world-consumer --replicas=1 -n kafka

Reference resources

https://github.com/shpaz/openshift-data-workshop/tree/main/2-amq-persistent-odf