This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new df044275 [fix][io] Interrupt Kafka Source instance thread on consumer
fatal error to prevent deadlock (#88)
df044275 is described below
commit df0442759c1f1dabaf008929c23b1a0b577e73b0
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Fri Jul 10 08:30:12 2026 -0700
[fix][io] Interrupt Kafka Source instance thread on consumer fatal error to
prevent deadlock (#88)
* [fix][io] Interrupt Kafka Source instance thread on consumer fatal error
to prevent deadlock
* build: Add Testcontainers dependency to kafka module for liveness test
* refactor: revert consumer.subscribe() back to main thread and check
inside the catch block
* updated Testcontainers to wait up to 5 minutes for it to boot, so it
survives the GitHub Actions CI
* refactor: the VM runs OOM,so completely eliminate the Testcontainers
entire Kafka broker and zookeeper
* [fix][test] Make the Kafka deadlock test deterministic
Address the review findings on KafkaSourceDeadlockTest:
- poll() blocked on Thread.sleep(5000). It now waits on a latch, so it
leaves only by being interrupted, with no wall-clock delay and no race
between the sleep and the interrupt.
- The simulator thread slept for a second, scanned every JVM thread for
one named 'Kafka Source Thread', and swallowed any exception. The
consumer thread now records itself when it enters poll(), and the test
interrupts it directly once a latch confirms it is blocked there. The
simulator thread is gone.
- Assert.fail() ran before source.close(), leaving the runner thread
alive and able to disturb later tests. The blocking section is now in
a try/finally so close() always runs.
Every wait either completes or fails the test with a message, so a
regression surfaces as a clear failure rather than a hang. Runtime drops
from ~15s of sleeping to ~1s.
---------
Co-authored-by: Praveenkumar76 <[email protected]>
---
.../pulsar/io/kafka/KafkaAbstractSource.java | 47 ++++++++-
.../pulsar/io/kafka/KafkaSourceDeadlockTest.java | 107 +++++++++++++++++++++
2 files changed, 152 insertions(+), 2 deletions(-)
diff --git
a/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
b/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
index 7eba7438..0452cd53 100644
--- a/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
+++ b/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.io.kafka;
+import com.google.common.annotations.VisibleForTesting;
import io.jsonwebtoken.io.Encoders;
import java.time.Duration;
import java.util.Collections;
@@ -65,9 +66,11 @@ public abstract class KafkaAbstractSource<V> extends
PushSource<V> {
private KafkaSourceConfig kafkaSourceConfig;
private Thread runnerThread;
private long maxPollIntervalMs;
+ private volatile Thread instanceThread;
@Override
public void open(Map<String, Object> config, SourceContext sourceContext)
throws Exception {
+ this.instanceThread = Thread.currentThread();
kafkaSourceConfig = KafkaSourceConfig.load(config, sourceContext);
Objects.requireNonNull(kafkaSourceConfig.getTopic(), "Kafka topic is
not set");
Objects.requireNonNull(kafkaSourceConfig.getBootstrapServers(), "Kafka
bootstrapServers is not set");
@@ -163,12 +166,31 @@ public abstract class KafkaAbstractSource<V> extends
PushSource<V> {
LOG.info("Kafka source stopped.");
}
+ @VisibleForTesting
+ void setInstanceThread(Thread instanceThread) {
+ this.instanceThread = instanceThread;
+ }
+
+ @VisibleForTesting
+ void setKafkaSourceConfig(KafkaSourceConfig kafkaSourceConfig) {
+ this.kafkaSourceConfig = kafkaSourceConfig;
+ }
+
+ @VisibleForTesting
+ void setConsumer(Consumer<Object, Object> consumer) {
+ this.consumer = consumer;
+ }
+
@SuppressWarnings("unchecked")
public void start() {
LOG.info("Starting subscribe kafka source on {}",
kafkaSourceConfig.getTopic());
+
+ // 1. REVERT: Keep subscribe on the main thread so initialization
errors fail synchronously
consumer.subscribe(Collections.singletonList(kafkaSourceConfig.getTopic()));
+
runnerThread = new Thread(() -> {
LOG.info("Kafka source started.");
+
while (running) {
try {
ConsumerRecords<Object, Object> consumerRecords =
consumer.poll(Duration.ofSeconds(1L));
@@ -184,20 +206,41 @@ public abstract class KafkaAbstractSource<V> extends
PushSource<V> {
index++;
}
if (!kafkaSourceConfig.isAutoCommitEnabled()) {
- // Wait about 2/3 of the time of maxPollIntervalMs.
- // so as to avoid waiting for the timeout to be kicked
out of the consumer group.
CompletableFuture.allOf(futures).get(maxPollIntervalMs
* 2 / 3, TimeUnit.MILLISECONDS);
consumer.commitSync();
}
} catch (Exception e) {
+ if (!running) {
+ LOG.info("Kafka source is shutting down gracefully.
Ignoring interrupt.");
+ break;
+ }
+
LOG.error("Error while processing records", e);
notifyError(e);
+
+ // Fire the flare to break the I/O deadlock
+ if (instanceThread != null && instanceThread.isAlive()) {
+ LOG.warn("Interrupting the Instance Thread to break
I/O deadlock.");
+ instanceThread.interrupt();
+ }
break;
}
}
});
running = true;
runnerThread.setName("Kafka Source Thread");
+
+ // Update the UncaughtExceptionHandler
+ runnerThread.setUncaughtExceptionHandler((t, e) -> {
+ LOG.error("[{}] Uncaught error while consuming records",
t.getName(), e);
+ notifyError(new RuntimeException(e));
+
+ if (running && instanceThread != null && instanceThread.isAlive())
{
+ LOG.warn("Interrupting the Instance Thread due to uncaught
consumer thread exception.");
+ instanceThread.interrupt();
+ }
+ });
+
runnerThread.start();
}
diff --git
a/kafka/src/test/java/org/apache/pulsar/io/kafka/KafkaSourceDeadlockTest.java
b/kafka/src/test/java/org/apache/pulsar/io/kafka/KafkaSourceDeadlockTest.java
new file mode 100644
index 00000000..7cd11ac6
--- /dev/null
+++
b/kafka/src/test/java/org/apache/pulsar/io/kafka/KafkaSourceDeadlockTest.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.kafka;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * When the Kafka consumer thread hits a fatal error it must interrupt the
instance thread,
+ * which is otherwise blocked on Pulsar network I/O and would deadlock.
+ *
+ * <p>The test plays the part of the instance thread. Coordination is by latch
rather than by
+ * sleeping: every wait either completes or fails the test, so a regression
surfaces as a
+ * timeout with a clear message instead of a hang.
+ */
+public class KafkaSourceDeadlockTest {
+
+ private static final long TIMEOUT_SECONDS = 10;
+
+ @Test(timeOut = 60_000)
+ public void testConnectorBreaksInstanceThreadDeadlock() throws Exception {
+ KafkaBytesSource source = new KafkaBytesSource();
+
+ // Inject dependencies using package-private @VisibleForTesting methods
+ source.setInstanceThread(Thread.currentThread());
+
+ KafkaSourceConfig config = Mockito.mock(KafkaSourceConfig.class);
+ Mockito.when(config.getTopic()).thenReturn("test-topic");
+ Mockito.when(config.isAutoCommitEnabled()).thenReturn(false);
+ source.setKafkaSourceConfig(config);
+
+ // Signals the consumer thread has entered poll() and is blocked there.
+ final CountDownLatch pollEntered = new CountDownLatch(1);
+ // Never counted down: poll() leaves this wait only by being
interrupted.
+ final CountDownLatch pollReleased = new CountDownLatch(1);
+ // The consumer thread identifies itself, so the test need not scan
JVM threads by name.
+ final AtomicReference<Thread> consumerThread = new AtomicReference<>();
+
+ @SuppressWarnings("unchecked")
+ Consumer<Object, Object> mockConsumer = Mockito.mock(Consumer.class);
+ source.setConsumer(mockConsumer);
+
+
Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))).thenAnswer(invocation
-> {
+ consumerThread.set(Thread.currentThread());
+ pollEntered.countDown();
+ try {
+ // Interruptible, unlike Thread.sleep with a guessed duration.
+ pollReleased.await();
+ } catch (InterruptedException e) {
+ // Stand in for a fatal Kafka error surfacing on the consumer
thread.
+ throw new RuntimeException("Simulated fatal Kafka network
error", e);
+ }
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+
+ boolean wasInterrupted = false;
+ try {
+ source.start();
+
+ Assert.assertTrue(pollEntered.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS),
+ "the consumer thread never reached poll()");
+
+ // Fail the consumer thread. Its fatal-error handling must
interrupt this thread.
+ consumerThread.get().interrupt();
+
+ try {
+ // Stands in for the instance thread blocking on Pulsar
network I/O. The interrupt
+ // may already have arrived, in which case this throws
immediately.
+ Thread.sleep(TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ Assert.fail("the instance thread remained deadlocked and was
never interrupted");
+ } catch (InterruptedException e) {
+ wasInterrupted = true;
+ }
+ } finally {
+ // Clear the interrupt flag so close() runs cleanly, and close
even if we failed above.
+ Thread.interrupted();
+ source.close();
+ }
+
+ Assert.assertTrue(wasInterrupted,
+ "the failing consumer thread should have interrupted the
instance thread");
+ }
+}
