confluentinc / confluentinc/parallel-consumer
Question: Is it possible to produce events using reactor?
- 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