Strimzi Kafka consume from closest replica

[brunooliveiramac]

Hi there,
Im following this doc to configure rack awareness in my consumers and also set the proper configs in my strimzi kafka:

https://strimzi.io/docs/operators/latest/configuring.html#consuming_messages_from_the_closest_replicas

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
  namespace: kafka-operator
spec:
  entityOperator:
    template:
      pod:
        affinity:
          nodeAffinity:
            preferredDuringSchedulingIgnoredDuringExecution:
            - preference:
                matchExpressions:
                - key: type
                  operator: In
                  values:
                  - kafka
              weight: 1
        tolerations:
        - effect: NoSchedule
          key: kafka-node-pool
          operator: Exists
    topicOperator:
      watchedNamespace: vikingcloud
    userOperator: {}
  kafka:
    config:
      default.replication.factor: 2
      min.insync.replicas: 1
      num.network.threads: 4
      num.partitions: 1
      offsets.topic.replication.factor: 1
      replica.selector.class: org.apache.kafka.common.replica.RackAwareReplicaSelector
      transaction.state.log.min.isr: 1
      transaction.state.log.replication.factor: 1
    listeners:
    - configuration:
        bootstrap:
          annotations:
            networking.gke.io/load-balancer-type: Internal
            service.kubernetes.io/topology-aware-hints: auto
          loadBalancerIP: 10.235.0.199
        brokers:
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 0
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 1
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 2
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 3
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 4
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 5
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 6
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 7
        - annotations:
            networking.gke.io/load-balancer-type: Internal
          broker: 8
      name: plain
      port: 9092
      tls: false
      type: loadbalancer
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          key: kafka-metrics-config.yml
          name: kafka-metrics
    rack:
      topologyKey: topology.kubernetes.io/zone
    replicas: 9
    resources:
      limits:
        cpu: "8"
        memory: 33Gi
      requests:
        cpu: "6"
        memory: 8Gi
    storage:
      class: kafka-pd
      deleteClaim: false
      size: 22000Gi
      type: persistent-claim
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: type
                  operator: In
                  values:
                  - kafka
        tolerations:
        - effect: NoSchedule
          key: kafka-node-pool
          operator: Exists
        topologySpreadConstraints:
        - maxSkew: 1
          nodeAffinityPolicy: Honor
          nodeTaintsPolicy: Honor
          topologyKey: kubernetes.io/hostname
          whenUnsatisfiable: DoNotSchedule
        - maxSkew: 1
          nodeAffinityPolicy: Honor
          nodeTaintsPolicy: Honor
          topologyKey: topology.kubernetes.io/zone
          whenUnsatisfiable: DoNotSchedule
  kafkaExporter:
    groupRegex: .*
    template:
      pod:
        affinity:
          nodeAffinity:
            preferredDuringSchedulingIgnoredDuringExecution:
            - preference:
                matchExpressions:
                - key: type
                  operator: In
                  values:
                  - kafka
              weight: 1
        tolerations:
        - effect: NoSchedule
          key: kafka-node-pool
          operator: Exists
    topicRegex: .*
  zookeeper:
    replicas: 3
    resources:
      limits:
        cpu: "2"
        memory: 5Gi
      requests:
        cpu: "0.5"
        memory: 2Gi
    storage:
      class: kafka-pd
      deleteClaim: false
      size: 10Gi
      type: persistent-claim
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: type
                  operator: In
                  values:
                  - kafka
        tolerations:
        - effect: NoSchedule
          key: kafka-node-pool
          operator: Exists
        topologySpreadConstraints:
        - maxSkew: 1
          nodeAffinityPolicy: Honor
          nodeTaintsPolicy: Honor
          topologyKey: kubernetes.io/hostname
          whenUnsatisfiable: DoNotSchedule
        - maxSkew: 1
          nodeAffinityPolicy: Honor
          nodeTaintsPolicy: Honor
          topologyKey: topology.kubernetes.io/zone
          whenUnsatisfiable: DoNotSchedule

