Copilot commented on code in PR #71:
URL: https://github.com/apache/pulsar-connectors/pull/71#discussion_r3554635108
##########
debezium/mysql/src/test/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSourceTest.java:
##########
@@ -49,15 +52,22 @@ public class DebeziumMysqlSourceTest {
private static final String PULSAR_IMAGE =
System.getenv().getOrDefault("PULSAR_TEST_IMAGE",
"apachepulsar/pulsar:4.1.3");
+ /** Rows seeded before open(), each of which the initial snapshot must
emit. */
+ private static final int EXPECTED_RECORDS = 2;
+
+ private static final int READ_TIMEOUT_SECONDS = 120;
+
private static final int MYSQL_PORT = 3306;
private GenericContainer<?> mysqlContainer;
private PulsarContainer pulsarContainer;
private PulsarClient pulsarClient;
private DebeziumMysqlSource source;
+ private ExecutorService readerExecutor;
@BeforeMethod
public void setup() throws Exception {
+ readerExecutor = Executors.newSingleThreadExecutor();
Review Comment:
The reader executor uses a non-daemon thread. If source.read() does not
respond to interruption (AbstractKafkaConnectSource.read() loops without
checking interrupts), shutdownNow()/Future.cancel(true) may not stop the task
and this non-daemon thread can keep the JVM alive, reintroducing the CI hang
you’re trying to prevent.
##########
debezium/postgres/src/test/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSourceTest.java:
##########
@@ -48,13 +51,20 @@ public class DebeziumPostgresSourceTest {
private static final String PULSAR_IMAGE =
System.getenv().getOrDefault("PULSAR_TEST_IMAGE",
"apachepulsar/pulsar:4.1.3");
+ /** Rows seeded before open(), each of which the initial snapshot must
emit. */
+ private static final int EXPECTED_RECORDS = 2;
+
+ private static final int READ_TIMEOUT_SECONDS = 120;
+
private PostgreSQLContainer<?> postgresContainer;
private PulsarContainer pulsarContainer;
private PulsarClient pulsarClient;
private DebeziumPostgresSource source;
+ private ExecutorService readerExecutor;
@BeforeMethod
public void setup() throws Exception {
+ readerExecutor = Executors.newSingleThreadExecutor();
Review Comment:
The reader executor uses a non-daemon thread. If source.read() does not
respond to interruption (AbstractKafkaConnectSource.read() loops without
checking interrupts), shutdownNow()/Future.cancel(true) may not stop the task
and this non-daemon thread can keep the JVM alive, reintroducing the CI hang
you’re trying to prevent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
