Repository: provectus/kafka-ui
Issue: Message tab infinite polling
Labels: type/bug
scope/backend
status/accepted
status/confirmed

<!--

Don't forget to check for existing issues/discussions regarding your proposal. We might already have it.
https://github.com/provectus/kafka-ui/issues
https://github.com/provectus/kafka-ui/discussions

-->

**Describe the bug**
<!--(A clear and concise description of what the bug is.)-->

After latest pull from master today, kafka-ui broke and no events are being displayed in ui. Switch to Kafka Cluster > topics > navigate to topic. Topic overview tab shows correct information. 

When navigate to messages tab, none of the events displayed in data grid and result is:
Polling 0 ms 0 Bytes 0 messages consumed

From docker logs, error is:
2022-11-15 00:06:24,483 ERROR [boundedElastic-4] r.c.s.Schedulers: Scheduler worker in group main failed with an uncaught exception
java.lang.NoClassDefFoundError: Could not initialize class org.xerial.snappy.pure.UnsafeUtil

The complete error stack is below.


**Set up**
<!--
How do you run the app? Please provide as much info as possible:
1. App version (docker image version or check commit hash in the top left corner in UI)
2. Helm chart version, if you use one
3. Any IAAC configs

We might close the issue without further explanation if you don't provide such information.
-->
docker pull provectuslabs/kafka-ui:master

