Test failures for console producer/consumer with min.insync.replicas=2

[allquantor]

Running the Kafka tests this one do not complete

NAME                    READY     STATUS    RESTARTS   AGE
kafkacat-4kvj4          3/3       Running   0          1h
produce-consume-kdd92   2/3       Running   0          1h

Logs from Testcase

Nothing consumed yet
Nothing consumed yet
Nothing consumed yet
Nothing consumed yet
Nothing consumed yet
Nothing consumed yet
Nothing consumed yet
....

Is it probably the producer does not generate enough data? No errors in producer or consumer.

Using kubernetes 1.8.5 on GKE

[StevenACoffman]

@solsson Actually, I synched with latest recently and I started getting the same test failures, where where I didn't with this repo pulled on Nov 12, 2017.

Not sure if it's related, but I have a java consumer where now I keep getting the coordinator marked as dead.

 teachers/splitterator-pure-749947f75f-r8d5k[splitterator]: [pool-1-thread-1] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=splitterator-pure2] (Re-)joining group
teachers/splitterator-pure-749947f75f-r8d5k[splitterator]: [pool-1-thread-1] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=splitterator-pure2] Marking the coordinator kafka-1.broker.kafka.svc.cluster.local:9092 (id: 2147483646 rack: null) dead
teachers/splitterator-pure-749947f75f-r8d5k[splitterator]: [pool-1-thread-1] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=splitterator-pure2] Discovered coordinator kafka-1.broker.kafka.svc.cluster.local:9092 (id: 2147483646 rack: null)
teachers/splitterator-pure-749947f75f-r8d5k[splitterator]: [pool-1-thread-1] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=splitterator-pure2] (Re-)joining group

Previously, the identical java consumer worked fine. I haven't figured this out yet.
Kubernetes 1.8.4 on AWS via kops.

[solsson]

Looks bad. I'm sorry about leaving master relatively untested. Could be the cause of #114 too. I've been caught up in not only holiday season but also #101 (comment).

I wrote these tests while working on new features, but now that they run constantly I find the readiness indication very useful. I will try to reproduce the above issues.

[StevenACoffman]

If I revert to d01c128 before #107 the tests go (eventually) back to:

kubectl get pods -l test-type=readiness --namespace=test-kafka
NAME                    READY     STATUS    RESTARTS   AGE
kafkacat-pflfs          3/3       Running   2          22m
produce-consume-xjbmg   3/3       Running   0          22m

Not sure what's causing the problem though.

[StevenACoffman]

Hrm, even with the tests succeeding on the old d01c128 , the yahoo kafka-manager addon is failing to do much of anything:

[info] k.m.a.KafkaManagerActor - baseZkPath=/kafka-manager
[info] o.a.z.ClientCnxn - Socket connection established to zookeeper.kafka.svc.cluster.local/100.70.176.112:2181, initiating session
[info] o.a.z.ClientCnxn - Unable to read additional data from server sessionid 0x0, likely server has closed socket, closing socket connection and attempting reconnect
[info] play.api.Play - Application started (Prod)
[info] p.c.s.NettyServer - Listening for HTTP on /0:0:0:0:0:0:0:0:80
[info] o.a.z.ClientCnxn - Opening socket connection to server zookeeper.kafka.svc.cluster.local/100.70.176.112:2181. Will not attempt to authenticate using SASL (unknown error)
[info] o.a.z.ClientCnxn - Socket connection established to zookeeper.kafka.svc.cluster.local/100.70.176.112:2181, initiating session
[info] o.a.z.ClientCnxn - Unable to read additional data from server sessionid 0x0, likely server has closed socket, closing socket connection and attempting reconnect

This is weird, since I can do this:

kubectl exec -it zoo-0 -n kafka -- /bin/bash
root@zoo-0:/opt/kafka# echo ruok | nc -w 1 -q 1 zookeeper.kafka.svc.cluster.local 2181
imokroot@zoo-0:/opt/kafka#

If I go to the actual kafka-manager (via dashboard /api/v1/namespaces/kafka/services/kafka-manager/proxy/):

