Java的Kafka消费者-未收到记录

Question

我的问题是关于Kafka在爪哇的消费者

>

已启动Kafka服务器

bin/kafka-server-start.sh config/server.properties

创建的主题

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

创作者

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

创建的消费者

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test

我在终端中做的所有这些事情，工作正常，能够在消费者处正确接收日志。运行下面的消费者（在Java），但没有收到任何记录。它继续汇集在

ConsumerRecords<String, String> records = kafkaConsumer.poll(1); 

也没有收到任何记录。

下面给出的我的 java 消费者代码

package com.myCompany.kafka.subscription;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaSubscription {

    public static void main(String args[]) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.95.217:9092");
        props.put("zookeeper.connect", "192.168.95.217:2181");
        props.put("group.id", "test-consumer-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "earliest");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        System.out.println("properties loaded");
        kafkaConsumer.subscribe(Arrays.asList("test"));
        kafkaConsumer.seekToBeginning(Collections.emptyList());
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(1);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, value = %s", record.offset(), record.value());
                System.out.println();
            }
        }
    }

}

请告诉我在java消费者类中接收消息，我在配置中做错了什么吗？属性中的“group.id”是怎么回事？

下面给出了java消费者端的日志。

0    [main] INFO  org.apache.kafka.clients.consumer.ConsumerConfig  - ConsumerConfig values: 
    auto.commit.interval.ms = 1000
    auto.offset.reset = earliest
    bootstrap.servers = [192.168.95.217:9092]
    check.crcs = true
    client.id = 
    connections.max.idle.ms = 540000
    enable.auto.commit = true
    exclude.internal.topics = true
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    group.id = test-consumer-group
    heartbeat.interval.ms = 3000
    interceptor.classes = null
    internal.leave.group.on.close = true
    isolation.level = read_uncommitted
    key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 305000
    retry.backoff.ms = 100
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    session.timeout.ms = 30000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

4    [main] DEBUG org.apache.kafka.clients.consumer.KafkaConsumer  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Initializing the Kafka consumer
118  [main] DEBUG org.apache.kafka.clients.Metadata  - Updated cluster metadata version 1 to Cluster(id = null, nodes = [192.168.95.217:9092 (id: -1 rack: null)], partitions = [])
138  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name fetch-throttle-time
157  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name connections-closed:
163  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name connections-created:
164  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name successful-authentication:
164  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name failed-authentication:
165  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name bytes-sent-received:
166  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name bytes-sent:
168  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name bytes-received:
169  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name select-time:
171  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name io-time:
198  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name heartbeat-latency
198  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name join-latency
199  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name sync-latency
202  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name commit-latency
209  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name bytes-fetched
210  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name records-fetched
210  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name fetch-latency
211  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name records-lag
220  [main] INFO  org.apache.kafka.common.utils.AppInfoParser  - Kafka version : 1.0.0
220  [main] INFO  org.apache.kafka.common.utils.AppInfoParser  - Kafka commitId : aaa7af6d4a11b29d
223  [main] DEBUG org.apache.kafka.clients.consumer.KafkaConsumer  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Kafka consumer initialized
properties loaded
224  [main] DEBUG org.apache.kafka.clients.consumer.KafkaConsumer  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Subscribed to topic(s): test
224  [main] DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Sending GroupCoordinator request to broker 192.168.95.217:9092 (id: -1 rack: null)
386  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Initiating connection to node 192.168.95.217:9092 (id: -1 rack: null)
488  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node--1.bytes-sent
489  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node--1.bytes-received
490  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node--1.latency
491  [main] DEBUG org.apache.kafka.common.network.Selector  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Created socket with SO_RCVBUF = 65536, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to node -1
492  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Completed connection to node -1. Fetching API versions.
492  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Initiating API versions fetch from node -1.
506  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Recorded API versions for node -1: (Produce(0): 0 to 5 [usable: 5], Fetch(1): 0 to 6 [usable: 6], ListOffsets(2): 0 to 2 [usable: 2], Metadata(3): 0 to 5 [usable: 5], LeaderAndIsr(4): 0 to 1 [usable: 1], StopReplica(5): 0 [usable: 0], UpdateMetadata(6): 0 to 4 [usable: 4], ControlledShutdown(7): 0 to 1 [usable: 1], OffsetCommit(8): 0 to 3 [usable: 3], OffsetFetch(9): 0 to 3 [usable: 3], FindCoordinator(10): 0 to 1 [usable: 1], JoinGroup(11): 0 to 2 [usable: 2], Heartbeat(12): 0 to 1 [usable: 1], LeaveGroup(13): 0 to 1 [usable: 1], SyncGroup(14): 0 to 1 [usable: 1], DescribeGroups(15): 0 to 1 [usable: 1], ListGroups(16): 0 to 1 [usable: 1], SaslHandshake(17): 0 to 1 [usable: 1], ApiVersions(18): 0 to 1 [usable: 1], CreateTopics(19): 0 to 2 [usable: 2], DeleteTopics(20): 0 to 1 [usable: 1], DeleteRecords(21): 0 [usable: 0], InitProducerId(22): 0 [usable: 0], OffsetForLeaderEpoch(23): 0 [usable: 0], AddPartitionsToTxn(24): 0 [usable: 0], AddOffsetsToTxn(25): 0 [usable: 0], EndTxn(26): 0 [usable: 0], WriteTxnMarkers(27): 0 [usable: 0], TxnOffsetCommit(28): 0 [usable: 0], DescribeAcls(29): 0 [usable: 0], CreateAcls(30): 0 [usable: 0], DeleteAcls(31): 0 [usable: 0], DescribeConfigs(32): 0 [usable: 0], AlterConfigs(33): 0 [usable: 0], AlterReplicaLogDirs(34): 0 [usable: 0], DescribeLogDirs(35): 0 [usable: 0], SaslAuthenticate(36): 0 [usable: 0], CreatePartitions(37): 0 [usable: 0])
506  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Sending metadata request (type=MetadataRequest, topics=test) to node 192.168.95.217:9092 (id: -1 rack: null)
514  [main] DEBUG org.apache.kafka.clients.Metadata  - Updated cluster metadata version 2 to Cluster(id = jm5HyHZbT1mlgdZxinD8oA, nodes = [localhost:9092 (id: 0 rack: null)], partitions = [Partition(topic = test, partition = 0, leader = 0, replicas = [0], isr = [0], offlineReplicas = [])])
516  [main] DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Received GroupCoordinator response ClientResponse(receivedTimeMs=1515749439145, latencyMs=134, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1, clientId=consumer-1, correlationId=0), responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=NONE, node=localhost:9092 (id: 0 rack: null)))
516  [main] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Discovered coordinator localhost:9092 (id: 2147483647 rack: null)
516  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Initiating connection to node localhost:9092 (id: 2147483647 rack: null)
520  [kafka-coordinator-heartbeat-thread | test-consumer-group] DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Heartbeat thread started
520  [main] DEBUG org.apache.kafka.clients.consumer.internals.ConsumerCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Sending synchronous auto-commit of offsets {}
520  [main] INFO  org.apache.kafka.clients.consumer.internals.ConsumerCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Revoking previously assigned partitions []
520  [main] DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Disabling heartbeat thread
520  [main] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] (Re-)joining group
523  [main] DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Sending JoinGroup ((type: JoinGroupRequest, groupId=test-consumer-group, sessionTimeout=30000, rebalanceTimeout=300000, memberId=, protocolType=consumer, groupProtocols=org.apache.kafka.common.requests.JoinGroupRequest$ProtocolMetadata@19d481b)) to coordinator localhost:9092 (id: 2147483647 rack: null)
525  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node-2147483647.bytes-sent
526  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node-2147483647.bytes-received
527  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node-2147483647.latency
528  [main] DEBUG org.apache.kafka.common.network.Selector  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Connection with localhost/127.0.0.1 disconnected
java.net.ConnectException: Connection refused
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
    at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50)
    at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:106)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:444)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:398)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:460)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:238)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:214)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:174)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:364)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1138)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1103)
    at com.myCompany.kafka.subscription.KafkaSubscription.main(KafkaSubscription.java:30)
