1. Install AMQ streams
    2. Create a Kafka instance
$oc apply -f kafka.yaml
$ cat kafka.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: openshift-operators
spec:
  kafka:
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: '3.0'
      inter.broker.protocol.version: '3.0'
    storage:
      type: ephemeral
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    version: 3.6.0
    replicas: 3
  entityOperator:
    topicOperator: {}
    userOperator: {}
  zookeeper:
    storage:
      type: ephemeral
    replicas: 3

Check the Kafka cluster status
$watch oc get kafka.kafka.strimzi.io/my-cluster -n openshift-operators

    3. Create Kafka Topic my-topic
$oc apply -f my-topic.yaml
$ cat topic.yaml
kind: KafkaTopic
apiVersion: kafka.strimzi.io/v1beta2
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
  namespace: openshift-operators
spec:
  partitions: 5
  replicas: 3
  config:
    retention.ms: 604800000
    segment.bytes: 1073741824
Check the topics
$oc get KafkaTopic my-topic -n openshift-operators

4. Deploy Kafka Consumer application
oc apply -f deployment.yaml
$ cat deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-amqstreams-consumer
  labels:
    app: kafka-amqstreams-consumer
spec:
  selector:
    matchLabels:
      app: kafka-amqstreams-consumer
  template:
    metadata:
      labels:
        app: kafka-amqstreams-consumer
    spec:
      containers:
      - name: kafka-amqstreams-consumer
        image: quay.io/zroubalik/kafka-app:latest
        env:
          - name: BOOTSTRAP_SERVERS
            value: my-cluster-kafka-bootstrap.openshift-operators.svc:9092
        resources:
            requests:
              cpu: 100m
              memory: 100Mi
            limits:
              cpu: 500m
              memory: 500Mi
        command:
        - /kafkaconsumerapp
        securityContext:
          #runAsNonRoot: true
          seLinuxOptions:
            level: "s0:c0,c26"
          allowPrivilegeEscalation: false
          capabilities:
            drop:
              - ALL
          seccompProfile:
            type: RuntimeDefault


Verify the consumer has been able to connect to Kafka instance
$oc logs deployment.apps/kafka-amqstreams-consumer

5. Deploy ScaledObject to enable Kafka Consumer application autoscaling
Deploy a ScaledObject
3.1 $oc apply -f scaledobject.yaml
$ cat scaledobject.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: kafka-amqstreams-consumer-scaledobject
spec:
  minReplicaCount: 0
  maxReplicaCount: 5
  cooldownPeriod: 5
  pollingInterval: 10
  scaleTargetRef:
    name: kafka-amqstreams-consumer
  triggers:
  - type: kafka
    metadata:
      topic: my-topic
      bootstrapServers: my-cluster-kafka-bootstrap.openshift-operators.svc:9092
      consumerGroup: my-group
      lagThreshold: '1'
      offsetResetPolicy: 'latest'
Check that KEDA has been able to access metrics and is correctly defined for autoscaling:
$oc get scaledobject kafka-amqstreams-consumer-scaledobject

5. Send more messages to Kafka to test the Kafka Consumer application autoscaling
 $oc create -f load.yaml
$cat load.yaml
apiVersion: batch/v1
kind: Job
metadata:
  generateName: kafka-amqstreams-producer-
spec:
  parallelism: 1
  completions: 1
  backoffLimit: 1
  template:
    metadata:
      name: kafka-amqstreams-producer
      labels:
        app: kafka-amqstreams-producer
    spec:
      restartPolicy: Never
      containers:
      - name: kafka-amqstreams-producer
        image: quay.io/zroubalik/kafka-app:latest
        command: [ "/kafkaproducerapp" ]
        env:
          - name: BOOTSTRAP_SERVERS
            value: my-cluster-kafka-bootstrap.openshift-operators.svc:9092
          - name: TOPIC
            value: my-topic
          - name: MESSAGE_COUNT
            value: "1500"                    # increase this for a bigger load
          - name: DELAY_MS
            value: "100"


Check  Kafka Consumer application 

#watch oc get deployment.apps/kafka-amqstreams-consumer

    

Every 2.0s: oc get deployment.apps/kafka-amqstreams-consumer                                                                     

NAME                        READY   UP-TO-DATE   AVAILABLE   AGE
kafka-amqstreams-consumer   1/1     1            1           20m  

    

 
  Every 2,0s: oc get deployment.apps/kafka-amqstreams-consumer
NAME                        READY   UP-TO-DATE   AVAILABLE   AGE
kafka-amqstreams-consumer   5/5     5            5           21m

### After some time the application should be autoscaled back to 0

Every 2,0s: oc get deployment.apps/kafka-amqstreams-consumer
NAME                        READY   UP-TO-DATE   AVAILABLE   AGE
kafka-amqstreams-consumer   0/0     0            0           23m