Yikes! Ask timed out on [ActorSelection[Anchor(akka://kafka-manager-system/), Path(/user/kafka-manager)]] after [5000 ms]

[solsson]

Let's assume that the issue(s?) is caused by min.insync.replicas=2. #108 and #114 were both caused by this change.

I fail to reproduce the test issue. Kafka Manager works fine too. @allquantor Did you upgrade an existing cluster, that used to have min.insync.replicas=1. Can you see any broker log messages like those in #108? I don't know if console-consumer and console-producer can produce debug output, like -b broker,topic with kafkacat.

The test can be improved. min.insync.replicas can be overridden per producer, by setting acks explicitly. Also the topic creation job can be removed. It specifies --replication-factor=2. Not sure how this works with min.insync.replicas. But the kafkacat test has the same topic creation, and looks ok.

[solsson]

I noticed when doing e059690 that default min.insync.replicas is 1. Thus the tests were probably unaffected by the new default 2.

The change might still have affected Kafka internals, as in #114.

[albamoro]

I'm seeing the same issue as @StevenACoffman but with the stock kafka-console-consumer using it both inside and outside the cluster: the consumer is stuck in a loop marking a coordinator resolved as id = max(int32) as dead.

[solsson]

@albamoro Any clues in broker logs?

[albamoro]

None that I could see on the server side, but then I'm fairly new to them - I've upped the logging and been tailing the kafka-request.log to not much avail so far.

Forgot to mention an important detail: the console works when using the old consumer, i.e., pointing to --zookeeper rather than --bootstrap-server. Clients and server versions are aligned as they are based on the same image.

Client-wise this is the block that keeps on repeating - note the node's id is lost between lines 2 & 3. I couldn't tell how relevant that is.

2018-01-09 14:20:32 DEBUG AbstractCoordinator:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Sending GroupCoordinator request to broker 192.168.1.21:32402 (id: 2 rack: null)
2018-01-09 14:20:32 DEBUG AbstractCoordinator:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Received GroupCoordinator response ClientResponse(receivedTimeMs=1515507632283, latencyMs=1, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1, clientId=consumer-1, correlationId=251), responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=NONE, node=192.168.1.21:32400 (id: 0 rack: null)))
2018-01-09 14:20:32 INFO  AbstractCoordinator:341 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Discovered coordinator 192.168.1.21:32400 (id: 2147483647 rack: null)
2018-01-09 14:20:32 DEBUG NetworkClient:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Initiating connection to node 192.168.1.21:32400 (id: 2147483647 rack: null)
2018-01-09 14:20:32 DEBUG AbstractCoordinator:177 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Disabling heartbeat thread
2018-01-09 14:20:32 INFO  AbstractCoordinator:336 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] (Re-)joining group
2018-01-09 14:20:32 DEBUG AbstractCoordinator:189 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Sending JoinGroup ((type: JoinGroupRequest, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0, sessionTimeout=10000, rebalanceTimeout=300000, memberId=consumer-1-5a07a421-70d7-4ae0-9be4-cbf0d2babfa1, protocolType=consumer, groupProtocols=org.apache.kafka.common.requests.JoinGroupRequest$ProtocolMetadata@21918021)) to coordinator 192.168.1.21:32400 (id: 2147483647 rack: null)
2018-01-09 14:20:32 DEBUG Selector:195 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Created socket with SO_RCVBUF = 65536, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to node 2147483647
2018-01-09 14:20:32 DEBUG NetworkClient:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Completed connection to node 2147483647. Fetching API versions.
2018-01-09 14:20:32 DEBUG NetworkClient:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Initiating API versions fetch from node 2147483647.
2018-01-09 14:20:32 DEBUG NetworkClient:189 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Recorded API versions for node 2147483647: (Produce(0): 0 to 5 [usable: 5], Fetch(1): 0 to 6 [usable: 6], ListOffsets(2): 0 to 2 [usable: 2], Metadata(3): 0 to 5 [usable: 5], LeaderAndIsr(4): 0 to 1 [usable: 1], StopReplica(5): 0 [usable: 0], UpdateMetadata(6): 0 to 4 [usable: 4], ControlledShutdown(7): 0 to 1 [usable: 1], OffsetCommit(8): 0 to 3 [usable: 3], OffsetFetch(9): 0 to 3 [usable: 3], FindCoordinator(10): 0 to 1 [usable: 1], JoinGroup(11): 0 to 2 [usable: 2], Heartbeat(12): 0 to 1 [usable: 1], LeaveGroup(13): 0 to 1 [usable: 1], SyncGroup(14): 0 to 1 [usable: 1], DescribeGroups(15): 0 to 1 [usable: 1], ListGroups(16): 0 to 1 [usable: 1], SaslHandshake(17): 0 to 1 [usable: 1], ApiVersions(18): 0 to 1 [usable: 1], CreateTopics(19): 0 to 2 [usable: 2], DeleteTopics(20): 0 to 1 [usable: 1], DeleteRecords(21): 0 [usable: 0], InitProducerId(22): 0 [usable: 0], OffsetForLeaderEpoch(23): 0 [usable: 0], AddPartitionsToTxn(24): 0 [usable: 0], AddOffsetsToTxn(25): 0 [usable: 0], EndTxn(26): 0 [usable: 0], WriteTxnMarkers(27): 0 [usable: 0], TxnOffsetCommit(28): 0 [usable: 0], DescribeAcls(29): 0 [usable: 0], CreateAcls(30): 0 [usable: 0], DeleteAcls(31): 0 [usable: 0], DescribeConfigs(32): 0 [usable: 0], AlterConfigs(33): 0 [usable: 0], AlterReplicaLogDirs(34): 0 [usable: 0], DescribeLogDirs(35): 0 [usable: 0], SaslAuthenticate(36): 0 [usable: 0], CreatePartitions(37): 0 [usable: 0])
2018-01-09 14:20:32 DEBUG AbstractCoordinator:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Received successful JoinGroup response: org.apache.kafka.common.requests.JoinGroupResponse@35b15e13
2018-01-09 14:20:32 DEBUG ConsumerCoordinator:189 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Performing assignment using strategy range with subscriptions {consumer-1-5a07a421-70d7-4ae0-9be4-cbf0d2babfa1=Subscription(topics=[clickstream_raw])}
2018-01-09 14:20:32 DEBUG ConsumerCoordinator:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Finished assignment for group: {consumer-1-5a07a421-70d7-4ae0-9be4-cbf0d2babfa1=Assignment(partitions=[clickstream_raw-0])}
2018-01-09 14:20:32 DEBUG AbstractCoordinator:189 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Sending leader SyncGroup to coordinator 192.168.1.21:32400 (id: 2147483647 rack: null): (type=SyncGroupRequest, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0, generationId=63, memberId=consumer-1-5a07a421-70d7-4ae0-9be4-cbf0d2babfa1, groupAssignment=consumer-1-5a07a421-70d7-4ae0-9be4-cbf0d2babfa1)
2018-01-09 14:20:32 DEBUG AbstractCoordinator:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] SyncGroup failed:
2018-01-09 14:20:32 INFO  AbstractCoordinator:341 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Marking the coordinator 192.168.1.21:32400 (id: 2147483647 rack: null) dead
2018-01-09 14:20:32 DEBUG NetworkClient:189 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Manually disconnected from 2147483647. Removed requests: .
2018-01-09 14:20:32 DEBUG AbstractCoordinator:183 - [Consumer clientId=consumer-1, groupId=spark-kafka-source-bc5d5601-5f1a-4b96-9193-7cb9e6c553fc--751033765-driver-0] Sending GroupCoordinator request to broker 192.168.1.21:32402 (id: 2 rack: null)

