Fix forceful shutdown of change event source coordinator [DBZ-9154]

Migrated from [DBZ-9154](https://issues.redhat.com/browse/DBZ-9154)

As reported [here|https://github.com/debezium/debezium/pull/6501#issuecomment-2975246312], there is a problem with forceful shutdown of change event source coordinator. This ticket is about exploring and fixing the same.

Some important log lines, in case the logs are erased

{code:java}
2025-06-13 20:28:15,362 ERROR Informix_Server||snapshot Error during snapshot [io.debezium.relational.RelationalSnapshotChangeEventSource]
java.util.concurrent.ExecutionException: org.apache.kafka.connect.errors.ConnectException: Snapshotting of table testdb.informix.b failed
at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191)
at io.debezium.relational.RelationalSnapshotChangeEventSource.createDataEvents(RelationalSnapshotChangeEventSource.java:531)
at io.debezium.relational.RelationalSnapshotChangeEventSource.doExecute(RelationalSnapshotChangeEventSource.java:179)
at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:96)
at io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:299)
at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$doBlockingSnapshot$4(ChangeEventSourceCoordinator.java:256)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: org.apache.kafka.connect.errors.ConnectException: Snapshotting of table testdb.informix.b failed
at io.debezium.relational.RelationalSnapshotChangeEventSource.lambda$createDataEventsForTableCallable$6(RelationalSnapshotChangeEventSource.java:590)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
... 5 common frames omitted
Caused by: java.sql.SQLSyntaxErrorException: A syntax error has occurred.
at com.informix.util.IfxErrMsg.buildExceptionWithMessage(IfxErrMsg.java:417)
at com.informix.util.IfxErrMsg.buildIsamException(IfxErrMsg.java:404)
at com.informix.jdbc.IfxSqli.addException(IfxSqli.java:3023)
at com.informix.jdbc.IfxSqli.receiveError(IfxSqli.java:3274)
at com.informix.jdbc.IfxSqli.dispatchMsg(IfxSqli.java:2269)
at com.informix.jdbc.IfxSqli.receiveMessage(IfxSqli.java:2194)
at com.informix.jdbc.IfxSqli.executeStatementQuery(IfxSqli.java:1442)
at com.informix.jdbc.IfxResultSet.executeQuery(IfxResultSet.java:183)
at com.informix.jdbc.IfxStatement.executeQueryImpl(IfxStatement.java:909)
at com.informix.jdbc.IfxStatement.executeQuery(IfxStatement.java:169)
at io.debezium.connector.informix.InformixSnapshotChangeEventSource.resultSetForDataEvents(InformixSnapshotChangeEventSource.java:239)
at io.debezium.relational.RelationalSnapshotChangeEventSource.doCreateDataEventsForTable(RelationalSnapshotChangeEventSource.java:622)
at io.debezium.relational.RelationalSnapshotChangeEventSource.lambda$createDataEventsForTableCallable$6(RelationalSnapshotChangeEventSource.java:583)
... 6 common frames omitted
2025-06-13 20:28:15,363 INFO Informix_Server||snapshot Snapshot - Final stage [io.debezium.pipeline.source.AbstractSnapshotChangeEventSource]
2025-06-13 20:28:15,363 WARN Informix_Server||snapshot Snapshot was not completed successfully [io.debezium.pipeline.source.AbstractSnapshotChangeEventSource]
2025-06-13 20:28:15,365 WARN Informix_Server||snapshot Error while executing requested blocking snapshot. [io.debezium.pipeline.ChangeEventSourceCoordinator]
io.debezium.DebeziumException: java.util.concurrent.ExecutionException: org.apache.kafka.connect.errors.ConnectException: Snapshotting of table testdb.informix.b failed
at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:105)
at io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:299)
at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$doBlockingSnapshot$4(ChangeEventSourceCoordinator.java:256)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.connect.errors.ConnectException: Snapshotting of table testdb.informix.b failed
at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191)
at io.debezium.relational.RelationalSnapshotChangeEventSource.createDataEvents(RelationalSnapshotChangeEventSource.java:531)
at io.debezium.relational.RelationalSnapshotChangeEventSource.doExecute(RelationalSnapshotChangeEventSource.java:179)
at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:96)
... 7 common frames omitted
Caused by: org.apache.kafka.connect.errors.ConnectException: Snapshotting of table testdb.informix.b failed
at io.debezium.relational.RelationalSnapshotChangeEventSource.lambda$createDataEventsForTableCallable$6(RelationalSnapshotChangeEventSource.java:590)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
... 5 common frames omitted
Caused by: java.sql.SQLSyntaxErrorException: A syntax error has occurred.
at com.informix.util.IfxErrMsg.buildExceptionWithMessage(IfxErrMsg.java:417)
at com.informix.util.IfxErrMsg.buildIsamException(IfxErrMsg.java:404)
at com.informix.jdbc.IfxSqli.addException(IfxSqli.java:3023)
at com.informix.jdbc.IfxSqli.receiveError(IfxSqli.java:3274)
at com.informix.jdbc.IfxSqli.dispatchMsg(IfxSqli.java:2269)
at com.informix.jdbc.IfxSqli.receiveMessage(IfxSqli.java:2194)
at com.informix.jdbc.IfxSqli.executeStatementQuery(IfxSqli.java:1442)
at com.informix.jdbc.IfxResultSet.executeQuery(IfxResultSet.java:183)
at com.informix.jdbc.IfxStatement.executeQueryImpl(IfxStatement.java:909)
at com.informix.jdbc.IfxStatement.executeQuery(IfxStatement.java:169)
at io.debezium.connector.informix.InformixSnapshotChangeEventSource.resultSetForDataEvents(InformixSnapshotChangeEventSource.java:239)
at io.debezium.relational.RelationalSnapshotChangeEventSource.doCreateDataEventsForTable(RelationalSnapshotChangeEventSource.java:622)
at io.debezium.relational.RelationalSnapshotChangeEventSource.lambda$createDataEventsForTableCallable$6(RelationalSnapshotChangeEventSource.java:583)
... 6 common frames omitted
2025-06-13 20:28:15,365 INFO Informix_Server||streaming Streaming resumed [io.debezium.connector.informix.InformixStreamingChangeEventSource]
2025-06-13 20:28:15,973 INFO BlockingSnapshotIT||test Requested thread factory for component JdbcConnection, id = JdbcConnection named = jdbc-connection-close [io.debezium.util.Threads]
2025-06-13 20:28:15,973 INFO BlockingSnapshotIT||test Creating thread debezium-jdbcconnection-JdbcConnection-jdbc-connection-close [io.debezium.util.Threads]
2025-06-13 20:28:15,974 INFO || Connection gracefully closed [io.debezium.jdbc.JdbcConnection]
2025-06-13 20:28:16,089 INFO || 2002 records sent during previous 00:00:05.121, last recorded offset of {databaseName=testdb} partition is {begin_lsn=17193279512, commit_lsn=17193559492, change_lsn=17193279568} [io.debezium.connector.common.BaseSourceTask]
2025-06-13 20:28:16,716 INFO BlockingSnapshotIT||test Stopping the connector [io.debezium.connector.informix.BlockingSnapshotIT]
2025-06-13 20:28:16,716 INFO BlockingSnapshotIT||test Stopping the engine [io.debezium.connector.informix.BlockingSnapshotIT]
2025-06-13 20:28:16,717 INFO BlockingSnapshotIT||test Engine state has changed from 'POLLING_TASKS' to 'STOPPING' [io.debezium.embedded.async.AsyncEmbeddedEngine]
2025-06-13 20:28:16,719 INFO || Stopping down connector [io.debezium.connector.common.BaseSourceTask]
2025-06-13 20:28:16,719 INFO BlockingSnapshotIT||engine Task interrupted while polling. [io.debezium.embedded.async.AsyncEmbeddedEngine]
2025-06-13 20:28:20,719 WARN || Coordinator didn't stop in the expected time, shutting down executor now [io.debezium.pipeline.ChangeEventSourceCoordinator]
2025-06-13 20:28:24,719 WARN BlockingSnapshotIT||test Stopping of the tasks was interrupted, shutting down immediately. [io.debezium.embedded.async.AsyncEmbeddedEngine]
2025-06-13 20:28:24,720 ERROR || Interrupted while stopping coordinator [io.debezium.connector.common.BaseSourceTask]
java.lang.InterruptedException: null
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:1770)
at java.base/java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1475)
at java.base/java.util.concurrent.Executors$DelegatedExecutorService.awaitTermination(Executors.java:780)
at io.debezium.pipeline.ChangeEventSourceCoordinator.stop(ChangeEventSourceCoordinator.java:384)
at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:478)
at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:461)
at io.debezium.embedded.async.AsyncEmbeddedEngine.lambda$stopSourceTasks$3(AsyncEmbeddedEngine.java:635)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
2025-06-13 20:28:24,720 INFO BlockingSnapshotIT||engine Engine is stopped. [io.debezium.embedded.async.AsyncEmbeddedEngine]
2025-06-13 20:28:24,720 INFO BlockingSnapshotIT||engine Engine state has changed from 'STOPPING' to 'STOPPED' [io.debezium.embedded.async.AsyncEmbeddedEngine]
2025-06-13 20:28:24,720 INFO BlockingSnapshotIT||engine Connector 'io.debezium.connector.informix.InformixConnector' completed normally. [io.debezium.connector.informix.BlockingSnapshotIT]
2025-06-13 20:28:24,821 INFO BlockingSnapshotIT||test Interrupting the engine [io.debezium.connector.informix.BlockingSnapshotIT]
20:28:25.030 IFX >>>2025-06-13 20:28:25.030 CDC: Log Reader ended. CDC session id 26214439. SQL session id 50.
2025-06-13 20:28:25,033 INFO Informix_Server||streaming Finished streaming [io.debezium.pipeline.ChangeEventSourceCoordinator]
2025-06-13 20:28:25,033 INFO Informix_Server||streaming Connected metrics set to 'false' [io.debezium.pipeline.ChangeEventSourceCoordinator]
2025-06-13 20:28:25,036 INFO BlockingSnapshotIT||test Stopping the connector [io.debezium.connector.informix.BlockingSnapshotIT]
2025-06-13 20:28:25,038 ERROR BlockingSnapshotIT||test Ignored failure for anErrorDuringBlockingSnapshotShouldLeaveTheConnectorInAGoodState(io.debezium.connector.informix.BlockingSnapshotIT), tracked with https://issues.redhat.com/browse/DBZ-8114 [io.debezium.junit.Flaky]
{code}