For consumers we set

  public ConsumerFactory<String, Event> eventIngestionConsumerFactory(@NotNull KafkaProperties properties) {
    Map<String, Object> consumerProps = properties.buildConsumerProperties();
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
    consumerProps.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
    consumerProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
    consumerProps.put(KafkaAvroDeserializerConfig.AVRO_USE_LOGICAL_TYPE_CONVERTERS_CONFIG, true);
    **consumerProps.put(ConsumerConfig.CLIENT_RACK_CONFIG, kafkaReplicaZone);**
    return new DefaultKafkaConsumerFactory<>(consumerProps);
  }

But during consumer, I'm not seeing it fetching from closest. This pod is running in zone c, but consuming from zone b.
The topic has replication set to 3 and I have 3 zones, a, b, and f.

Added READ_UNCOMMITTED fetch request for partition event-ingestion-topic-53 at position FetchPosition{offset=2450619361, offsetEpoch=Optional[6], currentLeader=LeaderAndEpoch{leader=Optional[10.235.0.5:9092 (id: 5 rack: us-central1-b)], epoch=6}} to node 10.235.0.5:9092 (id: 5 rack: us-central1-b)

Does anyone face something similar? This is affecting our cloud costs due to the cross-zone traffic.

[scholzj]

But during consumer, I'm not seeing it fetching from closest. This pod is running in zone c, but consuming from zone b.
The topic has replication set to 3 and I have 3 zones, a, b, and f.

If I read this right, then you have a topic with replicas in zones a, b and f. And a client in zone c? In that case, it will read from the leader (that is possibly in the zone b?). In order to have the client read from the zone c, you would need to have also some in-sync replica in the zone c.

[brunooliveiramac]

Im sorry
The corrct nodes are c, b anf f

[scholzj]

So, in that case it should work assuming:

• The partition has a replica there
• The replica is in-sync
• The rack IDs are the same

To be honest, it is a long time since I tried it (and wrote a blog post about it). But I don't think there were any major changes to this.

[brunooliveiramac]

Oh I followed that post, it's really good.
I have 9 nodes and 3 zones, meaning 2 nodes in each
Humm Im thinking then if the replicas aren't properly evenly spread among the zones

[brunooliveiramac]

@scholzj for example, in this case:

These are my nodes

Pod: cluster-kafka-0, Node: gke-prod-cluster-kafka-node-pool-3cb7dc61-2e1j, Zone: us-central1-c
Pod: cluster-kafka-1, Node: gke-prod-cluster-kafka-node-pool-a57bbd45-uhlk, Zone: us-central1-f
Pod: cluster-kafka-2, Node: gke-prod-cluster-kafka-node-pool-998f8ce7-irl2, Zone: us-central1-b
Pod: cluster-kafka-3, Node: gke-prod-cluster-kafka-node-pool-3cb7dc61-aj1x, Zone: us-central1-c
Pod: cluster-kafka-4, Node: gke-prod-cluster-kafka-node-pool-a57bbd45-f6d8, Zone: us-central1-f
Pod: cluster-kafka-5, Node: gke-prod-cluster-kafka-node-pool-998f8ce7-h1eu, Zone: us-central1-b
Pod: cluster-kafka-6, Node: gke-prod-cluster-kafka-node-pool-a57bbd45-wrt3, Zone: us-central1-f
Pod: cluster-kafka-7, Node: gke-prod-cluster-kafka-node-pool-998f8ce7-jr40, Zone: us-central1-b
Pod: cluster-kafka-8, Node: gke-prod-cluster-kafka-node-pool-3cb7dc61-zgcm, Zone: us-central1-c

For example, partition 0, has replicas in zones, C (replica 8 and 0) and F (replica 1)

Which means that if a consumer tries to fetch from partition 0 and its located in zone B, it wont' fetch from closest as there is no replication for that partition available, right?

[scholzj]

That means you don't have the replicas properly distributed. You should be able to redistribute it for example using Cruise Control.