[solsson]

From the port numbers it looks like you get the "outside" listener (#78). I'm quite curious how that can happen in the tests. An ordering issue was resolved in 4c202f4. Might #99 (comment) be helpful to you too?

[albamoro]

Same happens using the inside listener. I had attached the external client's logs because those are the only ones I've been able to configure for DEBUG logging. You can see below the console's output when used within my Minikube, v0.24.1.

Re. commit 4c202f4, my repo is pointing at bbed23e so I presume the fix applies? I haven't checked #99 yet - I was planning on rolling back to early Nov as @StevenACoffman had done and work from there.

root@pzoo-1:/opt/kafka/bin# ./kafka-console-consumer.sh --bootstrap-server kafka-0.broker.kafka.svc.cluster.local:9092 --topic xxxx --from-beginning
[2018-01-10 10:39:26,845] INFO ConsumerConfig values:
        auto.commit.interval.ms = 5000
        auto.offset.reset = earliest
        bootstrap.servers = [kafka-0.broker.kafka.svc.cluster.local:9092]
        check.crcs = true
        client.id =
        connections.max.idle.ms = 540000
        enable.auto.commit = true
        exclude.internal.topics = true
        fetch.max.bytes = 52428800
        fetch.max.wait.ms = 500
        fetch.min.bytes = 1
        group.id = console-consumer-77574
        heartbeat.interval.ms = 3000
        interceptor.classes = null
        internal.leave.group.on.close = true
        isolation.level = read_uncommitted
        key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
        max.partition.fetch.bytes = 1048576
        max.poll.interval.ms = 300000
        max.poll.records = 500
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
        receive.buffer.bytes = 65536
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        request.timeout.ms = 305000
        retry.backoff.ms = 100
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT
        send.buffer.bytes = 131072
        session.timeout.ms = 10000
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
        ssl.endpoint.identification.algorithm = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLS
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS
        value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
 (org.apache.kafka.clients.consumer.ConsumerConfig)
[2018-01-10 10:39:26,893] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser)
[2018-01-10 10:39:26,893] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser)
[2018-01-10 10:39:26,973] INFO [Consumer clientId=consumer-1, groupId=console-consumer-77574] Discovered coordinator kafka-0.broker.kafka.svc.cluster.local:9092 (id: 2147483647 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2018-01-10 10:39:26,976] INFO [Consumer clientId=consumer-1, groupId=console-consumer-77574] Revoking previously assigned partitions [] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
[2018-01-10 10:39:26,976] INFO [Consumer clientId=consumer-1, groupId=console-consumer-77574] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2018-01-10 10:39:27,031] INFO [Consumer clientId=consumer-1, groupId=console-consumer-77574] Marking the coordinator kafka-0.broker.kafka.svc.cluster.local:9092 (id: 2147483647 rack: null) dead (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2018-01-10 10:39:27,144] INFO [Consumer clientId=consumer-1, groupId=console-consumer-77574] Discovered coordinator kafka-0.broker.kafka.svc.cluster.local:9092 (id: 2147483647 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2018-01-10 10:39:27,145] INFO [Consumer clientId=consumer-1, groupId=console-consumer-77574] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)

[albamoro]

@solsson, just to confirm your comment above, #116 (comment), changing min.insync.replicas=1 whilst retaining default.replication.factor=3 clears these issues in my setup: produce-consume test fails & console consumer cannot use --bootstrap-server