Should Kafka Source Connector itsself after unrecoverable error?

[relief-melone]

This is concerning the pulsar-io/kafka-source. Before I create an issue regarding this topic I would like to have some insight if the current behaviour is actually wrong or if there are already config values to adress this.

We recently had an issue that caused the consumer in a kafka source connector to get kicked out of kafkas consumer group. The connector is running as a pod inside of kubernetes. The consumer is unrecoverable at that point so the source connector should have two options

• create a new consumer and connect
• end the process which would in turn result in another pod being spun up if this is a scenario where the connector is deployed in kubernetes

However I feel like the connector just sitting there not consuming messages from kafka anymore is the wrong behaviour. Am I missing a config value? Would be happy to get your opinion on that and gladly create an issue as well if necessary

[shibd]

end the process which would in turn result in another pod being spun up if this is a scenario where the connector is deployed in kubernetes

In the Kubernetes runtime, It should be this behavior.

However I feel like the connector just sitting there and not consuming messages from Kafka anymore is the wrong behavior. Am I missing a config value?

In pulsar-io-framework, if a connector throws an exception, it will trigger this process exit.

So, maybe the Kafka source connector does not throw an exception on your case.

My guess is the logic handled here:

pulsar/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java

Lines 62 to 64 in 82237d3

	} else {
	sourceRecord.fail();
	}

I found a way to compare hacks to get the connector to throw exceptions. Maybe you can give it a try: set your processingGuarantees = EFFECTIVELY_ONCE

pulsar/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java

Lines 153 to 156 in 82237d3

	if (pulsarSourceConfig.getProcessingGuarantees()
	== FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
	throw new RuntimeException("Failed to process message: " + message.getMessageId());
	}

Can you share the logs of your Kafka source connector? Maybe we can optimize for this scenario(let it throw an exception to trigger connector process exit).

[relief-melone]

Thanks for the answer. I think that mitigation is an option for the kafka sink connector no? In the Kafka source connector you should have a Kafka Consumer and a Pulsar Producer. In our case the failing thread was the Kafka Consumer though. Since a Consumer that has been removed from the consumer group cannot be recovered I'd interpret the current behaviour as a bug then no? The consumer in that case should either be recreated or the process should exit.

[RobertIndie]

I think the root cause is here:

pulsar/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java

Lines 191 to 192 in 1c813fd

	runnerThread.setUncaughtExceptionHandler(
	(t, e) -> LOG.error("[{}] Error while consuming records", t.getName(), e));

Did you see some messages like Error while consuming records in the connector log? The Kafka source connector will print these exceptions but not exit the current process.

I think this could be considered a bug. We need to fail the connector if there are any unexpected errors.

[relief-melone]

I think the root cause is here:

pulsar/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java

Lines 191 to 192 in 1c813fd

	runnerThread.setUncaughtExceptionHandler(
	(t, e) -> LOG.error("[{}] Error while consuming records", t.getName(), e));

Did you see some messages like Error while consuming records in the connector log? The Kafka source connector will print these exceptions but not exit the current process.

I think this could be considered a bug. We need to fail the connector if there are any unexpected errors.

Thanks for the PR. It seems like that could very well be the reason for the behaviour. This is the portion of the Error log in our particular case that might be interesting for you. So the Error while consuming records does indeed show. Happy to try it out.

2023-03-15T12:09:08,700+0000 [Kafka Source Thread] ERROR org.apache.pulsar.io.kafka.KafkaAbstractSource - [Kafka Source Thread] Error while consuming records org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group. at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:1139) ~[kafka-clients-2.7.2.jar:?] at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:1004) ~[kafka-clients-2.7.2.jar:?] at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1389) ~[kafka-clients-2.7.2.jar:?] at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1343) ~[kafka-clients-2.7.2.jar:?] at org.apache.pulsar.io.kafka.KafkaAbstractSource.lambda$start$0(KafkaAbstractSource.java:174) ~[juUFGcDeAgkU9W0VTrIV-g/:?] at java.lang.Thread.run(Thread.java:829) ~[?:?]

[hermotech]

I work together with @relief-melone.
I was tasked with testing the fix to see if this solves our problem.

Here is what I did:

• I cloned this repo and verified that the code changes for the fix is visible in the source code on my local repo
• I then compiled the required nar files and replaced the previous nar files on the docker image.
• Then I deployed a pod in our K8s cluster with the new Docker image that contains the newly compiled nar files (kafka source) and let it run for a couple of days.
• I checked this morning and the pod's logs was again showing the familiar error without having crashed and restarted as we hoped it would.
• I further tested to check if it could consume messages from Kafka in this state, but it couldn't, meaning that the source is still in a state where it cannot recover and is basically halted without consuming any messages.

Is there any other way I can test this?
Is there something I could have done wrong during compilation / testing?
Any suggestions to what I can try / test differently?

Here is the excerpt from the log for the source with the fix:
What is interesting is that it shows "Stopping Kafka Source", but it does not cause the pod to crash.

2023-06-11T22:20:29,237+0000 [Kafka Source Thread] ERROR org.apache.pulsar.io.kafka.KafkaAbstractSource - [Kafka Source Thread] Error while consuming records
org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that 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 message processing. 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.
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:1441) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:1341) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1260) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1235) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:206) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:169) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:129) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:617) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:427) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:312) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:214) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:1174) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1502) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1401) ~[kafka-clients-3.4.0.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1358) ~[kafka-clients-3.4.0.jar:?]
at org.apache.pulsar.io.kafka.KafkaAbstractSource.lambda$start$0(KafkaAbstractSource.java:181) ~[QwaU9xgIUZTeaeOtqJikUw/:?]
at java.lang.Thread.run(Thread.java:833) ~[?:?]
2023-06-11T22:20:29,238+0000 [Kafka Source Thread] INFO org.apache.pulsar.io.kafka.KafkaAbstractSource - Stopping kafka source

[hermotech]

Looking at the code here, the log entry "Kafka source stopped" is missing from the above excerpt.

https://github.com/apache/pulsar/blob/master/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java#L144-L157

Could it be that something goes wrong in the close() method and the source isn't properly stopped?

[RobertIndie]

@htbmw Thanks for reporting the issue. This PR could fix it: #20698

[poorbarcode]

@htbmw After we test the fix above, it does not work. And @shibd submitted 3 PRs to do the final fix

• [fix][io][branch-2.10] Not restart instance when kafka source poll exception. #20816
• [fix][io][branch-2.11] Not restart instance when kafka source poll ex… #20817
• [fix][io][branch-2.11] Not restart instance when kafka source poll ex… #20817

And there is a PIP to do the fix for the master branch

• [pip][design] PIP-281: Add notifyError method on PushSource #20807

[poorbarcode]

@htbmw

Could you provide a thread dump?

[hermotech]

If you could please tell me how to generate the thread dump, I can provide it.
At the moment I just run this in a kubernetes pod and will have to wait for the issue to reproduce it because I cannot manually force this kind of error.
Is there a java environment variable I can set on the pod to force a thread dump perhaps?
Perhaps attach to the pod and run a java command to force a threaddump?

[poorbarcode]

I am sorry. Maybe I do not clear it. it is jstack. You can get it by this flow

1. login into the pod
2. call ps -ef|grep java or jps to get the pid of the connector progress
3. call jstack -l <pid>, then you will get the thread's stack.

[mattisonchao]

2023-06-11T22:20:29,238+0000 [Kafka Source Thread] INFO org.apache.pulsar.io.kafka.KafkaAbstractSource - Stopping kafka source

Kafka Source Thread is trying to join it self.

[hermotech]

Thanks for the contributions here.