Styjun

Does norwegian.no airline overbook flights?

Which note goes on which side of the stem?

Start from ones

Is “I am getting married with my sister” ambiguous?

Average period of peer review process

What is the hex versus octal timeline?

Is using a hyperlink to close a modal a poor design decision?

Why does trim() NOT remove char 160?

Science fiction short story where aliens contact a drunk about Earth's impending destruction

Did it used to be possible to target a zone?

Why different interest rates for checking and savings?

See details of old sessions

What to say to a student who has failed?

Prove your innocence

What are some interesting features that are common cross-linguistically but don't exist in English?

What magic extends life or grants immortality?

Does travel insurance for short flight delays exist?

Why don't electrons take the shorter path in coils?

Why were movies shot on film shot at 24 frames per second?

Fried gnocchi with spinach, bacon, cream sauce in a single pan

In the MCU, why does Mjölnir retain its enchantments after Ragnarok?

Why can't an Airbus A330 dump fuel in an emergency?

Numbers Decrease while Letters Increase

C++20 constexpr std::copy optimizations for run-time

Kafka - Redirect messages from “Topic A” to “Topic B” based on header value

Zookeeper having KeeperException but Kafka able to create topics and produce/consumeKafka - Docker - Error when sending message from Host to Container (Batch Expired)Problems to consume from kafka created by docker-composeGetting error while consuming kafka messagespring-cloud-starter-stream-kafka does not creates topic provided to spring.boot.cloud.stream.bindings.output.destinationKafka consumer not able to consume messages using bootstrap server nameUnable to access topic from scaled Kafka cluster in DockerHow to expose kafka endpoint from azure and consume it from onpremise application?Spark Streaming not reading from Kafka topics

.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty margin-bottom:0;

0

I would like to redirect kafka messages from a topic called "all-topic" to a topic named "headervalue-topic" where headervalue is the value of a custom header each message has.

At the moment i'm using a custom console application that consumes messages and redirects the messages to the correct topic, but it only process 16 messages per second.

Both kafka and zookeeper are running in a docker container, configured as such :

zookeeper:
 image: "wurstmeister/zookeeper:latest"
 restart: always
 ports:
 - "2181:2181"
 environment:
 ZOOKEEPER_CLIENT_PORT: 2181
 ZOOKEEPER_SERVER_ID: 1

kafka:
 hostname: kafka
 image: "wurstmeister/kafka:latest"
 restart: always
 depends_on:
 - zookeeper
 ports:
 - "9092:9092"
 environment:
 KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
 KAFKA_ADVERTISED_HOST_NAME: kafka
 KAFKA_ADVERTISED_PORT: 9092

What is the best and fastest way to achieve my goal?

I do know about the existence of Kafka Streams, but i'm not familiar with Java so in case you'd like to suggest Kafka Streams a little example would be appreciated :)

Many Thanks!

apache-kafka apache-kafka-streams

share|improve this question

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  And yes, you need to write a Kafka Stream application to achieve this. There are lots of example of Kafka stream app available at kafka.apache.org/documentation/streams
  
  – JR ibkr
  Mar 27 at 19:48
  

  
  
  
  

add a comment | 

0

I would like to redirect kafka messages from a topic called "all-topic" to a topic named "headervalue-topic" where headervalue is the value of a custom header each message has.

At the moment i'm using a custom console application that consumes messages and redirects the messages to the correct topic, but it only process 16 messages per second.

Both kafka and zookeeper are running in a docker container, configured as such :

zookeeper:
 image: "wurstmeister/zookeeper:latest"
 restart: always
 ports:
 - "2181:2181"
 environment:
 ZOOKEEPER_CLIENT_PORT: 2181
 ZOOKEEPER_SERVER_ID: 1

kafka:
 hostname: kafka
 image: "wurstmeister/kafka:latest"
 restart: always
 depends_on:
 - zookeeper
 ports:
 - "9092:9092"
 environment:
 KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
 KAFKA_ADVERTISED_HOST_NAME: kafka
 KAFKA_ADVERTISED_PORT: 9092

What is the best and fastest way to achieve my goal?

I do know about the existence of Kafka Streams, but i'm not familiar with Java so in case you'd like to suggest Kafka Streams a little example would be appreciated :)

Many Thanks!

apache-kafka apache-kafka-streams

share|improve this question

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  And yes, you need to write a Kafka Stream application to achieve this. There are lots of example of Kafka stream app available at kafka.apache.org/documentation/streams
  
  – JR ibkr
  Mar 27 at 19:48
  

  
  
  
  

add a comment | 

I would like to redirect kafka messages from a topic called "all-topic" to a topic named "headervalue-topic" where headervalue is the value of a custom header each message has.

At the moment i'm using a custom console application that consumes messages and redirects the messages to the correct topic, but it only process 16 messages per second.

Both kafka and zookeeper are running in a docker container, configured as such :

zookeeper:
 image: "wurstmeister/zookeeper:latest"
 restart: always
 ports:
 - "2181:2181"
 environment:
 ZOOKEEPER_CLIENT_PORT: 2181
 ZOOKEEPER_SERVER_ID: 1

kafka:
 hostname: kafka
 image: "wurstmeister/kafka:latest"
 restart: always
 depends_on:
 - zookeeper
 ports:
 - "9092:9092"
 environment:
 KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
 KAFKA_ADVERTISED_HOST_NAME: kafka
 KAFKA_ADVERTISED_PORT: 9092

