java - 消费者不读取来自 Kafka 主题的消息(Akka Stream Kafka)
问题描述
我们使用 Akka Stream Kafka 来生产和消费消息,并使用 Strimzi Kafka 集群。以下是重要的版本:
com.typesafe.akka:akka-stream-kafka_2.13:2.0.7
com.typesafe.akka:akka-stream_2.13:2.6.14
org.apache.kafka:kafka-clients:2.4.1 (*)
org.scala-lang.modules:scala-collection-compat_2.13:2.1.6
org.scala-lang:scala-library:2.13.5
重构消息后,消费者停止工作。我们确实在主题中有消息,但消费者只是无休止地等待。
这是日志片段:
[2021-04-14 21:20:43,869] [INFO] [org.apache.kafka.common.utils.AppInfoParser] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-8] - Kafka version: 2.4.1
[2021-04-14 21:20:43,869] [INFO] [org.apache.kafka.common.utils.AppInfoParser] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-8] - Kafka commitId: c57222ae8cd7866b
[2021-04-14 21:20:43,869] [INFO] [org.apache.kafka.common.utils.AppInfoParser] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-8] - Kafka startTimeMs: 1618424443866
[2021-04-14 21:20:43,879] [INFO] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-8] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Subscribed to topic(s): xyz-abc-import-dev-abc-input-topic
[2021-04-14 21:20:45,907] [INFO] [org.apache.kafka.clients.Metadata] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-19] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Cluster ID: L9OdIPABTGa7V9OPdViAaw
[2021-04-14 21:20:45,973] [INFO] [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-21] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Discovered group coordinator kafka-staging-abc-cluster-kafka-0-kafka-abc.xyzabccluster-host:443 (id: 2147483647 rack: null)
[2021-04-14 21:20:46,245] [INFO] [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-21] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] (Re-)joining group
[2021-04-14 21:20:47,554] [INFO] [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-9] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] (Re-)joining group
[2021-04-14 21:20:50,780] [INFO] [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-22] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Finished assignment for group at generation 5: {consumer-xyz-abc-import-1-995fd3d7-24b5-480d-90bc-b0967f0898f0=Assignment(partitions=[xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1])}
[2021-04-14 21:20:51,114] [INFO] [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-16] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Successfully joined group with generation 5
[2021-04-14 21:20:51,125] [INFO] [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-16] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Adding newly assigned partitions: xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1
[2021-04-14 21:20:51,334] [INFO] [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-18] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Setting offset for partition xyz-abc-import-dev-abc-input-topic-0 to the committed offset FetchPosition{offset=38, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=kafka-staging-abc-cluster-kafka-0-kafka-abc.xyzabccluster-host:443 (id: 0 rack: null), epoch=0}}
[2021-04-14 21:20:51,336] [INFO] [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-18] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Setting offset for partition xyz-abc-import-dev-abc-input-topic-1 to the committed offset FetchPosition{offset=51, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=kafka-staging-abc-cluster-kafka-1-kafka-abc.xyzabccluster-host:443 (id: 1 rack: null), epoch=0}}
还有几点:
- 模式注册表配置正确且良好(否则生产者将无法工作)。
- 主题(和组协调员)很好,我可以像这样通过普通消费者消费消息:
KafkaConsumer<String, MyMsg> consumer = new KafkaConsumer<String, MyMsg>(props);
consumer.subscribe(Collections.singletonList(inputTopic), rebalanceListener);
ConsumerRecords<String, MyMsg> records = consumer.poll(Duration.of(60L, ChronoUnit.SECONDS));
- 我使用“最早”重置配置属性,所以它应该看到未使用的消息。
- 这就是我运行时在主题中的内容 - 在分区 1 中,我在偏移量 51 和 52 处有消息,并且消费者协调器在日志中设置为偏移量 51,所以消息就在那里(事实上,它必须读取我产生的 2 条消息用于此测试)
这是代码卡住的地方——我使用阻塞调用来获取 2 条消息(甚至不能得到 1 条)
final Source<ProducerMessage.Results<String, Object, ConsumerMessage.PartitionOffset>, NotUsed> stream = ...
stream.take(SUBMISSION_SIZE).runWith(Sink.ignore(), mat).toCompletableFuture().get();
不太确定如何调试它。8-(
[更新 1]
它可以以某种方式与交易有关吗?因为普通的 Akka Stream Consumer 可以看到消息并消费:
@Test
@Order(3)
void exploreTopic() throws IOException, ExecutionException, InterruptedException {
Consumer.DrainingControl<java.util.List<ConsumerRecord<String, Object>>> controlCompletionStagePair =
Consumer.plainSource(consumerSettings, Subscriptions.topics(inputTopic))
.take(SUBMISSION_SIZE)
.map(x -> {
System.out.println(x);
return x;
})
.toMat(Sink.seq(), Consumer::createDrainingControl)
.run(mat);
controlCompletionStagePair.streamCompletion().toCompletableFuture().get();
System.out.println("xxx");
}
[更新 2]
我打开了 DEBUG 级别的日志记录,看到我的分区处于暂停状态:
[2021-04-14 23:46:05,122] [DEBUG] [org.apache.kafka.clients.consumer.internals.Fetcher] [kafka-coordinator-heartbeat-thread | xyz-abc-import] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Fetch READ_COMMITTED at offset 40 for partition xyz-abc-import-dev-abc-input-topic-0 returned fetch data (error=NONE, highWaterMark=40, lastStableOffset = 40, logStartOffset = 40, preferredReadReplica = absent, abortedTransactions = [], recordsSizeInBytes=0)
[2021-04-14 23:46:05,725] [DEBUG] [org.apache.kafka.clients.consumer.internals.Fetcher] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-6] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Added READ_COMMITTED fetch request for partition xyz-abc-import-dev-abc-input-topic-0 at position FetchPosition{offset=40, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=kafka-staging-abc-cluster-kafka-0-kafka-abc.xyzabccluster-host:443 (id: 0 rack: null), epoch=0}} to node kafka-staging-abc-cluster-kafka-0-kafka-abc.xyzabccluster-host:443 (id: 0 rack: null)
[2021-04-14 23:46:05,725] [DEBUG] [org.apache.kafka.clients.consumer.internals.Fetcher] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-6] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Added READ_COMMITTED fetch request for partition xyz-abc-import-dev-abc-input-topic-1 at position FetchPosition{offset=61, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=kafka-staging-abc-cluster-kafka-1-kafka-abc.xyzabccluster-host:443 (id: 1 rack: null), epoch=0}} to node kafka-staging-abc-cluster-kafka-1-kafka-abc.xyzabccluster-host:443 (id: 1 rack: null)
[2021-04-14 23:46:05,725] [DEBUG] [org.apache.kafka.clients.FetchSessionHandler] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-6] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Built incremental fetch (sessionId=339828876, epoch=1) for node 0. Added 0 partition(s), altered 0 partition(s), removed 0 partition(s) out of 1 partition(s)
[2021-04-14 23:46:05,725] [DEBUG] [org.apache.kafka.clients.FetchSessionHandler] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-6] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Built incremental fetch (sessionId=23429972, epoch=1) for node 1. Added 0 partition(s), altered 1 partition(s), removed 0 partition(s) out of 1 partition(s)
[2021-04-14 23:46:05,725] [DEBUG] [org.apache.kafka.clients.consumer.internals.Fetcher] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-6] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Sending READ_COMMITTED IncrementalFetchRequest(toSend=(), toForget=(), implied=(xyz-abc-import-dev-abc-input-topic-0)) to broker kafka-staging-abc-cluster-kafka-0-kafka-abc.xyzabccluster-host:443 (id: 0 rack: null)
[2021-04-14 23:46:05,726] [DEBUG] [org.apache.kafka.clients.consumer.internals.Fetcher] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-6] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Sending READ_COMMITTED IncrementalFetchRequest(toSend=(xyz-abc-import-dev-abc-input-topic-1), toForget=(), implied=()) to broker kafka-staging-abc-cluster-kafka-1-kafka-abc.xyzabccluster-host:443 (id: 1 rack: null)
[2021-04-14 23:46:05,796] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-7] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:05,866] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-10] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:05,935] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-11] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,009] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-13] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,075] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-15] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,149] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-18] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,215] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-20] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,288] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-21] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,355] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-8] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,355] [DEBUG] [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [kafka-coordinator-heartbeat-thread | xyz-abc-import] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Sending Heartbeat request to coordinator kafka-staging-abc-cluster-kafka-0-kafka-abc.xyzabccluster-host:443 (id: 2147483647 rack: null)
[2021-04-14 23:46:06,427] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-9] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,459] [DEBUG] [org.apache.kafka.clients.FetchSessionHandler] [kafka-coordinator-heartbeat-thread | xyz-abc-import] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Node 1 sent an incremental fetch response for session 23429972 with 0 response partition(s), 1 implied partition(s)
[2021-04-14 23:46:06,459] [DEBUG] [org.apache.kafka.clients.FetchSessionHandler] [kafka-coordinator-heartbeat-thread | xyz-abc-import] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Node 0 sent an incremental fetch response for session 339828876 with 0 response partition(s), 1 implied partition(s)
[2021-04-14 23:46:06,500] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-12] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,557] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-14] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,559] [DEBUG] [org.apache.kafka.clients.consumer.internals.AbstractCoordinator] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-14] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Received successful Heartbeat response
[2021-04-14 23:46:06,625] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-16] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,697] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-17] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[2021-04-14 23:46:06,765] [DEBUG] [org.apache.kafka.clients.consumer.KafkaConsumer] [SchemaRegistrySerializationTest-akka.kafka.default-dispatcher-19] - [Consumer clientId=consumer-xyz-abc-import-1, groupId=xyz-abc-import] Pausing partitions [xyz-abc-import-dev-abc-input-topic-0, xyz-abc-import-dev-abc-input-topic-1]
[UPD 3] 它与自动创建的主题有关吗?
解决方案
很难说它是如何影响的,但是一旦我将此属性添加到消费者,消费者就开始正常工作:
props.put(AvroDatumProvider.REGISTRY_USE_SPECIFIC_AVRO_READER_CONFIG_PARAM, String.valueOf(Boolean.TRUE));
现在确定真正的原因是什么,以及为什么没有抛出错误,但是在上面添加属性有帮助,至少在我的情况下。
推荐阅读
- eclipse - XHTML 语法验证器在 Eclipse 中不可见(在 Window->Preferences 下的验证菜单选项中)
- javascript - Redux - 找不到目录来调用调度
- clang - 使用插件修改 clang AST
- swift - 表格视图中单元格内的自动布局图像。正确的布局,但只有一次向下滚动和备份?
- php - Laravel - 多个图像爆炸/内爆
- html - 尝试调整我的徽标大小并在徽标的侧面添加标题而不调整导航栏的大小
- qt - ChangeEvent 块 QMDISubWINdow
- spring-boot - 如何通过使用 ExecutorService 和 CompletableFuture 控制线程数,在 Spring Flux 中立即向客户端返回响应?
- sql - Oracle 中的 DBMS_ASSERT.enquote_name()
- reactjs - 将 ag-grid 的方向更改为 RTL