Uploaded image for project: 'Kafka'
  1. Kafka
  2. KAFKA-9518

NullPointerException on out-of-order topologies

    XMLWordPrintableJSON

Details

    • Bug
    • Status: Open
    • Minor
    • Resolution: Unresolved
    • 1.1.1, 2.4.0, 2.3.1
    • None
    • streams
    • None

    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:

      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

       

      The topology for both instances:

      // instance1
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [topicA])
      --> KSTREAM-SINK-0000000001
    Sink: KSTREAM-SINK-0000000001 (topic: topicA-repartitioned)
      <-- KSTREAM-SOURCE-0000000000


  Sub-topology: 1
    Source: KSTREAM-SOURCE-0000000002 (topics: [topicB])
      --> KSTREAM-SINK-0000000003
    Sink: KSTREAM-SINK-0000000003 (topic: topicB-repartitioned)
      <-- KSTREAM-SOURCE-0000000002

// instance2
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [topicB])
      --> KSTREAM-SINK-0000000001
    Sink: KSTREAM-SINK-0000000001 (topic: topicB-repartitioned)
      <-- KSTREAM-SOURCE-0000000000


  Sub-topology: 1
    Source: KSTREAM-SOURCE-0000000002 (topics: [topicA])
      --> KSTREAM-SINK-0000000003
    Sink: KSTREAM-SINK-0000000003 (topic: topicA-repartitioned)
      <-- KSTREAM-SOURCE-0000000002

      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.

      Attachments

        1. kafka-streams-testing.zip
          7 kB
          Murilo Tavares

        Issue Links

          relates to

          Improvement - An improvement or enhancement to an existing feature or task. KAFKA-8307 Kafka Streams should provide some mechanism to determine topology equality and compatibility

          • Major - Major loss of function.
          • Open

          Improvement - An improvement or enhancement to an existing feature or task. KAFKA-8810 Add mechanism to detect topology mismatch between streams instances

          • Major - Major loss of function.
          • Open

          Wish - General wishlist item. KAFKA-7669 Stream topology definition is not robust to the ordering changes

          • Major - Major loss of function.
          • Open

          Activity

            People

              Unassigned Unassigned
              murilofla Murilo Tavares
              Votes:
              0 Vote for this issue
              Watchers:
              6 Start watching this issue

              Dates

                Created:
                Updated: