confluentinc / confluentinc/parallel-consumer

Question: Is it possible to produce events using reactor?

Open
#480 8 comments 0 reactions 0 assignees View on GitHub
enhancement not-a-bug question
Dominant language
Java
Stars
299
Forks
172
PR merge metrics
No merged PRs in 30d

Description

Hi, I am working with `ReactorProcessor` and I would like to produce events. I tried the following code, but it did not write any new events into **_output_** topic.
Example code:

`lateinit var pConsumer: ReactorProcessor`

```
private fun createReactParallelConsumer(): ReactorProcessor {
val producer: Producer = KafkaProducerBuilder.getProducer(kafkaConsumerConfig)
val options = ParallelConsumerOptions.builder()
.ordering(ParallelConsumerOptions.ProcessingOrder.KEY)
.maxConcurrency(parallelConsumerConfig.maxConcurrency)
// .batchSize(parallelConsumerConfig.batchSize)
.consumer(buildConsumer(kafkaConsumerConfig))
.producer(producer)
.build()
return ReactorProcessor(options)
```

```
pConsumer.react { context ->
val event = context.singleConsumerRecord
// do something with event
val result = ProducerRecord("output", event.key(),
JsonObject(mapOf("someTest" to event.offset())))
Mono.just(result)
}
```

What am i missing?

Contributor guide

No contributing guide indexed for this repository

Research direction

The issue names ReactorProcessor.react, ProducerRecord, and the output topic but no repository file or test. Start by tracing the shown react callback and producer configuration; done means reproducing the missing output record and documenting or fixing the identified cause.

Written by the indexing model from the issue text.

Assessment

Tech stack
kafka, kotlin
Domain
distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.