KafkaMessageFilteringConsumer#findMessageWithTimeout concurrent access

[tschlat]

Citrus Version
4.4

Expected behavior
KafkaMessageFilteringConsumer should poll messages without running into exceptions.

Actual behavior
Consumer runs into the following exception:

org.citrusframework.exceptions.TestCaseFailedException: KafkaConsumer is not safe for multi-threaded access. currentThread(name: main, id: 1) otherThread(id: 50)

The Reason is, that the conusmer is queried asynchronously here:
KafkaMessageFilteringConsumer#findMessageWithTimeout:

Future<List<ConsumerRecord<Object, Object>>> handler = executorService.submit(() -> {
    return this.findMessagesSatisfyingMatcher(topic);
});

and then here from the main thread:

        } finally {
            executorService.shutdownNow();
            this.consumer.unsubscribe();
        }

When the consumber is unsubscribed, it is still locked by the asynchonous thread. So before unsubscription, we need a sync here to make sure the consumer is not in use any more.

Test case sample

Please, share the test case (as small as possible) which shows the issue

[bbortt]

thanks @tschlat. I will fix it by next week.

[bbortt]

Just wanted to note this down here...

When the consumber is unsubscribed, it is still locked by the asynchonous thread. So before unsubscription, we need a sync here to make sure the consumer is not in use any more.

That's not the source of the problem (although the order of shutdown might be wrong in this case). There are numerous threads and only one consumer which is not thread-safe. I've fixed this in the KafkaConsumer class of citrus. But I need some more minutes to double-check backwards compatibility.