2

我正在使用spring KafkaEmbedded test util编写一组测试。每个测试都单独建立一个嵌入式 kafka 实例，产生事件并断言产生的下游事件。

在 IDE（例如 IntelliJ）中运行时，测试始终通过，但是在使用 SBT 运行时，测试会间歇性地失败（大约 50% 的时间，没有规律性）。如果测试失败，我会看到以下错误：

o.a.zookeeper.server.ZooKeeperServer - ZKShutdownHandler is not registered, so ZooKeeper server won't take any action on ERROR or SHUTDOWN server state changes

此外，我看到许多 INFO 日志报告缺少 ZooKeeper 节点，例如：

o.a.z.server.PrepRequestProcessor - Got user-level KeeperException when processing sessionid:0x164a50fe9fd0001 type:create cxid:0x5 zxid:0x4 txntype:-1 reqpath:n/a Error Path:/brokers Error:KeeperErrorCode = NoNode for /brokers

这些日志不会出现在成功的测试中。当我说“我看到很多 INFO 日志”时，我的意思是很多，大约 40 个这样的日志，其中一些节点路径嵌套在先前报告的节点路径中。

研究表明错误日志是无辜的，我想信息日志也是如此，但它们在测试失败时是独立的。

7/17 更新：

KStreams/Producers/Consumers 配置：

Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, appName);
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapUrls);
properties.put(StreamsConfig.STATE_DIR_CONFIG, String.format("/tmp/kafka-streams/%s/%s",
properties.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
properties.put(StreamsConfig.CLIENT_ID_CONFIG, appName);
properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
properties.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, appName);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.cla
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);

嵌入式 Kafka 启动：

@Rule
public KafkaEmbedded kafka = new KafkaEmbedded(getNumKafkaServers(), true, getNumPartitionsPerTopic(), getTopics().keySet().toArray(new String[0]));

测试前：

@Before
public void before() {
    // build KStreams and start topology
    KStreamBuilder kStreamBuilder = new KStreamBuilder();
    buildStream(kStreamBuilder);
    KafkaStreams kafkaStreams = new KafkaStreams(kStreamBuilder, streamsConfig);
    kafkaStreams.start();
}

测试后：

@After
public void after() {
    kafka.destroy();
    FileUtils.deleteDirectory(new File(streamsConfig.getString(StreamsConfig.STATE_DIR_CONFIG)));
}

7/17 更新：

更多上下文，这是一个 Spring 项目，每个测试都这样注释：

@SpringBootTest(classes = <this-test-class>.class)
@ActiveProfiles("test")
@RunWith(SpringRunner.class)
4

0 回答 0