Iceberg 1.7.0 java.lang.IllegalStateException: Connection pool shut down

[lawrencemq]

Apache Iceberg version

1.7.0 (latest release)

Query engine

Kafka Connect

Please describe the bug 🐞

I've been using the Tabular connector for some time and have transitioned over to the Apache connector.

Using version 1.7.0 and a Hive catalog. A Kafka Connect deployment with 3 nodes, 6 tasks, committing every 5 minutes.

Seeing new failures, happening every 10 minutes (give or take a few), where tasks will fail with the following stacktrace.
I do sometimes see Iceberg-commit messages in the log before; sometimes not.

I thought it might be similar to #11089 but the Iceberg version and catalog are different.

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:632)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:350)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:250)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:219)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:204)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:237)
	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.lang.IllegalStateException: Connection pool shut down
	at org.apache.http.util.Asserts.check(Asserts.java:34)
	at org.apache.http.impl.conn.PoolingHttpClientConnectionManager.requestConnection(PoolingHttpClientConnectionManager.java:269)
	at software.amazon.awssdk.http.apache.internal.conn.ClientConnectionManagerFactory$DelegatingHttpClientConnectionManager.requestConnection(ClientConnectionManagerFactory.java:75)
	at software.amazon.awssdk.http.apache.internal.conn.ClientConnectionManagerFactory$InstrumentedHttpClientConnectionManager.requestConnection(ClientConnectionManagerFactory.java:57)
	at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:176)
	at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:186)
	at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
	at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:56)
	at software.amazon.awssdk.http.apache.internal.impl.ApacheSdkHttpClient.execute(ApacheSdkHttpClient.java:72)
	at software.amazon.awssdk.http.apache.ApacheHttpClient.execute(ApacheHttpClient.java:254)
	at software.amazon.awssdk.http.apache.ApacheHttpClient.access$500(ApacheHttpClient.java:104)
	at software.amazon.awssdk.http.apache.ApacheHttpClient$1.call(ApacheHttpClient.java:231)
	at software.amazon.awssdk.http.apache.ApacheHttpClient$1.call(ApacheHttpClient.java:228)
	at software.amazon.awssdk.core.internal.util.MetricUtils.measureDurationUnsafe(MetricUtils.java:99)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.MakeHttpRequestStage.executeHttpRequest(MakeHttpRequestStage.java:79)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.MakeHttpRequestStage.execute(MakeHttpRequestStage.java:57)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.MakeHttpRequestStage.execute(MakeHttpRequestStage.java:40)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptTimeoutTrackingStage.execute(ApiCallAttemptTimeoutTrackingStage.java:74)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptTimeoutTrackingStage.execute(ApiCallAttemptTimeoutTrackingStage.java:43)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.TimeoutExceptionHandlingStage.execute(TimeoutExceptionHandlingStage.java:79)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.TimeoutExceptionHandlingStage.execute(TimeoutExceptionHandlingStage.java:41)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptMetricCollectionStage.execute(ApiCallAttemptMetricCollectionStage.java:55)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptMetricCollectionStage.execute(ApiCallAttemptMetricCollectionStage.java:39)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage2.executeRequest(RetryableStage2.java:93)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage2.execute(RetryableStage2.java:56)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage2.execute(RetryableStage2.java:36)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.StreamManagingStage.execute(StreamManagingStage.java:53)
	at software.amazon.awssdk.core.internal.http.StreamManagingStage.execute(StreamManagingStage.java:35)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.executeWithTimer(ApiCallTimeoutTrackingStage.java:82)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.execute(ApiCallTimeoutTrackingStage.java:62)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.execute(ApiCallTimeoutTrackingStage.java:43)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallMetricCollectionStage.execute(ApiCallMetricCollectionStage.java:50)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallMetricCollectionStage.execute(ApiCallMetricCollectionStage.java:32)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ExecutionFailureExceptionReportingStage.execute(ExecutionFailureExceptionReportingStage.java:37)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ExecutionFailureExceptionReportingStage.execute(ExecutionFailureExceptionReportingStage.java:26)
	at software.amazon.awssdk.core.internal.http.AmazonSyncHttpClient$RequestExecutionBuilderImpl.execute(AmazonSyncHttpClient.java:210)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.invoke(BaseSyncClientHandler.java:103)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.doExecute(BaseSyncClientHandler.java:173)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.lambda$execute$0(BaseSyncClientHandler.java:66)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.measureApiCallSuccess(BaseSyncClientHandler.java:182)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.execute(BaseSyncClientHandler.java:60)
	at software.amazon.awssdk.core.client.handler.SdkSyncClientHandler.execute(SdkSyncClientHandler.java:52)
	at software.amazon.awssdk.awscore.client.handler.AwsSyncClientHandler.execute(AwsSyncClientHandler.java:60)
	at software.amazon.awssdk.services.s3.DefaultS3Client.getObject(DefaultS3Client.java:5404)
	at org.apache.iceberg.aws.s3.S3InputStream.openStream(S3InputStream.java:240)
	at org.apache.iceberg.aws.s3.S3InputStream.openStream(S3InputStream.java:225)
	at org.apache.iceberg.aws.s3.S3InputStream.positionStream(S3InputStream.java:221)
	org.apache.iceberg.aws.s3.S3InputStream.read (File: S3InputStream.java, Line: 143, Column: undefined)
    com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.ensureLoaded (File: ByteSourceJsonBootstrapper.java, Line: 547, Column: undefined)
    com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.detectEncoding (File: ByteSourceJsonBootstrapper.java, Line: 137, Column: undefined)
    com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper.constructParser (File: ByteSourceJsonBootstrapper.java, Line: 266, Column: undefined)
    com.fasterxml.jackson.core.JsonFactory._createParser (File: JsonFactory.java, Line: 1874, Column: undefined)
    com.fasterxml.jackson.core.JsonFactory.createParser (File: JsonFactory.java, Line: 1273, Column: undefined)
    com.fasterxml.jackson.databind.ObjectMapper.readValue (File: ObjectMapper.java, Line: 3897, Column: undefined)
    org.apache.iceberg.TableMetadataParser.read (File: TableMetadataParser.java, Line: 280, Column: undefined)
    org.apache.iceberg.TableMetadataParser.read (File: TableMetadataParser.java, Line: 273, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.lambda$refreshFromMetadataLocation$0 (File: BaseMetastoreTableOperations.java, Line: 179, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.lambda$refreshFromMetadataLocation$1 (File: BaseMetastoreTableOperations.java, Line: 198, Column: undefined)
    org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry (File: Tasks.java, Line: 413, Column: undefined)
    org.apache.iceberg.util.Tasks$Builder.runSingleThreaded (File: Tasks.java, Line: 219, Column: undefined)
    org.apache.iceberg.util.Tasks$Builder.run (File: Tasks.java, Line: 203, Column: undefined)
    org.apache.iceberg.util.Tasks$Builder.run (File: Tasks.java, Line: 196, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.refreshFromMetadataLocation (File: BaseMetastoreTableOperations.java, Line: 198, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.refreshFromMetadataLocation (File: BaseMetastoreTableOperations.java, Line: 175, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.refreshFromMetadataLocation (File: BaseMetastoreTableOperations.java, Line: 170, Column: undefined)
    org.apache.iceberg.hive.HiveTableOperations.doRefresh (File: HiveTableOperations.java, Line: 167, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.refresh (File: BaseMetastoreTableOperations.java, Line: 87, Column: undefined)
    org.apache.iceberg.BaseMetastoreTableOperations.current (File: BaseMetastoreTableOperations.java, Line: 70, Column: undefined)
    org.apache.iceberg.BaseMetastoreCatalog.loadTable (File: BaseMetastoreCatalog.java, Line: 49, Column: undefined)
    org.apache.iceberg.connect.data.IcebergWriterFactory.createWriter (File: IcebergWriterFactory.java, Line: 59, Column: undefined)
    org.apache.iceberg.connect.data.SinkWriter.lambda$writerForTable$3 (File: SinkWriter.java, Line: 139, Column: undefined)
    at java.base/java.util.HashMap.computeIfAbsent(HashMap.java:1228)
    org.apache.iceberg.connect.data.SinkWriter.writerForTable (File: SinkWriter.java, Line: 138, Column: undefined)
    org.apache.iceberg.connect.data.SinkWriter.lambda$routeRecordStatically$1 (File: SinkWriter.java, Line: 98, Column: undefined)
    at java.base/java.util.Arrays$ArrayList.forEach(Arrays.java:4305)
    org.apache.iceberg.connect.data.SinkWriter.routeRecordStatically (File: SinkWriter.java, Line: 96, Column: undefined)
    org.apache.iceberg.connect.data.SinkWriter.save (File: SinkWriter.java, Line: 85, Column: undefined)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
    org.apache.iceberg.connect.data.SinkWriter.save (File: SinkWriter.java, Line: 68, Column: undefined)
    org.apache.iceberg.connect.channel.Worker.save (File: Worker.java, Line: 124, Column: undefined)
    org.apache.iceberg.connect.channel.CommitterImpl.save (File: CommitterImpl.java, Line: 88, Column: undefined)
    org.apache.iceberg.connect.IcebergSinkTask.put (File: IcebergSinkTask.java, Line: 87, Column: undefined)
    org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages (File: WorkerSinkTask.java, Line: 601, Column: undefined)
    ... 11 more

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[hussein-awala]

I believe #11609 will fix the issue.

[Fokko]

@hussein-awala Thanks for taking a stab at this. When cherry-picking your work, and testing against apache/iceberg-python#1323 I'm still seeing the error 🤔

[Fokko]

I'm going to close this one as well, to ensure that we have all the discussion in one central place: #11633