java - Kafka 流关闭并且不运行

Question

大家早上好

我正在尝试运行 Kafka Stream 应用程序，但每次我尝试时，它都会按顺序启动和关闭。下面是控制台打印的结果

[main] WARN org.apache.kafka.clients.consumer.ConsumerConfig - The configuration 'admin.retries' was supplied but isn't a known config.
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 2.1.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : eec43959745f444f
[application-brute-test-client-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] Starting
[main] INFO org.apache.kafka.streams.KafkaStreams - stream-client [application-brute-test-client] Started Streams client
[application-brute-test-client-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] State transition from CREATED to RUNNING
[Thread-0] INFO org.apache.kafka.streams.KafkaStreams - stream-client [application-brute-test-client] State transition from RUNNING to PENDING_SHUTDOWN
[kafka-streams-close-thread] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] Informed to shut down
[kafka-streams-close-thread] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
[application-brute-test-client-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] Shutting down
[application-brute-test-client-StreamThread-1] INFO org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=application-brute-test-client-StreamThread-1-restore-consumer, groupId=] Unsubscribed all topics or patterns and assigned partitions
[application-brute-test-client-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD
[application-brute-test-client-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [application-brute-test-client-StreamThread-1] Shutdown complete
[kafka-admin-client-thread | application-brute-test-client-admin] INFO org.apache.kafka.clients.admin.internals.AdminMetadataManager - [AdminClient clientId=application-brute-test-client-admin] Metadata update failed
org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call.
[kafka-streams-close-thread] INFO org.apache.kafka.streams.KafkaStreams - stream-client [application-brute-test-client] State transition from PENDING_SHUTDOWN to NOT_RUNNING
[Thread-0] INFO org.apache.kafka.streams.KafkaStreams - stream-client [application-brute-test-client] Streams client stopped completely

注意以下行:

[application-brute-test-client-StreamThread-1] Informed to shut down

应用程序被通知关闭，但我不知道为什么。有人可以帮我解决这个问题吗？

这是我的简单代码，仅用于测试流:

Properties properties = new Properties();
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "myserver");
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "application-brute-test");
properties.put(StreamsConfig.CLIENT_ID_CONFIG, "application-brute-test-client");
properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE); // Enable exacly once feature
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); // Set a default key serde
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); // Set a default key serde

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> input = builder.stream("neurotech_propostas", Consumed.with(Serdes.String(), Serdes.String()));
input.print(Printed.toSysOut());
KStream<String, String> output = input.mapValues((value) -> value.toUpperCase());

output.to("brute-test-out");

KafkaStreams stream = new KafkaStreams(builder.build(), properties);
stream.cleanUp();
stream.start();

Runtime.getRuntime().addShutdownHook(new Thread(stream::close));

Answer 1

为了解决这个问题，我只是停止使用 JUnit 来运行 Stream 并通过 Main 类执行。通过 JUnit 运行 Kafka Streams 导致了这个问题。

也许在这种环境下，JUnit 不保持线程执行？