Dynamically connecting a Kafka input stream to multiple output streams

by

If you want to create topics dynamically based on your data, you do not get any support within Kafka’s Streaming API at the moment (v0.10.2 and earlier). You will need to create a KafkaProducer and implement your dynamic “routing” by yourself (for example using KStream#foreach() or KStream#process()). Note, that you need to do synchronous writes to avoid data loss (which are not very performant unfortunately). There are plans to extend Streaming API with dynamic topic routing, but there is no concrete timeline for this feature right now.

There is one more consideration you should take into account. If you do not know your destination topic(s) ahead of time and just rely on the so-called “topic auto creation” feature, you should make sure that those topics are being created with the desired configuration settings (e.g., number of partitions or replication factor).

As an alternative to “topic auto creation” you can also use Admin Client (available since v0.10.1) to create topics with correct configuration. See https://cwiki.apache.org/confluence/display/KAFKA/KIP-4+-+Command+line+and+centralized+administrative+operations

More Related Contents:

  1. Handling exceptions in Kafka streams
  2. Is it possible to access message headers with Kafka Streams?
  3. Connect to Kafka running in Docker
  4. How can I send large messages with Kafka (over 15MB)?
  5. How to create a Topic in Kafka through Java
  6. How Can we create a topic in Kafka from the IDE using API
  7. What determines Kafka consumer offset?
  8. Writing Custom Kafka Serializer
  9. Java, How to get number of messages in a topic in apache kafka
  10. How to Delete a topic in apache kafka [duplicate]
  11. When does the Apache Kafka client throw a “Batch Expired” exception?
  12. What is a simple, effective way to debug custom Kafka connectors?
  13. Is it possible to add partitions to an existing topic in Kafka 0.8.2
  14. How to get kafka consume lag in java program
  15. How to create Custom serializer in kafka?
  16. Kafka Consumer CommitFailedException
  17. Can a Kafka consumer(0.8.2.2) read messages in batch
  18. How can I delete a topic in Apache Kafka? [duplicate]
  19. Is it 2 character search possible in lucence
  20. Replace implicit wait with explicit wait (selenium webdriver & java)
  21. Is it a bad practice to catch Throwable?
  22. Sorting hashmap based on keys
  23. Image/Graphic into a Shape
  24. Setting Android images from string value
  25. How to check a string against null in java?
  26. Find out which classes of a given API are used
  27. Java – Rotating array
  28. String to Date Conversion mm/dd/yy to YYYY-MM-DD in java [duplicate]
  29. How to assign bean’s property an Enum value in Spring config file?
  30. JSP : JSTL’s tag

Leave a Comment