[jira] [Commented] (KAFKA-9504) Memory leak in KafkaMetrics registered to MBean
[
https://issues.apache.org/jira/browse/KAFKA-9504?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17033663#comment-17033663
]
Murilo Tavares commented on KAFKA-9504:
---
I just ran the above against version 2.3.1, and it looks fine on that release.
So apparently this was really introduced on 2.4.0.
> Memory leak in KafkaMetrics registered to MBean
> ---
>
> Key: KAFKA-9504
> URL: https://issues.apache.org/jira/browse/KAFKA-9504
> Project: Kafka
> Issue Type: Bug
> Components: clients
>Affects Versions: 2.4.0
>Reporter: Andreas Holmén
>Priority: Major
>
> After close() called on a KafkaConsumer some registered MBeans are not
> unregistered causing leak.
>
>
> {code:java}
> import static
> org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
> import java.lang.management.ManagementFactory;
> import java.util.HashMap;
> import java.util.Map;
> import javax.management.MBeanServer;
> import org.apache.kafka.clients.consumer.KafkaConsumer;
> import org.apache.kafka.common.serialization.ByteArrayDeserializer;
> public class Leaker {
> private static String bootstrapServers = "hostname:9092";
>
> public static void main(String[] args) throws InterruptedException {
> MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
> Map props = new HashMap<>();
> props.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
>
> int beans = mBeanServer.getMBeanCount();
> for (int i = 0; i < 100; i++) {
>KafkaConsumer consumer = new KafkaConsumer<>(props, new
> ByteArrayDeserializer(), new ByteArrayDeserializer());
>consumer.close();
> }
> int newBeans = mBeanServer.getMBeanCount();
> System.out.println("\nbeans delta: " + (newBeans - beans));
> }
> }
> {code}
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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}
The topology for both instances:
{code:java}
// instance1
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [topicA])
--> KSTREAM-SINK-01
Sink: KSTREAM-SINK-01 (topic: topicA-repartitioned)
<-- KSTREAM-SOURCE-00
Sub-topology: 1
Source: KSTREAM-SOURCE-02 (topics: [topicB])
--> KSTREAM-SINK-03
Sink: KSTREAM-SINK-03 (topic: topicB-repartitioned)
<-- KSTREAM-SOURCE-02
// instance2
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [topicB])
--> KSTREAM-SINK-01
Sink: KSTREAM-SINK-01 (topic: topicB-repartitioned)
<-- KSTREAM-SOURCE-00
Sub-topology: 1
Source: KSTREAM-SOURCE-02 (topics: [topicA])
--> KSTREAM-SINK-03
Sink: KSTREAM-SINK-03 (topic: topicA-repartitioned)
<-- KSTREAM-SOURCE-02{code}
In my actual project, I fixed the issue by sorting the topics map accordingly,
but it would be nice to have at least a better error handling in this case.
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}
Exception in thread
"MY-APP-81e9dc0b-1459-4499-93d6-b5c03da
[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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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}
{code:java}
The topology for both instances:
// instance1
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [topicA])
--> KSTREAM-SINK-01
Sink: KSTREAM-SINK-01 (topic: topicA-repartitioned)
<-- KSTREAM-SOURCE-00
Sub-topology: 1
Source: KSTREAM-SOURCE-02 (topics: [topicB])
--> KSTREAM-SINK-03
Sink: KSTREAM-SINK-03 (topic: topicB-repartitioned)
<-- KSTREAM-SOURCE-02
// instance2
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-00 (topics: [topicB])
--> KSTREAM-SINK-01
Sink: KSTREAM-SINK-01 (topic: topicB-repartitioned)
<-- KSTREAM-SOURCE-00
Sub-topology: 1
Source: KSTREAM-SOURCE-02 (topics: [topicA])
--> KSTREAM-SINK-03
Sink: KSTREAM-SINK-03 (topic: topicA-repartitioned)
<-- KSTREAM-SOURCE-02{code}
In my actual project, I fixed the issue by sorting the topics map accordingly,
but it would be nice to have at least a better error handling in this case.
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}
Exception in thread
"MY-APP-81e9dc0b-1459-4499-93d6-b5c03da6
[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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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}
In my actual project, I fixed the issue by sorting the topics map accordingly,
but it would be nice to have at least a better error handling in this case.
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}
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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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.creat
[jira] [Created] (KAFKA-9518) NullPointerException on out-of-order topologies
Murilo Tavares created KAFKA-9518:
-
Summary: 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.3.1, 2.4.0
Reporter: Murilo Tavares
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}
*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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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.(StreamTask.java:234)
at
org.apache.kafka.streams.processor.internals.StreamTask.(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$AbstractTaskCrea