What is the best and fastest way to achieve my goal?

I do know about the existence of Kafka Streams, but i'm not familiar with Java so in case you'd like to suggest Kafka Streams a little example would be appreciated :)

Many Thanks!

apache-kafka apache-kafka-streams

share|improve this question

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

zookeeper:
 image: "wurstmeister/zookeeper:latest"
 restart: always
 ports:
 - "2181:2181"
 environment:
 ZOOKEEPER_CLIENT_PORT: 2181
 ZOOKEEPER_SERVER_ID: 1

kafka:
 hostname: kafka
 image: "wurstmeister/kafka:latest"
 restart: always
 depends_on:
 - zookeeper
 ports:
 - "9092:9092"
 environment:
 KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
 KAFKA_ADVERTISED_HOST_NAME: kafka
 KAFKA_ADVERTISED_PORT: 9092

share|improve this question

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

share|improve this question

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

edited Mar 27 at 20:15

Robin Moffatt

10.5k1818 silver badges3131 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

asked Mar 27 at 16:43

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

1 Answer
1

active

oldest

votes

0

Here is the solution i came up with, using kafka-streams nodejs library :

const KafkaStreams = require("kafka-streams");
const nativeConfig: config = require("./config.js");

const kafkaStreams = new KafkaStreams(config);
const myConsumerStream = kafkaStreams.getKStream("all-topic");

myConsumerStream
 .mapJSONConvenience()
 .filter((element) => 
 return element.value.type == "Article";
 )
 .tap((element) => console.log("Got Article"))
 .mapWrapKafkaValue()
 .to("Article-topic", 1, "buffer");

myConsumerStream.start();

share|improve this answer

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

add a comment | 

Your Answer

StackExchange.ifUsing("editor", function ()
StackExchange.using("externalEditor", function ()
StackExchange.using("snippets", function ()
StackExchange.snippets.init();
);
);
, "code-snippets");

StackExchange.ready(function()
var channelOptions =
tags: "".split(" "),
id: "1"
;
initTagRenderer("".split(" "), "".split(" "), channelOptions);

StackExchange.using("externalEditor", function()
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled)
StackExchange.using("snippets", function()
createEditor();
);

else
createEditor();

);

function createEditor()
StackExchange.prepareEditor(
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader:
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
,
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
);



);

draft saved

draft discarded

Sign up or log in

StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);

Sign up using Google

Sign up using Facebook

Sign up using Email and Password

Post as a guest

Name

Email

Required, but never shown

StackExchange.ready(
function ()
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f55382449%2fkafka-redirect-messages-from-topic-a-to-topic-b-based-on-header-value%23new-answer', 'question_page');

);

Post as a guest

Name

Email

Required, but never shown

0

Here is the solution i came up with, using kafka-streams nodejs library :

const KafkaStreams = require("kafka-streams");
const nativeConfig: config = require("./config.js");

const kafkaStreams = new KafkaStreams(config);
const myConsumerStream = kafkaStreams.getKStream("all-topic");

myConsumerStream
 .mapJSONConvenience()
 .filter((element) => 
 return element.value.type == "Article";
 )
 .tap((element) => console.log("Got Article"))
 .mapWrapKafkaValue()
 .to("Article-topic", 1, "buffer");

myConsumerStream.start();

share|improve this answer

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

add a comment | 

0

Here is the solution i came up with, using kafka-streams nodejs library :

const KafkaStreams = require("kafka-streams");
const nativeConfig: config = require("./config.js");

const kafkaStreams = new KafkaStreams(config);
const myConsumerStream = kafkaStreams.getKStream("all-topic");

myConsumerStream
 .mapJSONConvenience()
 .filter((element) => 
 return element.value.type == "Article";
 )
 .tap((element) => console.log("Got Article"))
 .mapWrapKafkaValue()
 .to("Article-topic", 1, "buffer");

myConsumerStream.start();

share|improve this answer

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

add a comment | 

Here is the solution i came up with, using kafka-streams nodejs library :

const KafkaStreams = require("kafka-streams");
const nativeConfig: config = require("./config.js");

const kafkaStreams = new KafkaStreams(config);
const myConsumerStream = kafkaStreams.getKStream("all-topic");

myConsumerStream
 .mapJSONConvenience()
 .filter((element) => 
 return element.value.type == "Article";
 )
 .tap((element) => console.log("Got Article"))
 .mapWrapKafkaValue()
 .to("Article-topic", 1, "buffer");

myConsumerStream.start();

share|improve this answer

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

const KafkaStreams = require("kafka-streams");
const nativeConfig: config = require("./config.js");

const kafkaStreams = new KafkaStreams(config);
const myConsumerStream = kafkaStreams.getKStream("all-topic");

myConsumerStream
 .mapJSONConvenience()
 .filter((element) => 
 return element.value.type == "Article";
 )
 .tap((element) => console.log("Got Article"))
 .mapWrapKafkaValue()
 .to("Article-topic", 1, "buffer");

myConsumerStream.start();

share|improve this answer

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges

answered Mar 29 at 14:31

LorenzoScebbaLorenzoScebba

3911 silver badge55 bronze badges