Kafka Stream 启动问题 - org.apache.kafka.streams.errors.LockException

Question

我有一个 Kafka Streams 应用程序版本 - 0.11，它从几个主题中获取数据并将数据连接起来并将其放入另一个主题中。

卡夫卡配置：

5 kafka brokers - version 0.11
Kafka Topics - 15 partitions and 3 replication factor.

每小时消耗/产生几百万条记录。每当我关闭任何 kafka 代理时，它都会抛出异常：

org.apache.kafka.streams.errors.LockException: task [4_10] Failed to lock the state directory for task 4_10
    at org.apache.kafka.streams.processor.internals.ProcessorStateManager.<init>(ProcessorStateManager.java:99)
    at org.apache.kafka.streams.processor.internals.AbstractTask.<init>(AbstractTask.java:80)
    at org.apache.kafka.streams.processor.internals.StandbyTask.<init>(StandbyTask.java:62)
    at org.apache.kafka.streams.processor.internals.StreamThread.createStandbyTask(StreamThread.java:1325)
    at org.apache.kafka.streams.processor.internals.StreamThread.access$2400(StreamThread.java:73)
    at org.apache.kafka.streams.processor.internals.StreamThread$StandbyTaskCreator.createTask(StreamThread.java:313)
    at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:254)
    at org.apache.kafka.streams.processor.internals.StreamThread.addStandbyTasks(StreamThread.java:1366)
    at org.apache.kafka.streams.processor.internals.StreamThread.access$1200(StreamThread.java:73)
    at org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:185)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:265)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:363)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:310)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:297)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
    at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:582)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:553)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527)

我已经阅读了几个 jira 问题，清理流可能有助于解决问题。但是每次启动 Kafka 流应用程序时清理流是正确的解决方案还是补丁？还有，stream cleanUp 会延迟应用启动对吧？

注意：每次启动 Kafka Streams 应用程序时，是否需要在调用streams.start() 之前调用streams.cleanUp()

Answer 1

看到org.apache.kafka.streams.errors.LockException: task [4_10] Failed to lock the state directory for task 4_10 实际上是预期的，应该会自行解决。线程将退出以等待另一个线程释放锁并稍后重试。因此，您甚至可能会多次看到此 WARN 消息是日志，以防在第二个线程释放锁之前发生重试。

但是，最终锁应该由第二个线程释放，第一个线程将能够获得锁。之后，Streams 应该继续前进。请注意，这是一条 WARN 消息，而不是错误。