Kafka scala使用者未从主题阅读消息控制台中无错误

我有下面的示例代码来阅读Kafka主题的消息

package com.krushna
package kafkademo

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, KafkaConsumer}

import java.time.Duration
import java.util.{Collections, Properties}

object TestConsumer extends App {

  val consumer= new KafkaConsumer[String,String](getProperties())
  consumer.subscribe(Collections.singletonList("wm-cth-salesstreams"))

  while(true){
    val data = consumer.poll(Duration.ofSeconds(3))
    data.forEach(println(_))
  }

  def getProperties(): Properties = {
    val properties: Properties = new Properties
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "scala-c1-1234")
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    properties
  }
}

它不从主题中阅读任何消息，而具有以下配置的Kafka控制台使用者正在读取消息。

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic wm-cth-salesstream  --from-beginning --group c2

在Scala/java控制台中，我可以看到下面的消息连续打印。

13:12:25.621 [main] DEBUG org.apache.kafka.clients.FetchSessionHandler - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Node 1001 sent an incremental fetch response for session 386731835 with 0 response partition(s), 1 implied partition(s)
13:12:25.621 [main] DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Added READ_UNCOMMITTED fetch request for partition wm-cth-salesstreams-0 at position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=localhost:9092 (id: 1001 rack: null), epoch=0}} to node localhost:9092 (id: 1001 rack: null)
13:12:25.621 [main] DEBUG org.apache.kafka.clients.FetchSessionHandler - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Built incremental fetch (sessionId=386731835, epoch=794) for node 1001. Added 0 partition(s), altered 0 partition(s), removed 0 partition(s) out of 1 partition(s)
13:12:25.621 [main] DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Sending READ_UNCOMMITTED IncrementalFetchRequest(toSend=(), toForget=(), implied=(wm-cth-salesstreams-0)) to broker localhost:9092 (id: 1001 rack: null)
13:12:26.123 [main] DEBUG org.apache.kafka.clients.FetchSessionHandler - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Node 1001 sent an incremental fetch response for session 386731835 with 0 response partition(s), 1 implied partition(s)
13:12:26.123 [main] DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Added READ_UNCOMMITTED fetch request for partition wm-cth-salesstreams-0 at position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=localhost:9092 (id: 1001 rack: null), epoch=0}} to node localhost:9092 (id: 1001 rack: null)
13:12:26.123 [main] DEBUG org.apache.kafka.clients.FetchSessionHandler - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Built incremental fetch (sessionId=386731835, epoch=795) for node 1001. Added 0 partition(s), altered 0 partition(s), removed 0 partition(s) out of 1 partition(s)
13:12:26.123 [main] DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Sending READ_UNCOMMITTED IncrementalFetchRequest(toSend=(), toForget=(), implied=(wm-cth-salesstreams-0)) to broker localhost:9092 (id: 1001 rack: null)
13:12:26.374 [main] DEBUG org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Sending asynchronous auto-commit of offsets {wm-cth-salesstreams-0=OffsetAndMetadata{offset=0, leaderEpoch=null, metadata=''}}
13:12:26.378 [main] DEBUG org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Committed offset 0 for partition wm-cth-salesstreams-0
13:12:26.378 [main] DEBUG org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Completed asynchronous auto-commit of offsets {wm-cth-salesstreams-0=OffsetAndMetadata{offset=0, leaderEpoch=null, metadata=''}}
13:12:26.625 [main] DEBUG org.apache.kafka.clients.FetchSessionHandler - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Node 1001 sent an incremental fetch response for session 386731835 with 0 response partition(s), 1 implied partition(s)
13:12:26.626 [main] DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Added READ_UNCOMMITTED fetch request for partition wm-cth-salesstreams-0 at position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=localhost:9092 (id: 1001 rack: null), epoch=0}} to node localhost:9092 (id: 1001 rack: null)
13:12:26.626 [main] DEBUG org.apache.kafka.clients.FetchSessionHandler - [Consumer clientId=consumer-scala-c1-1, groupId=scala-c1] Built incremental fetch (sessionId=386731835, epoch=796) for node 1001. Added 0 partition(s), altered 0 partition(s), removed 0 partition(s) out of 1 partition(s)

可能的错误是什么？
编辑
/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups的输出

GROUP           TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
c1              wm-cth-salesstream 0          360             360             0               console-consumer-3cba8e14-0835-48e6-9620-296aa32aa551 /127.0.0.1      console-consumer

GROUP           TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
c10             wm-cth-salesstream 0          360             360             0               console-consumer-614744ee-33ac-4dc9-87c7-66c6a1cdaa3a /127.0.0.1      console-consumer

Consumer group 'c2' has no active members.

GROUP           TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
c2              wm-cth-salesstream 0          360             360             0               -               -               -

Consumer group 'kafka-java-consumer' has no active members.

GROUP               TOPIC               PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
kafka-java-consumer wm-cth-salesstreams 0          0               0               0               -               -               -

Consumer group 'kafka-java-consumer-v1' has no active members.

GROUP                  TOPIC               PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
kafka-java-consumer-v1 wm-cth-salesstreams 0          0               0               0               -               -               -

Consumer group 'new-c1' has no active members.

GROUP           TOPIC               PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
new-c1          wm-cth-salesstreams 0          0               0               0               -               -               -

Consumer group 'scala-c1' has no active members.

GROUP           TOPIC               PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
scala-c1        wm-cth-salesstreams 0          0               0               0               -               -               -

GROUP           TOPIC               PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                   HOST            CLIENT-ID
scala-c1-1234   wm-cth-salesstreams 0          0               0               0               consumer-scala-c1-1234-1-bcc35a35-eb8b-495a-acd9-b98c0d528f93 /172.18.0.1     consumer-scala-c1-1234-1

Kafka scala使用者未从主题阅读消息控制台中无错误

1条答案

相关问题

热门标签

最新问答