Kafka source connector doesn't reconnect after kafka restart

[aravindtga]

Describe the bug
Kafka source connector doesn't reconnect with Kafka after Kafka node failed and boots up again.

To Reproduce
Steps to reproduce the behaviour:

1. Run Kafka
2. Run pulsar(2.6.1 pulsar-all image)
3. Create and run source connector in pulsar to connect with Kafka
4. Now it can be seen that source connector fetches the message from Kafka and put it back in pulsar topic
5. Bring down Kafka
6. Run Kafka
7. Source connector fails to connect Kafka with the below error and doesn't fetch the Kafka messages anymore

14:06:16.273 [kafka-coordinator-heartbeat-thread | 2.0] WARN  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=2.0] This member will leave the group because consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
14:06:16.273 [kafka-coordinator-heartbeat-thread | 2.0] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=2.0] Member consumer-1-25a80c1d-78bc-4faa-b3d7-16bc259eaa07 sending LeaveGroup request to coordinator kafkaserver:9092 (id: 2147483647 rack: null)

7. source connector status is also showing as running without any error or restarts

root@c44d5fd34c32:/pulsar/bin# ./pulsar-admin sources status --name testsource
{
  "numInstances" : 1,
  "numRunning" : 1,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : true,
      "error" : "",
      "numRestarts" : 0,
      "numReceivedFromSource" : 2,
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "numSourceExceptions" : 0,
      "latestSourceExceptions" : [ ],
      "numWritten" : 2,
      "lastReceivedTime" : 1603366780740,
      "workerId" : "c-standalone-fw-localhost-8080"
    }
  } ]
}

Expected behaviour
Source connector should reconnect with Kafka.
If the failure is not recoverable, source connector status should throw the error/running=false

[michaeljmarshall]

Is there any work being done to resolve this bug?

[lhotari]

One thing to try is to enable "autoCommitEnabled" configuration for the Kafka source connector. It should be enabled by default unless it's explicitly disabled.

The reason why this might help is that Kafka Consumer API's commitSync method throws an unchecked exception according to this SO answer: https://stackoverflow.com/a/65193460 .
That case might not be handled in the Pulsar IO Kafka source connector around https://github.com/apache/pulsar-connectors/blob/master/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java#L139 .
commitSync won't be called when auto commit is enabled in the Pulsar IO Kafka source connector.

It might help to look at apache/kafka#5492 to see how a similiar issue (related to handling org.apache.kafka.common.errors.TimeoutException in commitSync) was fixed in Kafka's Mirror Maker.

[tisonkun]

Closed as stale. Please open a new issue if it's still relevant to the maintained versions.