KafkaListener OffsetStrategy.DISABLED Manual Acknowledgement Failure (Reactive Type)

[brianwyka]

With a @KafkaListener that has offset strategy diabled, when using the reactive approach, the Acknowledgement.ack() produces an exception when called within the @KafkaListener.

@KafkaListener(groupId = "archive-upload", offsetStrategy = OffsetStrategy.DISABLED)
public class ArchiveUploadMessageProcessor {

    private final ArchiveService archiveService;

    @Topic("${kafka-topics.archive-upload}")
    public Single<String> processMessage(@Body final PayloadArchiveUploadMessage payloadArchiveUploadMessage, final Acknowledgement acknowledgement) {
        val payloadKey = payloadArchiveUploadMessage.getPayloadRequest().getPayloadKey();
        val payloadKeyString = PayloadKeyFormatter.asString(payloadKey);
        return archive(payloadArchiveUploadMessage, payloadKeyString, acknowledgement)
                .doOnSuccess(payloadUrl -> acknowledgement.ack());
    }

Below is the exception:

java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

My assumption is the error is triggered from the lambda implementation of Acknowledgement below in KafkaConsumerProcessor:

micronaut-kafka/kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java

Lines 488 to 492 in 0d9f891

	if (ackArg.isPresent()) {
	boundArguments.put(ackArg.get(), (KafkaAcknowledgement) () -> kafkaConsumer.commitSync(
	currentOffsets
	));
	}

Relevant consumer thread configuration:

micronaut:
  executors:
    consumer:
      type: fixed
      nThreads: ${KAFKA_CONSUMER_THREAD_COUNT:4}

[brianwyka]

Perhaps its happening in a different thread since its chained in the reactive stream?

[brianwyka]

After reading around the documentation a bit, I was able to find this:

Alternatively, you can use the @Blocking annotation to tell Micronaut to subscribe to the returned reactive type in a blocking manner which will result in blocking the poll loop, preventing offsets from being committed automatically:

After adding the @Blocking annotation, its working as expected, however, kind of defeats the purpose of using reactive style of programming in the first place. Any way this can be achieved without @Blocking annotation on the kafka listener?

[ctoestreich]

@brianwyka Not sure that it will change the exception, but are you trying to bee reactive on the way IN or the way OUT or both? Just writing pseudo code here from your above example.

 public void processMessage(@Body final Single<PayloadArchiveUploadMessage> payloadArchiveUploadMessage, final Acknowledgement c) {
            payloadArchiveUploadMessage.doOnSuccess((payload) -> {
            val payloadKey = payload.getPayloadRequest().getPayloadKey();
            val payloadKeyString = PayloadKeyFormatter.asString(payloadKey);
            archive(payload, payloadKeyString) //call out to the archive process?  Is this reative too?
            acknowledgement.ack();
       });
    }

Also should you synchronize the method?

[graemerocher]

Kafka consumers are inherently single threaded so you have to always acknowledge the message on the same thread as the listener executed on. That means waiting for all other publishers to complete

[brianwyka]

@graemerocher is there any way to bind a method argument which would be a callback to the publishers being completed which is a Supplier<KafkaConsumer> or Supplier<Acknowledgement> that way it can be executed at the appropriate time and in the correct thread? Or perhaps, a configuration on the annotation to ackOnSuccess ?

[brianwyka]

@ctoestreich I am trying to be reactive on the way out, such that Micronaut treats it as a non-blocking subscription. I could also take the @Body parameter as a Single as well. Not sure what the benefit of that is, however... The archive method call inside the method is also reactive, so the example above would not work since it would never be subscribed to...

[brianwyka]

Thinking about it more, seems like it would be non-trivial to have some sort of callback (in reactive style) which executes in the same thread as the polling loop...