11.14.2022 11:03:06
(
[8181499](https://github.com/provectus/kafka-ui/commit/8181499)
)


**Steps to Reproduce**
<!-- We'd like you to provide an example setup (via docker-compose, helm, etc.) 
to reproduce the problem, especially with a complex setups. -->
Steps to reproduce the behavior:

1.  docker pull provectuslabs/kafka-ui:master

**Expected behavior**
<!--
(A clear and concise description of what you expected to happen)
-->

Events from kafka topics to be displayed in data grid.

**Screenshots**
<!--
(If applicable, add screenshots to help explain your problem)
-->


**Additional context**
<!--
(Add any other context about the problem here)
-->

2022-11-15 00:06:24,224 INFO  [boundedElastic-4] o.a.k.c.Metadata: [Consumer clientId=kafka-ui-25f7d8b6-b232-4643-b483-40b0df3f28e7, groupId=null] Cluster ID: Zp6dCNprQFWXYfHx6rwlkw
2022-11-15 00:06:24,274 INFO  [boundedElastic-4] o.a.k.c.c.KafkaConsumer: [Consumer clientId=kafka-ui-25f7d8b6-b232-4643-b483-40b0df3f28e7, groupId=null] Subscribed to partition(s): TEST-TOPIC_V3-1, TEST-TOPIC_V3-2, TEST-TOPIC_V3-0
2022-11-15 00:06:24,275 INFO  [boundedElastic-4] o.a.k.c.c.KafkaConsumer: [Consumer clientId=kafka-ui-25f7d8b6-b232-4643-b483-40b0df3f28e7, groupId=null] Seeking to offset 131339 for partition TEST-TOPIC_V3-1
2022-11-15 00:06:24,275 INFO  [boundedElastic-4] o.a.k.c.c.KafkaConsumer: [Consumer clientId=kafka-ui-25f7d8b6-b232-4643-b483-40b0df3f28e7, groupId=null] Seeking to offset 134131 for partition TEST-TOPIC_V3-2
2022-11-15 00:06:24,275 INFO  [boundedElastic-4] o.a.k.c.c.KafkaConsumer: [Consumer clientId=kafka-ui-25f7d8b6-b232-4643-b483-40b0df3f28e7, groupId=null] Seeking to offset 132305 for partition TEST-TOPIC_V3-0
2022-11-15 00:06:24,480 INFO  [boundedElastic-4] o.a.k.c.m.Metrics: Metrics scheduler closed
2022-11-15 00:06:24,480 INFO  [boundedElastic-4] o.a.k.c.m.Metrics: Closing reporter org.apache.kafka.common.metrics.JmxReporter
2022-11-15 00:06:24,480 INFO  [boundedElastic-4] o.a.k.c.m.Metrics: Metrics reporters closed
2022-11-15 00:06:24,482 INFO  [boundedElastic-4] o.a.k.c.u.AppInfoParser: App info kafka.consumer for kafka-ui-25f7d8b6-b232-4643-b483-40b0df3f28e7 unregistered
2022-11-15 00:06:24,483 WARN  [boundedElastic-4] r.c.Exceptions: throwIfFatal detected a jvm fatal exception, which is thrown and logged below:
java.lang.NoClassDefFoundError: Could not initialize class org.xerial.snappy.pure.UnsafeUtil
        at org.xerial.snappy.pure.SnappyRawDecompressor.getUnsignedByteSafe(SnappyRawDecompressor.java:323)
        at org.xerial.snappy.pure.SnappyRawDecompressor.readUncompressedLength(SnappyRawDecompressor.java:288)
        at org.xerial.snappy.pure.SnappyRawDecompressor.getUncompressedLength(SnappyRawDecompressor.java:42)
        at org.xerial.snappy.pure.PureJavaSnappy.uncompressedLength(PureJavaSnappy.java:238)
        at org.xerial.snappy.Snappy.uncompressedLength(Snappy.java:638)
        at org.xerial.snappy.SnappyInputStream.hasNextChunk(SnappyInputStream.java:435)
        at org.xerial.snappy.SnappyInputStream.read(SnappyInputStream.java:466)
        at java.base/java.io.DataInputStream.readByte(DataInputStream.java:271)
        at org.apache.kafka.common.utils.ByteUtils.readUnsignedVarint(ByteUtils.java:170)
        at org.apache.kafka.common.utils.ByteUtils.readVarint(ByteUtils.java:205)
        at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:281)
        at org.apache.kafka.common.record.DefaultRecordBatch$2.doReadRecord(DefaultRecordBatch.java:296)
        at org.apache.kafka.common.record.DefaultRecordBatch$StreamRecordIterator.readNext(DefaultRecordBatch.java:638)
        at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:603)
        at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:572)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.nextFetchedRecord(Fetcher.java:1618)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1653)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1900(Fetcher.java:1494)
        at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:716)
        at org.apache.kafka.clients.consumer.internals.Fetcher.collectFetch(Fetcher.java:682)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1313)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1242)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1215)
        at com.provectus.kafka.ui.emitter.AbstractEmitter.poll(AbstractEmitter.java:40)
        at com.provectus.kafka.ui.emitter.AbstractEmitter.poll(AbstractEmitter.java:34)
        at com.provectus.kafka.ui.emitter.ForwardRecordEmitter.accept(ForwardRecordEmitter.java:45)
        at com.provectus.kafka.ui.emitter.ForwardRecordEmitter.accept(ForwardRecordEmitter.java:14)
        at reactor.core.publisher.FluxCreate.subscribe(FluxCreate.java:95)
        at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62)
        at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194)
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
2022-11-15 00:06:24,483 WARN  [boundedElastic-4] r.c.Exceptions: throwIfFatal detected a jvm fatal exception, which is thrown and logged below:
java.lang.NoClassDefFoundError: Could not initialize class org.xerial.snappy.pure.UnsafeUtil
        at org.xerial.snappy.pure.SnappyRawDecompressor.getUnsignedByteSafe(SnappyRawDecompressor.java:323)
        at org.xerial.snappy.pure.SnappyRawDecompressor.readUncompressedLength(SnappyRawDecompressor.java:288)
        at org.xerial.snappy.pure.SnappyRawDecompressor.getUncompressedLength(SnappyRawDecompressor.java:42)
        at org.xerial.snappy.pure.PureJavaSnappy.uncompressedLength(PureJavaSnappy.java:238)
        at org.xerial.snappy.Snappy.uncompressedLength(Snappy.java:638)
        at org.xerial.snappy.SnappyInputStream.hasNextChunk(SnappyInputStream.java:435)
        at org.xerial.snappy.SnappyInputStream.read(SnappyInputStream.java:466)
        at java.base/java.io.DataInputStream.readByte(DataInputStream.java:271)
        at org.apache.kafka.common.utils.ByteUtils.readUnsignedVarint(ByteUtils.java:170)
        at org.apache.kafka.common.utils.ByteUtils.readVarint(ByteUtils.java:205)
        at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:281)
        at org.apache.kafka.common.record.DefaultRecordBatch$2.doReadRecord(DefaultRecordBatch.java:296)
        at org.apache.kafka.common.record.DefaultRecordBatch$StreamRecordIterator.readNext(DefaultRecordBatch.java:638)
        at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:603)
        at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:572)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.nextFetchedRecord(Fetcher.java:1618)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1653)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1900(Fetcher.java:1494)
        at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:716)
        at org.apache.kafka.clients.consumer.internals.Fetcher.collectFetch(Fetcher.java:682)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1313)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1242)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1215)
        at com.provectus.kafka.ui.emitter.AbstractEmitter.poll(AbstractEmitter.java:40)
        at com.provectus.kafka.ui.emitter.AbstractEmitter.poll(AbstractEmitter.java:34)
        at com.provectus.kafka.ui.emitter.ForwardRecordEmitter.accept(ForwardRecordEmitter.java:45)
        at com.provectus.kafka.ui.emitter.ForwardRecordEmitter.accept(ForwardRecordEmitter.java:14)
        at reactor.core.publisher.FluxCreate.subscribe(FluxCreate.java:95)
        at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62)
        at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194)
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
2022-11-15 00:06:24,483 ERROR [boundedElastic-4] r.c.s.Schedulers: Scheduler worker in group main failed with an uncaught exception
java.lang.NoClassDefFoundError: Could not initialize class org.xerial.snappy.pure.UnsafeUtil
        at org.xerial.snappy.pure.SnappyRawDecompressor.getUnsignedByteSafe(SnappyRawDecompressor.java:323)
        at org.xerial.snappy.pure.SnappyRawDecompressor.readUncompressedLength(SnappyRawDecompressor.java:288)
        at org.xerial.snappy.pure.SnappyRawDecompressor.getUncompressedLength(SnappyRawDecompressor.java:42)
        at org.xerial.snappy.pure.PureJavaSnappy.uncompressedLength(PureJavaSnappy.java:238)
        at org.xerial.snappy.Snappy.uncompressedLength(Snappy.java:638)
        at org.xerial.snappy.SnappyInputStream.hasNextChunk(SnappyInputStream.java:435)
        at org.xerial.snappy.SnappyInputStream.read(SnappyInputStream.java:466)
        at java.base/java.io.DataInputStream.readByte(DataInputStream.java:271)
        at org.apache.kafka.common.utils.ByteUtils.readUnsignedVarint(ByteUtils.java:170)
        at org.apache.kafka.common.utils.ByteUtils.readVarint(ByteUtils.java:205)
        at org.apache.kafka.common.record.DefaultRecord.readFrom(DefaultRecord.java:281)
        at org.apache.kafka.common.record.DefaultRecordBatch$2.doReadRecord(DefaultRecordBatch.java:296)
        at org.apache.kafka.common.record.DefaultRecordBatch$StreamRecordIterator.readNext(DefaultRecordBatch.java:638)
        at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:603)
        at org.apache.kafka.common.record.DefaultRecordBatch$RecordIterator.next(DefaultRecordBatch.java:572)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.nextFetchedRecord(Fetcher.java:1618)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1653)
        at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1900(Fetcher.java:1494)
        at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:716)
        at org.apache.kafka.clients.consumer.internals.Fetcher.collectFetch(Fetcher.java:682)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1313)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1242)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1215)
        at com.provectus.kafka.ui.emitter.AbstractEmitter.poll(AbstractEmitter.java:40)
        at com.provectus.kafka.ui.emitter.AbstractEmitter.poll(AbstractEmitter.java:34)
        at com.provectus.kafka.ui.emitter.ForwardRecordEmitter.accept(ForwardRecordEmitter.java:45)
        at com.provectus.kafka.ui.emitter.ForwardRecordEmitter.accept(ForwardRecordEmitter.java:14)
        at reactor.core.publisher.FluxCreate.subscribe(FluxCreate.java:95)
        at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62)
        at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194)
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
2022-11-15 00:06:25,123 DEBUG [parallel-3] c.p.k.u.s.ClustersStatisticsScheduler: Metrics updated for cluster: KAFKA_QA


