时间戳

首页 > 解决方案 > 在 kafka 流记录转发期间，接收器的分区计数减少

apache-kafka - 在 kafka 流记录转发期间，接收器的分区计数减少

问题描述

我正在使用 kafka 流来处理一些 kafka 记录，我有两个节点，一个用于进行一些转换，另一个是最终接收器。

我的主题是 INTER_TOPIC 和 FINAL_TOPIC 每个有 20 个分区。我的生产者写入 INTER_TOPIC 正在写入键值，分区器是循环法。

下面是我的间变换节点的代码。

public void streamHandler() {

        Properties props = getKafkaProperties();

        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> processStream = builder.stream("INTER_TOPIC",
                Consumed.with(Serdes.String(), Serdes.String()));

        //processStream.peek((key,value)->System.out.println("key :"+key+" value :"+value));

        processStream.map((key, value) -> getTransformer().transform(key, value)).filter((key,value)->filteroutFailedRequest(key,value)).to("FINAL_TOPIC", Produced.with(Serdes.String(), Serdes.String()));


        KafkaStreams IStreams = new KafkaStreams(builder.build(), props);

        IStreams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            @Override
            public void uncaughtException(Thread t, Throw-able e) {

                logger.error("Thread Name :" + t.getName() + " Error while processing:", e);
            }
        });

        IStreams.cleanUp();
        IStreams.start();

        try {
            System.in.read();
        } catch (IOException e) {

            logger.error("Failed streaming ",e);
        }
    }

但是我的接收器仅在 2 个分区中获取数据，但我配置了 20 个流线程，并且我验证了我的生产者正在写入所有 20 个分区，如何知道我的转换节点转发到我的 FINAL_TOPIC 的所有 20 个分区

30 Sep 2019 10:39:41,416 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-3] Received
30 Sep 2019 10:39:41,416 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-4] Received
30 Sep 2019 10:39:41,416 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-3] Received
30 Sep 2019 10:39:41,416 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-4] Received
30 Sep 2019 10:40:57,427 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-3] Received
30 Sep 2019 10:40:57,427 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-4] Received
30 Sep 2019 10:40:57,427 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-3] Received
30 Sep 2019 10:40:57,427 INFO  c.j.m.s.StreamHandler [289] [streams-user-61a77203-9afc-4c66-843d-94c20a509793-StreamThread-4] Received

标签： apache-kafkaapache-kafka-streams

解决方案

并且 partition-er 是循环的

为什么你认为分区器是循环的？默认情况下，Kafka Streams 基于键应用基于散列的分区。

如果要更改默认分区器，可以实现接口StreamPartitioner并通过以下方式传递：

Produced.with(Serdes.String(), Serdes.String())
        .withStreamPartitioner(...)

推荐阅读