Skip to content

Commit

Permalink
[fix][io] Close the kafka source connector if there is uncaught excep…
Browse files Browse the repository at this point in the history
…tion (#20424)

Fixes #19880

### Motivation

If any unexpected and uncaught exceptions are thrown in the `runnerThread` of the Kafka source connector, the connector will just log that message and get stuck. The connector won't exit but won't do anything either.

### Modifications

* Close the connector if there is uncaught exception

Signed-off-by: Zike Yang <[email protected]>
(cherry picked from commit 6079a91)
  • Loading branch information
RobertIndie committed Jun 2, 2023
1 parent c776e89 commit a125e15
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,14 @@ public void start() {
}
});
runnerThread.setUncaughtExceptionHandler(
(t, e) -> LOG.error("[{}] Error while consuming records", t.getName(), e));
(t, e) -> {
LOG.error("[{}] Error while consuming records", t.getName(), e);
try {
this.close();
} catch (InterruptedException ex) {
// The interrupted exception is thrown by the runnerThread itself. Ignore it.
}
});
runnerThread.setName("Kafka Source Thread");
runnerThread.start();
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,18 @@


import com.google.common.collect.ImmutableMap;
import java.util.Collection;
import java.util.Collections;
import java.lang.reflect.Field;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.io.core.SourceContext;
import org.apache.pulsar.io.kafka.KafkaAbstractSource;
import org.apache.pulsar.io.kafka.KafkaSourceConfig;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.Test;

Expand Down Expand Up @@ -153,6 +157,25 @@ public final void loadFromSaslYamlFileTest() throws IOException {
assertEquals(config.getSslTruststorePassword(), "cert_pwd");
}

@Test
public final void closeConnectorWhenUnexpectedExceptionThrownTest() throws Exception {
KafkaAbstractSource source = new DummySource();
Consumer consumer = mock(Consumer.class);
Mockito.doThrow(new RuntimeException("Uncaught exception")).when(consumer)
.subscribe(Mockito.any(Collection.class));

Field consumerField = KafkaAbstractSource.class.getDeclaredField("consumer");
consumerField.setAccessible(true);
consumerField.set(source, consumer);

source.start();

Field runningField = KafkaAbstractSource.class.getDeclaredField("running");
runningField.setAccessible(true);

Assert.assertFalse((boolean) runningField.get(source));
}

private File getFile(String name) {
ClassLoader classLoader = getClass().getClassLoader();
return new File(classLoader.getResource(name).getFile());
Expand Down

0 comments on commit a125e15

Please sign in to comment.