Issue comments:
@nitin1677 thank you for issue!
updated docker image published (master tag) 
can you please check on your setup?
Updated the docker image with pull from master. No more error in docker logs. 

1. Topics without schema (Key serde: String, Value serde: String, Partitions: All items are selected) are able to pull the events and display in data grid.

2. Topics with schema (Key serde: SchemaRegistry, Value serde: SchemaRegistry, Partitions: All items are selected) are not able to pull the events and result in data grid is, No messages found (2997 ms, 0 Bytes, 0 messages consumed)
    2.1 From partitions list, select a partition (Partition#0) and data grid will display the events from select partition, 123 ms, 663 
    KB, 500 messages consumed.

Result from docker logs when all items are selected for partitions:
2022-11-15 13:33:09,908 INFO  [boundedElastic-5] o.a.k.c.c.KafkaConsumer: [Consumer clientId=kafka-ui-92fe475e-27e5-4c74-8381-46b86892c0bb, groupId=null] Seeking to offset 654588 for partition TEST-TOPIC_V6-39
2022-11-15 13:33:09,908 INFO  [boundedElastic-5] o.a.k.c.c.KafkaConsumer: [Consumer clientId=kafka-ui-92fe475e-27e5-4c74-8381-46b86892c0bb, groupId=null] Seeking to offset 704717 for partition TEST-TOPIC_V6-8
2022-11-15 13:33:10,908 INFO  [boundedElastic-5] c.p.k.u.e.ForwardRecordEmitter: 0 records polled
2022-11-15 13:33:11,914 INFO  [boundedElastic-5] c.p.k.u.e.ForwardRecordEmitter: 0 records polled
2022-11-15 13:33:12,916 INFO  [boundedElastic-5] c.p.k.u.e.ForwardRecordEmitter: 0 records polled
2022-11-15 13:33:12,917 INFO  [boundedElastic-5] c.p.k.u.e.ForwardRecordEmitter: Polling finished
2022-11-15 13:33:12,917 INFO  [boundedElastic-5] o.a.k.c.m.Metrics: Metrics scheduler closed
2022-11-15 13:33:12,917 INFO  [boundedElastic-5] o.a.k.c.m.Metrics: Closing reporter org.apache.kafka.common.metrics.JmxReporter
2022-11-15 13:33:12,917 INFO  [boundedElastic-5] o.a.k.c.m.Metrics: Metrics reporters closed



@nitin1677 thank you for checking
looks like snappy issue is solved, as for 2. that you mention - looks like it is separate issue, lets create ticket for that.

To finally verify that snappy works fine - can you please run topic analysis (Statistics tab) on both SR and non-SR topic?
Yes, the snappy error is resolved and UI is functional to retrieve the events in data grid.

Ran topics analysis successfully for both Schema-Enabled topics and Topics without Schema.
@nitin1677 thank you for confirmation!

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/Dockerfile b/kafka-ui-api/Dockerfile
index c0c68ff0041..96e0eb414b1 100644
--- a/kafka-ui-api/Dockerfile
+++ b/kafka-ui-api/Dockerfile
@@ -1,5 +1,6 @@
 FROM azul/zulu-openjdk-alpine:17
 
+RUN apk add --no-cache gcompat # need to make snappy codec work
 RUN addgroup -S kafkaui && adduser -S kafkaui -G kafkaui
 
 USER kafkaui


Test patch: