PC 0.5.2.5 - pollAndProduce with zero message result ?

[mikhelef]

Hello,

Is it possible to use pollAndProduce method and for some records not producing messages/ProducerRecords to the broker ?
I've read in the documentation: [8.3.2. Process and Produce a Response Message] that this interface allows you to process your message, then publish back to the broker zero, one or more result messages. but how ?
Thank you for your help.

Regards,
Massi

[mikhelef]

Hello,

Any update on this ? how to not produce producer records to output for some input consumer records ?

Thanks in advance,
Massi

[rkolesnev]

Hi @mikhelef,

Sorry - i've missed the original question - you can use pollAndProduceMany instead.
Return type for it is List - and you can return empty list when you don't want to produce a record.
As a quick example :

    parallelConsumer.pollAndProduceMany(context -> {
                    var consumerRecord = context.getSingleRecord().getConsumerRecord();
                    Optional<Result> optionalResult = processBrokerRecord(consumerRecord);
                    if(optionalResult.isEmpty()) {
                        return Collections.emptyList();
                    }else {
                        Result result = optionalResult.get();
                        return Collections.singletonList(new ProducerRecord<>(outputTopic, consumerRecord.key(), result.payload));
                    }
                }, consumeProduceResult -> {
                    log.debug("Message {} saved to broker at offset {}",
                            consumeProduceResult.getOut(),
                            consumeProduceResult.getMeta().offset());
                }
        );

or with Java streams / functional interfaces

    parallelConsumer.pollAndProduceMany(context -> {
                    var consumerRecord = context.getSingleRecord().getConsumerRecord();
                    Optional<Result> optionalResult = processBrokerRecord(consumerRecord);
                    return optionalResult.map(result-> Collections.singletonList(new ProducerRecord<>(outputTopic, consumerRecord.key(), result.payload)))
                            .orElse(Collections.emptyList());
                }, consumeProduceResult -> {
                    log.debug("Message {} saved to broker at offset {}",
                            consumeProduceResult.getOut(),
                            consumeProduceResult.getMeta().offset());
                }
        );

[mikhelef]

Hello,

Thank you for your reply!

That's exactly what I have implemented and it works perfectly.

Regards,
Massi