Kafka Connector Record Writer因缺少要分配的内存而卡在S3OutputStream中，但在数小时内不会保持空闲

如何解决Kafka Connector Record Writer因缺少要分配的内存而卡在S3OutputStream中，但在数小时内不会保持空闲

我的行为我不知道该如何修改。 我正在测试s3 Kafka同步连接器，而我的主题中的数据很少。

此刻，我可以通过使用Kafka管理器看到主题中有数据，但是我的连接器读取数据，并且从未移动偏移量，也从未将其推送到Kafka。 在其他主题中，此方法有效，但在此特定主题中，则无效。 我认为这与超时有关，但是我找不到要设置的正确配置属性，因此刷新会更快一些。

这是我的配置：

      curl -X PUT -s -o /dev/null -H ""Content-Type:application/json""
      http://localhost:$$CONNECT_REST_PORT/connectors/s3_connector_doc_cmg/config
      \
        -d '{
          ""connector.class"": ""io.confluent.connect.s3.S3SinkConnector"",""storage.class"": ""io.confluent.connect.s3.storage.S3Storage"",""s3.region"": ""us-east-1"",""s3.bucket.name"": ""confluent-pipeline"",""topics.dir"": ""topics"",""topics"": ""com.acp.bde.doc_cmg"",""flush.size"": ""25"",""rotate.interval.ms"": ""5000"",""auto.register.schemas"": ""false"",""tasks.max"": ""1"",""s3.part.size"": ""5242880"",""timezone"": ""UTC"",""parquet.codec"": ""snappy"",""offset.flush.interval.ms"": ""5000"",""offset.flush.timeout.ms"": ""1000"",""s3.credentials.provider.class"": ""com.amazonaws.auth.DefaultAWSCredentialsProviderChain"",""format.class"": ""io.confluent.connect.s3.format.avro.AvroFormat"",""value.converter"": ""com.insight.connect.protobuf.ProtobufConverter"",""key.converter"": ""org.apache.kafka.connect.storage.StringConverter"",""partitioner.class"": ""io.confluent.connect.storage.partitioner.DailyPartitioner"",""locale"": ""de-CH"",""timezone"": ""Europe/Zurich"",""store.url"": ""http://minio-server-svc:9000/""
        }'"

这是我在日志中看到的：

[2020-10-23 10:35:47,594] INFO Opening record writer for: topics/com.acp.bde.doc_cmg/year=2020/month=10/day=23/com.acp.bde.doc_cmg+1+0000000000.avro (io.confluent.connect.s3.format.avro.AvroRecordWriterProvider)
[2020-10-23 10:35:48,017] INFO Opening record writer for: topics/com.acp.bde.doc_cmg/year=2020/month=10/day=23/com.acp.bde.doc_cmg+3+0000000000.avro (io.confluent.connect.s3.format.avro.AvroRecordWriterProvider)
[2020-10-23 10:35:48,075] INFO Opening record writer for: topics/com.acp.bde.doc_cmg/year=2020/month=10/day=23/com.acp.bde.doc_cmg+2+0000000000.avro (io.confluent.connect.s3.format.avro.AvroRecordWriterProvider)
[2020-10-23 11:35:37,989] INFO [Worker clientId=connect-1,groupId=kafka-connect-01] Session key updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder)

所以它们已经打开了将近1小时，什么也没有发生，我想知道我的配置是否完全错误，或者是否需要一些属性和配置，因此数据推送速度更快。

更新： 我仍然没有正确解决此问题，但实际上似乎是内存不足的问题。

程序只是停留在这一行 this.buffer = ByteBuffer.allocate（this.partSize）;

https://github.com/confluentinc/kafka-connect-storage-cloud/blob/master/kafka-connect-s3/src/main/java/io/confluent/connect/s3/storage/S3OutputStream.java#L85

困扰我的部分是它根本不会抱怨，只是停留在那里。它是否应该因内存不足问题而崩溃？ 还是不应该更快地释放内存？ 它几乎可以在通话中停留3或4小时以上，而没有任何反馈。

我仍然认为配置可能有问题，但是我不知道应该看什么或在哪里看。

解决方法

您的分区器是基于时间的。因此，这可能是由于 rotate.schedule.interval.ms 参数不存在时的行为而发生的。看看下面的话题https://stackoverflow.com/a/51160834/1551246