ConsumeService: fix client close() causing ConcurrentModificationExeption

[mhratson]

Problem

• calling _baseConsumer.close(), outside of the thread the consumer is running in, is invalid
• as docummented in kafka consumer docs¹

The Kafka consumer is NOT thread-safe. All network I/O happens in the thread of the application * making the call. It is the responsibility of the user to ensure that multi-threaded access * is properly synchronized. Un-synchronized access will result in ConcurrentModificationException`.

The exception thrown

2021/01/25 23:10:25.961 WARN [ConsumeService] [Thread-1] [kafka-monitoring] [] kac-lc/ConsumeService while trying to close consumer.
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access. competing thread is kac-lc consume-service
        at com.linkedin.kafka.clients.utils.KafkaConsumerLock.lock(KafkaConsumerLock.java:31) ~[li-apache-kafka-clients-1.0.59.jar:?]
        at com.linkedin.kafka.clients.utils.CloseableLock.<init>(CloseableLock.java:18) ~[li-apache-kafka-clients-1.0.59.jar:?]
        at com.linkedin.kafka.clients.consumer.LiKafkaInstrumentedConsumerImpl.close(LiKafkaInstrumentedConsumerImpl.java:716) ~[li-apache-kafka-clients-1.0.59.jar:?]

Solution

The recommended solution¹ is

• to use consumer.wakeup(); method
• but the method is not yet adopted by the KMBaseConsumer interface
• so for now _baseConsumer.close() is moved into the thread
• calling stop now only sets _running.compareAndSet(true, false), so the runloop exits

Testing Done

Increased thread.join(5000) timeout as this implementation is slower to stop due to not interrupting the consumer thread.

- ./gradlew test

Footnotes

1. KafkaConsumer.java ↩ ↩²