我们需要开发一个代码，让消费者在运行时监听特定的 kafka 生产者，然后在同一函数中将处理后的数据从当前消费的生产者生成到不同的生产者主题。
这是为了将 flinks 代码与 Java 集成，其中 Java 向一个主题生成一条消息，flink 使用它并为另一个主题生成一个新数据，供 Java 进一步处理。
如果有其他方法可以执行此过程，请告诉我们。
【问题讨论】：
标签： scala apache-kafka apache-flink kafka-consumer-api kafka-producer-api
这听起来像是“读-处理-写”模式。你可以利用 Kafka 的transaction functionality 来使这个过程原子化（或者不原子，这取决于你，但下面的例子使用了事务）：
KafkaProducer producer = createKafkaProducer(
"bootstrap.servers", "localhost:9092",
"transactional.id", "my-transactional-id");
producer.initTransactions();
KafkaConsumer consumer = createKafkaConsumer(
"bootstrap.servers", "localhost:9092",
"group.id", "my-group-id",
"isolation.level", "read_committed");
consumer.subscribe(singleton("inputTopic"));
while (true) {
ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
producer.beginTransaction();
for (ConsumerRecord record : records)
producer.send(producerRecord("outputTopic", record));
producer.sendOffsetsToTransaction(currentOffsets(consumer), group);
producer.commitTransaction();
}
这是为了将 flinks 代码与 Java 集成，其中 Java 向一个主题生成一条消息，flink 使用它并为另一个主题生成一个新数据，供 Java 进一步处理。
您可能需要考虑 Kafka Streams：https://docs.confluent.io/current/streams/developer-guide/index.html
【讨论】：
这听起来像是使用akka streams 的好地方，
val done =
Consumer.committableSource(consumerSettings, Subscriptions.topics("topic1"))
.map(msg => ProducerMessage.Message(new ProducerRecord[Array[Byte], String]("topic2", msg.record.value), msg.committableOffset))
.via(Producer.flow(producerSettings))
.map(_.message.passThrough)
.batch(max = 20, first => CommittableOffsetBatch.empty.updated(first)) { (batch, elem) =>
batch.updated(elem)
}
.mapAsync(3)(_.commitScaladsl())
.runWith(Sink.ignore)
【讨论】：
Flink integrates nicely with Kafka，如果需要，可以利用 Kafka 事务。这样的应用程序看起来像这样：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkKafkaConsumer011<> consumer = new FlinkKafkaConsumer011<IN>(topic_in, serializer_in, kafkaProperties);
FlinkKafkaProducer011<> producer = new FlinkKafkaProducer011<OUT>(broker, topic_out, serializer_out)
env.addSource(consumer)
.map(new SuitableTransformation())
.addSink(producer)
.execute()
【讨论】：