533  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Node 2147483647 disconnected.
534  [main] WARN  org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Connection to node 2147483647 could not be established. Broker may not be available.
534  [main] DEBUG org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Cancelled JOIN_GROUP request RequestHeader(apiKey=JOIN_GROUP, apiVersion=2, clientId=consumer-1, correlationId=3) with correlation id 3 due to node 2147483647 being disconnected
534  [main] INFO  org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Marking the coordinator localhost:9092 (id: 2147483647 rack: null) dead
635  [main] DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Sending GroupCoordinator request to broker localhost:9092 (id: 0 rack: null)
635  [main] DEBUG org.apache.kafka.clients.NetworkClient  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Initiating connection to node localhost:9092 (id: 0 rack: null)
635  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node-0.bytes-sent
636  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node-0.bytes-received
637  [main] DEBUG org.apache.kafka.common.metrics.Metrics  - Added sensor with name node-0.latency
638  [main] DEBUG org.apache.kafka.common.network.Selector  - [Consumer clientId=consumer-1, groupId=test-consumer-group] Connection with localhost/127.0.0.1 

Answer 1

由于您正在远程服务器上运行 kafka，因此需要在服务器属性中取消注释播发的侦听器。

例如：adverted.listeners=PLAINTEXT://192.168.95.217:9092

之后，重启kafka服务器和远程消费者。

有关更多信息，请查看此博客文章https://rmoff.net/2018/08/02/kafka-listeners-explained/.

tl；dr：您需要设置广告。监听器（或KAFKA_ADVERTISED_listeners，如果您使用Docker映像）连接到外部地址（主机/IP），以便客户端可以正确连接到它。否则，他们会尝试连接到内部主机地址——如果无法访问，则会出现问题。

它给出了比实际文档更简洁的解释。

我运行的是Kafka2.13版

Answer 2

您的应用程序在我的基于Kafka 1.0.0的安装上工作正常，您使用的是什么版本？其他一些值得一提的事情：

控制台使用者应用程序正在使用已弃用的旧使用者运行，因此您可以使用--bootstrap server localhost:9092而不是使用--zookeeper localhost:2181。

然后在应用程序中使用＜code＞zookeeper。连接属性不是消费者属性，您不需要它。您只需要＜code＞引导。服务器用于连接到kafka代理的属性。

最后，如您所问，组。id是消费者将属于的消费者组名称。