问题：

升级到Kafka-Streams:5.5.0-CSS（Apache Kafka 2.5.0)[已解决]

扶誉

2023-03-14

所以这是我的配置：

@Configuration
@EnableKafkaStreams
public class GlobalTableConfiguration {

    public GlobalTableConfiguration() {
    }

    @Bean
    public GlobalKTable<String, String> table(StreamsBuilder kStreamsBuilder) {
        return kStreamsBuilder.globalTable("topic1", Consumed.with(null, null), 
                                            Materialized.as("topic1-store"));
    }
}

我要把商店弄成这样：

streamsBuilderFactoryBean.getKafkaStreams().
                store("topic1-store", QueryableStoreTypes.keyValueStore());

这将失败：

Request processing failed; nested exception is java.lang.IllegalStateException: KafkaStreams is not running. State is ERROR.
org.springframework.web.util.NestedServletException: Request processing failed; nested exception is java.lang.IllegalStateException: KafkaStreams is not running. State is ERROR.
    at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1014)
    at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898)


Caused by: java.lang.IllegalStateException: KafkaStreams is not running. State is ERROR.
    at org.apache.kafka.streams.KafkaStreams.validateIsRunningOrRebalancing(KafkaStreams.java:316)
    at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1182)
    at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1169)

2020-06-16 13:22:46.943  INFO 72423 --- [    Test worker] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.5.0
        2020-06-16 13:22:46.944  INFO 72423 --- [    Test worker] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 66563e712b0b9f84
        2020-06-16 13:22:46.944  INFO 72423 --- [    Test worker] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1592299366943
        2020-06-16 13:22:46.946  INFO 72423 --- [ad | producer-2] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-2] Cluster ID: aKrIp_7wQcqF9OlSUoBgSQ
        2020-06-16 13:22:47.496  INFO 72423 --- [    Test worker] org.apache.kafka.streams.KafkaStreams    : stream-client [app-d09c3f52-8d77-4814-944b-ba08b79ed8a4] State transition from ERROR to PENDING_SHUTDOWN
        2020-06-16 13:22:47.497  INFO 72423 --- [ms-close-thread] o.a.k.s.p.internals.StreamThread         : stream-thread [app-d09c3f52-8d77-4814-944b-ba08b79ed8a4-StreamThread-1] Informed to shut down
        2020-06-16 13:22:47.497  INFO 72423 --- [ms-close-thread] o.a.k.s.p.internals.GlobalStreamThread   : global-stream-thread [app-d09c3f52-8d77-4814-944b-ba08b79ed8a4-GlobalStreamThread] State transition from RUNNING to PENDING_SHUTDOWN
        2020-06-16 13:22:47.557  INFO 72423 --- [balStreamThread] o.a.k.s.p.internals.GlobalStreamThread   : global-stream-thread [app-d09c3f52-8d77-4814-944b-ba08b79ed8a4-GlobalStreamThread] Shutting down
        2020-06-16 13:22:47.571  INFO 72423 --- [balStreamThread] o.a.k.s.p.internals.GlobalStreamThread   : global-stream-thread [app-d09c3f52-8d77-4814-944b-ba08b79ed8a4-GlobalStreamThread] State transition from PENDING_SHUTDOWN to DEAD
        2020-06-16 13:22:47.571  INFO 72423 --- [balStreamThread] o.a.k.s.p.internals.GlobalStreamThread   : global-stream-thread [app-d09c3f52-8d77-4814-944b-ba08b79ed8a4-GlobalStreamThread] Shutdown complete

    @Bean
    public KStream kStream(StreamsBuilder kStreamsBuilder) {
        return kStreamsBuilder.stream("some-topic", Consumed.with(null, null));
    }

这里讨论了一个类似的问题：Kafka Streams 2.5.0 requires input主题看起来这将在kafka-streams 2.5.1和util中修复，那么设置num.stream.threads:0比声明虚拟流更好。

共有1个答案

孙夕

2023-03-14

这似乎与Spring无关，而是由kafka-streams类的一些内部变化引起的。

这在Boot 2.2.x（Kafka-streams 2.3.x）中运行良好。

@SpringBootApplication
@EnableKafkaStreams
public class So62406117Application {

    public static void main(String[] args) {
        SpringApplication.run(So62406117Application.class, args);
    }

