【问题标题】:org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partitionorg.apache.kafka.clients.consumer.NoOffsetForPartitionException:未定义的偏移量,没有分区的重置策略
【发布时间】:2022-01-25 21:19:25
【问题描述】:

我们正在使用 Kafka Connect HDFS 连接器,它不断从 Kafka 主题中提取数据并将它们提交到 HDFS。

在成功加载 12 + 11 小时后,我们突然在连接器端发现了这个错误。

    org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partition:Prd_IN_GeneralEvents-39
    at org.apache.kafka.clients.consumer.internals.Fetcher.resetOffset(Fetcher.java:374)at org.apache.kafka.clients.consumer.internals.Fetcher.resetOffsetsIfNeeded(Fetcher.java:227)at org.apache.kafka.clients.consumer.KafkaConsumer.updateFetchPositions(KafkaConsumer.java:1592)at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1035)at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995)
at org.apache.kafka.connect.runtime.WorkerSinkTask.pollConsumer(WorkerSinkTask.java:360)at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:245)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
at
org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:744)

然后几个(100 个中的 9 个)HDFS 工作线程被杀死,我们开始丢失数据。

这个错误的根本原因是什么?

我们在 connect.distributed.properties 文件中设置了auto.offset.reset=latest

【问题讨论】:

  • 首先我注意到 auto.offset.reset 应该有消费者作为前缀(consumer.auto.offset.reset)。您能否提供代理日志,以防万一您在失败时为偏移主题存储了状态。
  • 我早上尝试了 consumer.auto.offset.reset=latest。我在日志中看到 auto.offset.reset = early 已设置。我不知道它是如何指向最早的。将 auto.offset.reset 和 consumer.auto.offset.reset 都保留在 connect.distributed.properties 中,但仍然在日志中我看到 auto.offset.reset = early。还有一些属性未被我的 connect.distributed.properties 文件中的值覆盖。 consumer.auto.commit.interval.ms 在我的 connect.distributed.properties 中设置为 4000,但我在日志中看到 auto.commit.interval.ms = 5000。
  • 我也看到了 ----- [2017-06-21 16:10:01,321] WARN 提供了配置“auto.offset.reset”,但不是已知配置。 (org.apache.kafka.clients.producer.ProducerConfig:188) 和 [2017-06-21 16:10:01,366] WARN 提供了配置“consumer.auto.offset.reset”,但不是已知配置。 (org.apache.kafka.clients.consumer.ConsumerConfig:188) 在我的日志文件中。许多配置都存在类似的警告,例如 consumer.enable.auto.commit、consumer.max.poll.interval.ms、internal.key.converter.schemas.enable、consumer.auto.offset.reset

标签: hadoop apache-kafka hdfs apache-kafka-connect confluent-platform


【解决方案1】:
 'scan.startup.mode' = 'earliest-offset',
 'properties.auto.offset.reset' = 'earliest'

【讨论】:

  • 请记住,Stack Overflow 不仅仅是为了解决眼前的问题,而是为了帮助未来的读者找到类似问题的解决方案,这需要了解底层代码。这对于我们社区的初学者和不熟悉语法的成员来说尤其重要。鉴于此,您能否edit 您的答案包括对您正在做什么的解释以及为什么您认为这是最好的方法?
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2016-09-16
  • 1970-01-01
  • 2019-07-20
  • 2013-08-18
  • 2011-03-13
  • 1970-01-01
相关资源
最近更新 更多