bug: end-to-end source test failed: Interrupted while stopping coordinator

[fuyufjh]

Describe the bug

https://buildkite.com/risingwavelabs/pull-request/builds/44654#018e4f81-15e2-4e5c-8fe6-6d292d9304ee

Note that there are 2 logs in the next section. The 2nd seems to be caused by the 1st one, and eventually caused the CN node to exited

Error message/log

2024-03-18T03:13:18.52456753Z ERROR risingwave_connector_node: Interrupted while stopping coordinator: java.lang.InterruptedException
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2109)
	at java.base/java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1454)
	at java.base/java.util.concurrent.Executors$DelegatedExecutorService.awaitTermination(Executors.java:709)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.stop(ChangeEventSourceCoordinator.java:308)
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:289)
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:273)
	at io.debezium.embedded.EmbeddedEngine.stopTaskAndCommitOffset(EmbeddedEngine.java:1047)
	at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759)
	at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:229)
	at com.risingwave.connector.source.core.DbzCdcEngine.run(DbzCdcEngine.java:64)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
 thread="rw-dbz-engine-runner-2" class="io.debezium.connector.common.BaseSourceTask"
2024-03-18T03:13:18.525471586Z ERROR risingwave_connector_node: engine#2 terminated with error. message: Error while trying to stop the task and commit the offsets: org.apache.kafka.connect.errors.ConnectException: Interrupted while stopping coordinator, failing the task
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:296)
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:273)
	at io.debezium.embedded.EmbeddedEngine.stopTaskAndCommitOffset(EmbeddedEngine.java:1047)
	at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759)
	at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:229)
	at com.risingwave.connector.source.core.DbzCdcEngine.run(DbzCdcEngine.java:64)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
 thread="rw-dbz-engine-runner-2" class="com.risingwave.connector.source.core.DbzCdcEngineRunner"
Mon Mar 18 03:18:18 UTC 2024 [risedev]: Program exited with 139

To Reproduce

No response

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[StrikeW]

2024-03-18T03:13:18.521158054Z  INFO risingwave_connector_node: Engine#2: JNI sender broken detected, stop the engine thread="Thread-1" class="com.risingwave.connector.source.core.JniDbzSourceHandler"
2024-03-18T03:13:18.521262817Z  INFO risingwave_connector_node: Stopping the embedded engine thread="Thread-1" class="io.debezium.embedded.EmbeddedEngine"
2024-03-18T03:13:18.521379047Z  INFO risingwave_connector_node: Waiting for PT0.001S for connector to stop thread="Thread-1" class="io.debezium.embedded.EmbeddedEngine"
2024-03-18T03:13:18.522470024Z  INFO risingwave_connector_node: Stopping the task and engine thread="rw-dbz-engine-runner-2" class="io.debezium.embedded.EmbeddedEngine"
2024-03-18T03:13:18.522535726Z  INFO risingwave_connector_node: Stopping down connector thread="rw-dbz-engine-runner-2" class="io.debezium.connector.common.BaseSourceTask"
2024-03-18T03:13:18.522678066Z  INFO risingwave_connector_node: engine#2 terminated thread="Thread-1" class="com.risingwave.connector.source.core.DbzCdcEngineRunner"
2024-03-18T03:13:18.522723105Z  INFO risingwave_connector::source::cdc::source::reader: end of jni call runJniDbzSourceThread source_id=2
2024-03-18T03:13:18.52456753Z ERROR risingwave_connector_node: Interrupted while stopping coordinator: java.lang.InterruptedException
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2109)
	at java.base/java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1454)
	at java.base/java.util.concurrent.Executors$DelegatedExecutorService.awaitTermination(Executors.java:709)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.stop(ChangeEventSourceCoordinator.java:308)
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:289)
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:273)
	at io.debezium.embedded.EmbeddedEngine.stopTaskAndCommitOffset(EmbeddedEngine.java:1047)
	at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759)
	at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:229)
	at com.risingwave.connector.source.core.DbzCdcEngine.run(DbzCdcEngine.java:64)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
 thread="rw-dbz-engine-runner-2" class="io.debezium.connector.common.BaseSourceTask"
2024-03-18T03:13:18.525471586Z ERROR risingwave_connector_node: engine#2 terminated with error. message: Error while trying to stop the task and commit the offsets: org.apache.kafka.connect.errors.ConnectException: Interrupted while stopping coordinator, failing the task
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:296)
	at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:273)
	at io.debezium.embedded.EmbeddedEngine.stopTaskAndCommitOffset(EmbeddedEngine.java:1047)
	at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759)
	at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:229)
	at com.risingwave.connector.source.core.DbzCdcEngine.run(DbzCdcEngine.java:64)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
 thread="rw-dbz-engine-runner-2" class="com.risingwave.connector.source.core.DbzCdcEngineRunner"

These logs are expected when the connector is being drop.

I remembered the ALTER command will drop the old stream job, from the log of CN, the cdc connector indeed detected the channel is broken and it will wait a while for the exit of connector (Waiting for PT0.001S for connector to stop). In the mean time, we will call executor.shutdownNow() to interrupt the connector thread.

We may find ways to eliminate these logs.