    @Bean
    public GlobalKTable<String, String> table(StreamsBuilder kStreamsBuilder) {
        return kStreamsBuilder.globalTable("topic1", Consumed.with(null, null),
                Materialized.as("topic1-store"));
    }

    @Bean
    public ApplicationRunner runner(StreamsBuilderFactoryBean fb) {
        return args -> {
            ReadOnlyKeyValueStore<Object, Object> store =
                    fb.getKafkaStreams().store("topic1-store", QueryableStoreTypes.keyValueStore());
            System.out.println(store);
        };
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1").partitions(1).replicas(1).build();
    }

}

但在boot2.3(kafka-streams2.5.0)时失败。


java.lang.IllegalStateException: Consumer is not subscribed to any topics or assigned any partitions
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1228) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216) ~[kafka-clients-2.5.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:853) ~[kafka-streams-2.5.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:753) ~[kafka-streams-2.5.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697) ~[kafka-streams-2.5.0.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670) ~[kafka-streams-2.5.0.jar:na]

2020-06-16 17:44:02.700  INFO 10635 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [foo-235af8e6-6618-4e73-86ad-75307130004b-StreamThread-1] State transition from STARTING to PENDING_SHUTDOWN
2020-06-16 17:44:02.700  INFO 10635 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [foo-235af8e6-6618-4e73-86ad-75307130004b-StreamThread-1] Shutting down
2020-06-16 17:44:02.700  INFO 10635 --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=foo-235af8e6-6618-4e73-86ad-75307130004b-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
2020-06-16 17:44:02.700  INFO 10635 --- [-StreamThread-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=foo-235af8e6-6618-4e73-86ad-75307130004b-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
2020-06-16 17:44:02.704  INFO 10635 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [foo-235af8e6-6618-4e73-86ad-75307130004b-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD
2020-06-16 17:44:02.704  INFO 10635 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [foo-235af8e6-6618-4e73-86ad-75307130004b] State transition from REBALANCING to ERROR
2020-06-16 17:44:02.704 ERROR 10635 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [foo-235af8e6-6618-4e73-86ad-75307130004b] All stream threads have died. The instance will be in error state and should be closed.
2020-06-16 17:44:02.704  INFO 10635 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [foo-235af8e6-6618-4e73-86ad-75307130004b-StreamThread-1] Shutdown complete

类似资料：

Kafka消费升级到.9

嗨，我正在将kafka升级到.9，并将kafka consumer升级到与.9一起发布的新java consumer。在升级时，我使用的是现有的主题，步骤只是停止.8 kafka并开始指向相同log.dirs的.9 kafka。在消费者端，我使用的是相同的组名和主题名，但是新的消费者从主题中的起始位置再次使用消息。我已经把他们交了。我正在添加auto.offset.reset=最早。任何想法为什
Kafka Streams API：KStream到KTable

我能做到写作和阅读的中间主题：有没有简单的方法从中获取？这是我第一个使用Kafka Streams的应用程序，所以我可能错过了一些明显的东西。
Kafka Streams

streams streams_overview Kafka Streams is a client library for processing and analyzing data stored in Kafka and either write the resulting data back to Kafka or send the final output to an external s
Redis Streams vs Kafka Streams/NAT

Redis团队为Redis 5.0引入了新的Streams数据类型。由于Streams从第一视角看起来像Kafka主题，因此似乎很难找到使用它的真实世界示例。在streams intro中，我们与Kafka streams进行了比较：运行时消费者组处理。例如，如果三个消费者中的一个永久失败，Redis将继续服务第一个和第二个，因为现在我们只有两个逻辑分区（消费者）。 Redis流更快。他们从内
9. Kafka Streams

Kafka Streams 是一个用于处理和分析存储在 Kafka 系统中的数据的客户端库。它建立在重要的流处理概念上，如恰当地区分事件时间（event time）和处理时间（processing time），支持窗口操作（window），exactly-once 处理语义以及简单高效的应用程序状态管理。 Kafka Streams 的入门门槛很低。我们可以在单节点环境上快速实现一个小规模的验证
Kafka到Elasticsearch、HDFS with Logstash或Kafka Streams/Connect

null 在没有任何测试的情况下，我会说第二种选择更好/更干净、更可靠？

升级到Kafka-Streams:5.5.0-CSS（Apache Kafka 2.5.0)[已解决]

共有1个答案

相关问答

相关文章

相关阅读

相关工具

相关文档