Add backend errors among retriable for Postgres connector

Step1: Upgrade to version 1.7 Step2: create a postgres debezium source connector and start stream data Step3: Wait for a while, problem happen Check log below 2022-01-04 09:32:07,648] INFO WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask) [2022-01-04 09:32:07,648] INFO WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [2022-01-04 09:32:07,650] INFO WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} Finished commitOffsets successfully in 2 ms (org.apache.kafka.connect.runtime.WorkerSourceTask) [2022-01-04 09:32:13,527] ERROR Failed to read column metadata for 'public.order_part2' (io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder) [2022-01-04 09:32:13,527] ERROR Producer failure (io.debezium.pipeline.ErrorHandler) org.postgresql.util.PSQLException: An I/O error occurred while sending to the backend. at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:349) at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:481) at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:401) at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:322) at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:308) at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:284) at org.postgresql.jdbc.PgStatement.executeQuery(PgStatement.java:236) at org.postgresql.jdbc.PgDatabaseMetaData.getColumns(PgDatabaseMetaData.java:1579) at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.getTableColumnsFromDatabase(PgOutputMessageDecoder.java:333) at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.handleRelationMessage(PgOutputMessageDecoder.java:272) at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.processNotEmptyMessage(PgOutputMessageDecoder.java:176) at io.debezium.connector.postgresql.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:33) at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:493) at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:485) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.processMessages(PostgresStreamingChangeEventSource.java:205) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:167) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:40) at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:166) at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:127) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) 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:834) Caused by: java.net.SocketException: Broken pipe (Write failed) at java.base/java.net.SocketOutputStream.socketWrite0(Native Method) at java.base/java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:110) at java.base/java.net.SocketOutputStream.write(SocketOutputStream.java:150) at java.base/java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:81) at java.base/java.io.BufferedOutputStream.flush(BufferedOutputStream.java:142) at org.postgresql.core.PGStream.flush(PGStream.java:665) at org.postgresql.core.v3.QueryExecutorImpl.sendSync(QueryExecutorImpl.java:1469) at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:320) ... 23 more [2022-01-04 09:32:13,529] INFO Connection gracefully closed (io.debezium.jdbc.JdbcConnection) [2022-01-04 09:32:13,532] INFO Connection gracefully closed (io.debezium.jdbc.JdbcConnection) [2022-01-04 09:32:13,532] INFO Finished streaming (io.debezium.pipeline.ChangeEventSourceCoordinator) [2022-01-04 09:32:13,533] INFO Connected metrics set to 'false' (io.debezium.pipeline.metrics.StreamingChangeEventSourceMetrics) [2022-01-04 09:32:13,743] INFO WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask) [2022-01-04 09:32:13,743] INFO WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [2022-01-04 09:32:13,743] ERROR WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask) org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped. at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:170) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:40) at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:166) at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:127) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) 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:834) Caused by: org.postgresql.util.PSQLException: An I/O error occurred while sending to the backend. at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:349) at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:481) at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:401) at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:322) at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:308) at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:284) at org.postgresql.jdbc.PgStatement.executeQuery(PgStatement.java:236) at org.postgresql.jdbc.PgDatabaseMetaData.getColumns(PgDatabaseMetaData.java:1579) Caused by: java.net.SocketException: Broken pipe (Write failed) at java.base/java.net.SocketOutputStream.socketWrite0(Native Method) at java.base/java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:110) at java.base/java.net.SocketOutputStream.write(SocketOutputStream.java:150) at java.base/java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:81) at java.base/java.io.BufferedOutputStream.flush(BufferedOutputStream.java:142) at org.postgresql.core.PGStream.flush(PGStream.java:665) at org.postgresql.core.v3.QueryExecutorImpl.sendSync(QueryExecutorImpl.java:1469) at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:320) ... 23 more [2022-01-04 09:32:13,744] ERROR WorkerSourceTask\{id=source.postgresql_debezium.s14e_order.v1-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [2022-01-04 09:32:13,744] INFO Stopping down connector (io.debezium.connector.common.BaseSourceTask) [2022-01-04 09:32:13,747] INFO Connection gracefully closed (io.debezium.jdbc.JdbcConnection) [2022-01-04 09:32:13,748] INFO Connection gracefully closed (io.debezium.jdbc.JdbcConnection) [2022-01-04 09:32:13,748] INFO [Producer clientId=connector-producer-source.postgresql_debezium.s14e_order.v1-0] Closing the Kafka producer with timeoutMillis = 30000 ms. (org.apache.kafka.clients.producer.KafkaProducer) [2022-01-04 09:32:13,750] INFO Publish thread interrupted for client_id=connector-producer-source.postgresql_debezium.s14e_order.v1-0 client_type=PRODUCER session= cluster=4-QDyBAsQemtjp7MZCIcJA (io.confluent.monitoring.clients.interceptor.MonitoringInterceptor) [2022-01-04 09:32:13,750] INFO Publishing Monitoring Metrics stopped for client_id=connector-producer-source.postgresql_debezium.s14e_order.v1-0 client_type=PRODUCER session= cluster=4-QDyBAsQemtjp7MZCIcJA (io.confluent.monitoring.clients.interceptor.MonitoringInterceptor) [2022-01-04 09:32:13,750] INFO [Producer clientId=confluent.monitoring.interceptor.connector-producer-source.postgresql_debezium.s14e_order.v1-0] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. (org.apache.kafka.clients.producer.KafkaProducer)