You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Murilo Tavares (Jira)" <ji...@apache.org> on 2020/02/06 19:58:00 UTC
[jira] [Updated] (KAFKA-9518) NullPointerException on out-of-order
topologies
[ https://issues.apache.org/jira/browse/KAFKA-9518?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Murilo Tavares updated KAFKA-9518:
----------------------------------
Description:
I have a KafkaStreams that dinamically builds a topology based on a Map of input-to-output topics. Since the map was not sorted, iteration was unpredictable, and different instances could have different orders. When this happen, KafkaStreams throws an exception during REBALANCE.
I was able to reproduce this using the attached java project. The project is a pretty simple Maven project with one class. It starts 2 instances in parallel, with the same input-to-output topics, but one instance takes the topics in a reversed order.
The exception is this:
{noformat}
Exception in thread "MY-APP-81e9dc0b-1459-4499-93d6-b5c03da60e18-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: stream-thread [MY-APP-81e9dc0b-1459-4499-93d6-b5c03da60e18-StreamThread-1] Failed to rebalance. at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:852) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:743) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:698) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:671) Caused by: java.lang.NullPointerException at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:234) at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:176) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:355) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:313) at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:298) at org.apache.kafka.streams.processor.internals.TaskManager.addNewActiveTasks(TaskManager.java:160) at org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:120) at org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsAssigned(StreamsRebalanceListener.java:77) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsAssigned(ConsumerCoordinator.java:272) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:400) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:421) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:340) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:471) at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1267) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1231) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:843) ... 3 more{noformat}
was:
I have a KafkaStreams that dinamically builds a topology based on a Map of input-to-output topics. Since the map was not sorted, iteration was unpredictable, and different instances could have different orders. When this happen, KafkaStreams throws an exception during REBALANCE.
I was able to reproduce this using the attached java project. The project is a pretty simple Maven project with one class. It starts 2 instances in parallel, with the same input-to-output topics, but one instance takes the topics in a reversed order.
The exception is this:
{noformat}
*no* further _formatting_ is done here{noformat}
Exception in thread "MY-APP-81e9dc0b-1459-4499-93d6-b5c03da60e18-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: stream-thread [MY-APP-81e9dc0b-1459-4499-93d6-b5c03da60e18-StreamThread-1] Failed to rebalance. at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:852) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:743) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:698) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:671) Caused by: java.lang.NullPointerException at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:234) at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:176) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:355) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:313) at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:298) at org.apache.kafka.streams.processor.internals.TaskManager.addNewActiveTasks(TaskManager.java:160) at org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:120) at org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsAssigned(StreamsRebalanceListener.java:77) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsAssigned(ConsumerCoordinator.java:272) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:400) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:421) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:340) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:471) at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1267) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1231) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:843) ... 3 more
> NullPointerException on out-of-order topologies
> -----------------------------------------------
>
> Key: KAFKA-9518
> URL: https://issues.apache.org/jira/browse/KAFKA-9518
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.4.0, 2.3.1
> Reporter: Murilo Tavares
> Priority: Minor
> Attachments: kafka-streams-testing.zip
>
>
> I have a KafkaStreams that dinamically builds a topology based on a Map of input-to-output topics. Since the map was not sorted, iteration was unpredictable, and different instances could have different orders. When this happen, KafkaStreams throws an exception during REBALANCE.
>
> I was able to reproduce this using the attached java project. The project is a pretty simple Maven project with one class. It starts 2 instances in parallel, with the same input-to-output topics, but one instance takes the topics in a reversed order.
>
> The exception is this:
> {noformat}
> Exception in thread "MY-APP-81e9dc0b-1459-4499-93d6-b5c03da60e18-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: stream-thread [MY-APP-81e9dc0b-1459-4499-93d6-b5c03da60e18-StreamThread-1] Failed to rebalance. at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:852) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:743) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:698) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:671) Caused by: java.lang.NullPointerException at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:234) at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:176) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:355) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:313) at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:298) at org.apache.kafka.streams.processor.internals.TaskManager.addNewActiveTasks(TaskManager.java:160) at org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:120) at org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsAssigned(StreamsRebalanceListener.java:77) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsAssigned(ConsumerCoordinator.java:272) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:400) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:421) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:340) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:471) at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1267) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1231) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:843) ... 3 more{noformat}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)