[
https://issues.apache.org/jira/browse/KAFKA-16201?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18105026#comment-18105026
]
Arnab Nandy commented on KAFKA-16201:
-------------------------------------
[~carbonrider] thanks for the report. NotLeaderOrFollowerException is normally
retriable while the producer refreshes metadata, but the deployment appears to
advertise invalid broker addresses.
The StatefulSet creates pods named kafka-0, kafka-1, etc., under the kafka-svc
headless service, while *advertised.listeners* uses {*}kraft-$NODE_ID{*}. Kafka
clients receive these advertised addresses after bootstrapping, so they may be
unable to connect to the correct partition leader.
Please try:
{code:java}
ADVERTISED_LISTENERS="SASL://kafka-$NODE_ID.$SERVICE.$NAMESPACE.svc.cluster.local:9092,INTERNAL://kafka-$NODE_ID.$SERVICE.$NAMESPACE.svc.cluster.local:29092"
{code}
There are also deployment concerns to address:
- Each broker should have its own persistent volume, normally through
StatefulSet volumeClaimTemplates; all replicas currently share one
ReadWriteOnce PVC.
- hostPath storage is not appropriate for an EKS HA deployment.
- Cluster ID creation has a startup race because other pods may start before
pod 0 writes the file.
- *kafka-storage.sh format* should not run unconditionally on every restart.
- The reproduction shows {*}kafka-topics.bat{*}, but an interactive prompt
would come from {*}kafka-console-producer.bat{*}.
If the problem continues after fixing the advertised addresses and storage
layout, please provide the complete producer output, broker/controller logs,
effective broker configurations, and output from:
{code:java}
kafka-topics.sh --bootstrap-server <broker> \
--command-config producer.properties \
--describe --topic hello {code}
Based on the information currently available, this appears to be a deployment
configuration issue rather than a Kafka defect.
> Kafka exception - org.apache.kafka.common.errors.NotLeaderOrFollowerException
> -----------------------------------------------------------------------------
>
> Key: KAFKA-16201
> URL: https://issues.apache.org/jira/browse/KAFKA-16201
> Project: Kafka
> Issue Type: Bug
> Components: core
> Affects Versions: 3.6.1
> Environment: AWS EKS
> Reporter: Yogesh
> Priority: Blocker
>
> I am deploying Kafka inside Kubernetes cluster in HA mode (multiple brokers).
> The deployment consists of
> Kubernetes
> Kafka 3.6.1
> Refer to the following files used in the deployment
> Dockerfile
>
> {code:java}
> FROM eclipse-temurin:17.0.9_9-jdk-jammy
> ENV KAFKA_VERSION=3.6.1
> ENV SCALA_VERSION=2.13
> ENV KAFKA_HOME=/opt/kafka
> ENV PATH=${PATH}:${KAFKA_HOME}/bin
> LABEL name="kafka" version=${KAFKA_VERSION}
> RUN wget -O /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz
> https://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz
> \
> && tar xfz /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz -C /opt \
> && rm /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz \
> && ln -s /opt/kafka_${SCALA_VERSION}-${KAFKA_VERSION} ${KAFKA_HOME} \
> && rm -rf /tmp/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz
> COPY ./entrypoint.sh /
> RUN ["chmod", "+x", "/entrypoint.sh"]
> ENTRYPOINT ["/entrypoint.sh"] {code}
>
>
> entrypoint.sh
>
> {code:java}
> #!/bin/bash
> NODE_ID=${HOSTNAME:6}
> LISTENERS="SASL://:9092,CONTROLLER://:9093,INTERNAL://:29092"
> ADVERTISED_LISTENERS="SASL://kraft-$NODE_ID:9092,INTERNAL://kafka-$NODE_ID.$SERVICE.$NAMESPACE.svc.cluster.local:29092"
> CONTROLLER_QUORUM_VOTERS=""
> for i in $( seq 0 $REPLICAS); do
> if [[ $i != $REPLICAS ]]; then
>
> CONTROLLER_QUORUM_VOTERS="$CONTROLLER_QUORUM_VOTERS$i@kafka-$i.$SERVICE.$NAMESPACE.svc.cluster.local:9093,"
> else
> CONTROLLER_QUORUM_VOTERS=${CONTROLLER_QUORUM_VOTERS::-1}
> fi
> done
> mkdir -p $SHARE_DIR/$NODE_ID
> if [[ ! -f "$SHARE_DIR/cluster_id" && "$NODE_ID" = "0" ]]; then
> CLUSTER_ID=$(kafka-storage.sh random-uuid)
> echo $CLUSTER_ID > $SHARE_DIR/cluster_id
> else
> CLUSTER_ID=$(cat $SHARE_DIR/cluster_id)
> fi
> sed -e "s+^node.id=.*+node.id=$NODE_ID+" \
> -e
> "s+^controller.quorum.voters=.*+controller.quorum.voters=$CONTROLLER_QUORUM_VOTERS+"
> \
> -e "s+^listeners=.*+listeners=$LISTENERS+" \
> -e "s+^advertised.listeners=.*+advertised.listeners=$ADVERTISED_LISTENERS+" \
> -e "s+^log.dirs=.*+log.dirs=$SHARE_DIR/$NODE_ID+" \
> /opt/kafka/config/kraft/server.properties > server.properties.updated \
> && mv server.properties.updated /opt/kafka/config/kraft/server.properties
> JAAS="org.apache.kafka.common.security.plain.PlainLoginModule required
> username=\"admin\" password=\"admin-secret\" user_admin=\"admin-secret\";"
> echo -e "\nlistener.name.sasl.plain.sasl.jaas.config=${JAAS}" >>
> /opt/kafka/config/kraft/server.properties
> echo -e "\nsasl.enabled.mechanisms=PLAIN" >>
> /opt/kafka/config/kraft/server.properties
> echo -e
> "\nlistener.security.protocol.map=SASL:SASL_PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"
> >> /opt/kafka/config/kraft/server.properties
> echo -e "\ninter.broker.listener.name=INTERNAL" >>
> /opt/kafka/config/kraft/server.properties
> kafka-storage.sh format -t $CLUSTER_ID -c
> /opt/kafka/config/kraft/server.properties
> exec kafka-server-start.sh /opt/kafka/config/kraft/server.properties {code}
>
>
> Kafka.yaml
> {code:java}
> apiVersion: v1
> kind: Namespace
> metadata:
> name: kafka-kraft
> ---
> apiVersion: v1
> kind: PersistentVolume
> metadata:
> name: kafka-pv-volume
> labels:
> type: local
> spec:
> storageClassName: manual
> capacity:
> storage: 10Gi
> accessModes:
> - ReadWriteOnce
> hostPath:
> path: '/mnt/data'
> ---
> apiVersion: v1
> kind: PersistentVolumeClaim
> metadata:
> name: kafka-pv-claim
> namespace: kafka-kraft
> spec:
> storageClassName: manual
> accessModes:
> - ReadWriteOnce
> resources:
> requests:
> storage: 3Gi
> ---
> apiVersion: v1
> kind: Service
> metadata:
> name: kafka-svc
> labels:
> app: kafka-app
> namespace: kafka-kraft
> spec:
> clusterIP: None
> ports:
> - name: '9092'
> port: 9092
> protocol: TCP
> targetPort: 9092
> selector:
> app: kafka-app
> ---
> apiVersion: apps/v1
> kind: StatefulSet
> metadata:
> name: kafka
> labels:
> app: kafka-app
> namespace: kafka-kraft
> spec:
> serviceName: kafka-svc
> replicas: 5
> selector:
> matchLabels:
> app: kafka-app
> template:
> metadata:
> labels:
> app: kafka-app
> spec:
> volumes:
> - name: kafka-storage
> persistentVolumeClaim:
> claimName: kafka-pv-claim
> containers:
> - name: kafka-container
> image: myimage/kafka-kraft:1.0
> ports:
> - containerPort: 9092
> - containerPort: 9093
> env:
> - name: REPLICAS
> value: '5'
> - name: SERVICE
> value: kafka-svc
> - name: NAMESPACE
> value: kafka-kraft
> - name: SHARE_DIR
> value: /mnt/kafka
> volumeMounts:
> - name: kafka-storage
> mountPath: /mnt/kafka {code}
> After the deployment all the containers are up and running. I then connect
> the broker using following command
>
> {code:java}
> .\kafka-topics.bat --bootstrap-server
> kraft-0:9092,kraft-1:9092,kraft-2:9092,kraft-3:9092,kraft-4:9092
> --command-config producer.properties --topic hello --create
> --replication-factor 5 {code}
>
>
> producer.properties
>
> {code:java}
> sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
> required username=admin password=admin-secret;
> security.protocol=SASL_PLAINTEXT
> sasl.mechanism=PLAIN
> metadata.max.age.ms=1000 {code}
>
>
> A prompt is displayed to enter a messag. Upon a sample text it throws
> following error.
> [Producer clientId=console-producer] Received invalid metadata error in
> produce request on partition hello2-1 due to
> org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests
> intended only for the leader, this error indicates that the broker is not the
> current leader. For requests intended for any replica, this error indicates
> that the broker is not a replica of the topic partition.. Going to request
> metadata update now
>
> What I have tried so far
> * Tried zookeeper and kraft mode
> * Tried deleting and recreating the topics (This works randomly)
> Unfortunately, the problem persists and not able to produce messages.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
