Kafka Streams 未读取输入主题答案

【问题标题】：Kafka Streams not reading input topicKafka Streams 未读取输入主题
【发布时间】：2018-03-07 18:54:15
【问题描述】：

我根据教程创建了示例 Kafka Streams 应用程序：

    public static void main(String[] args) throws Exception {
    Logger log = Logger.getLogger("Name");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordprint");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.99.100:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    final KStreamBuilder builder = new KStreamBuilder();
    builder.stream("onecon_postgres").print();

    final KafkaStreams streams = new KafkaStreams(builder, props);
    final CountDownLatch latch = new CountDownLatch(1);

    // attach shutdown handler to catch control-c
    Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") {
        @Override
        public void run() {
            streams.close();
            latch.countDown();
        }
    });

    try {
        streams.start();
        log.info("After Start");
        latch.await();
    } catch (Throwable e) {
        System.exit(1);
    }
    System.exit(0);
    }

很遗憾，此应用程序不读取输入流。我有一个来自 PostgreSQL 的 JDBC 源连接器，它可以很好地处理来自一个数据库的流数据（我可以在本主题中看到 Kafka Connect UI 数据）。

我遇到的问题是即使我在属性 IP 中的 BOOTSTRAP_SERVERS_CONFIG 中更改了 IP 是 localhost 我不知道为什么。

[main] INFO org.apache.kafka.streams.StreamsConfig - StreamsConfig values: 
    application.id = streams-linesplit
    application.server = 
    **bootstrap.servers = [localhost:9092]**
    buffered.records.per.partition = 1000
    cache.max.bytes.buffering = 10485760
    client.id = 
    commit.interval.ms = 30000
    connections.max.idle.ms = 540000
    default.key.serde = class org.apache.kafka.common.serialization.Serdes$StringSerde
    default.timestamp.extractor = class org.apache.kafka.streams.processor.FailOnInvalidTimestamp
    default.value.serde = class org.apache.kafka.common.serialization.Serdes$StringSerde
    key.serde = null
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    num.standby.replicas = 0
    num.stream.threads = 1
    partition.grouper = class org.apache.kafka.streams.processor.DefaultPartitionGrouper
    poll.ms = 100
    processing.guarantee = at_least_once
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    replication.factor = 1
    request.timeout.ms = 40000
    retry.backoff.ms = 100
    rocksdb.config.setter = null
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    state.cleanup.delay.ms = 600000
    state.dir = /tmp/kafka-streams
    timestamp.extractor = null
    value.serde = null
    windowstore.changelog.additional.retention.ms = 86400000
    zookeeper.connect =

为了克服这个问题，我使用 netsh 转发流量，但我看不到这个应用程序消耗我的流。

netsh interface portproxy add v4tov4 listenport=9092 listenaddress=127.0.0.1 connectport=9092 connectaddress=192.168.99.100

【问题讨论】：

您是否尝试调试到您的应用程序？在StreamsThread 构造函数中设置断点以查看传递给它的配置。还要检查 StreamsConfig 在解析配置时是否正确。

标签： java apache-kafka apache-kafka-streams apache-kafka-connect

【解决方案1】：

很遗憾，这个应用程序不读取输入流。

您的 Kafka Streams 应用程序和 Kafka 代理之间似乎存在网络问题。 “Kafka Streams 不起作用”的可能性不大。

此外，如果您不提供更多信息，很难为您提供帮助：

您的 Kafka 代理使用哪个 Kafka 版本？
您的应用程序使用什么 Kafka (Streams) 版本？
哪个操作系统？
什么是网络设置？
- 运行您的应用程序的机器的 IP 地址。
- 您的 Kafka 代理（或多个代理）在哪个 IP + 端口上侦听新连接？是192.168.99.100:9092吗？
您在应用程序的日志中看到了什么？您是否看到 ERROR 或 WARN 日志消息？

我遇到的问题是即使我在属性中更改了 BOOTSTRAP_SERVERS_CONFIG 中的 IP IP 是 localhost 我不知道为什么。

我不明白 - 为什么您认为将 BOOTSTRAP_SERVERS_CONFIG 更改为 localhost:9092 会解决您原来的问题？我知道 Kafka 经纪人实际上在听 192.168.99.100:9092？

为了克服这个问题，我使用 netsh 转发流量，但我看不到这个应用程序消耗我的流。

端口转发很可能无济于事。在不更新 Kafka 代理的配置的情况下，代理默认情况下只会在其“真实”IP + 端口上进行通信。稍微简化：配置为侦听192.168.99.100:9092 的代理不会响应您的Kafka Streams 应用程序发送的localhost:9092 请求，即使您在运行Kafka Streams 应用程序的机器上从localhost:9092 -> 192.168.99.100:9092 进行端口转发。

希望这会有所帮助！

【讨论】：