Kafka LZ4 decompression causes IllegalAccessError in lz4-java when reading from topic

user_16526675 0
Viewed 252

On the project I am working on currently, we have four spring based apps that are writing and reading from various Kafka topics. Due to the large amount of data being sent, we have decided to enable LZ4 compression. This worked perfectly fine for three of the apps, but caused a huge issue in the fourth.

The flow is as follows: App A receives HTTP requests and converts the payload to Kafka messages. App B reads the messages from app A, does some processing on them, and sends them further on two different topics to apps C and D. We enabled compression on the producers in both A and B by setting the compression type config to lz4. We then tested, and it worked fine for the messages from A to B, and also from B to D. However, when app C received a compressed message, the following error started being spammed:

java.lang.IllegalAccessError: failed to access class net.jpountz.lz4.LZ4SafeUtils (class loader System@7452) from class net.jpountz.lz4.LZ4JavaSafeSafeDecompressor (class loader org.springframework.boot.loader.LaunchedURLClassLoader@11104)

I am not very familiar with classloaders, but it seems to me that although they belong to the same package, they seem to be loaded by different class loaders causing the access error?

The version of spring-kafka being used is 2.7.3, and the version of lz4-java is 1.7.1, imported as a dependency of spring-kafka.

I do not understand why this issue is occurring only on one of the apps and not on the others, as all the configs are the same (shared and loaded via a commons library used by all four apps to create required beans). Initially, in this app that presents this issue, we had another library being imported that used a modified version of lz4-java, so we thought that's where the issue is coming from. However, even after completely removing it, the same error posted above is received.

Has anyone experienced a similar issue or could give me a lead as to what to do next to solve this issue?

Thank you.

Edit: The application is running in the cloud and this is the stack trace as seen in the application logs. The following block is spammed repeatedly until it is shut down or the compressed messages are removed from the topics that are being listened to.

[container-1-C-1] essageListenerContainer$ListenerConsumer : Consumer exception
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT org.springframework.kafka.KafkaException: Seek to current after exception; nested exception is org.apache.kafka.common.KafkaException: Received exception when fetching the next record from ETD-NewUserContextSystemData.ibsoetdcfsubacc4.eb419dde-795b-48a2-977a-8117ac15cb4e-2. If needed, please seek past the record to continue consumption.
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler.handle(SeekToCurrentBatchErrorHandler.java:79) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.RecoveringBatchErrorHandler.handle(RecoveringBatchErrorHandler.java:124) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1603) [spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1210) [spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_301]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_301]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at java.lang.Thread.run(Thread.java:836) [?:1.8.0_301]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT Caused by: org.apache.kafka.common.KafkaException: Received exception when fetching the next record from ETD-NewUserContextSystemData.ibsoetdcfsubacc4.eb419dde-795b-48a2-977a-8117ac15cb4e-2. If needed, please seek past the record to continue consumption.
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1611) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1432) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:684) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:635) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1303) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1237) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1410) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1249) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1161) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     ... 3 more
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT Caused by: org.apache.kafka.common.KafkaException: java.lang.IllegalAccessError: failed to access class net.jpountz.lz4.LZ4SafeUtils (class loader System@7452) from class net.jpountz.lz4.LZ4JavaSafeSafeDecompressor (class loader org.springframework.boot.loader.LaunchedURLClassLoader@11104)
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.CompressionType$4.wrapForInput(CompressionType.java:113) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.DefaultRecordBatch.compressedIterator(DefaultRecordBatch.java:261) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.DefaultRecordBatch.streamingIterator(DefaultRecordBatch.java:346) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.nextFetchedRecord(Fetcher.java:1554) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1591) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1432) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:684) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:635) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1303) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1237) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1410) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1249) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1161) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     ... 3 more
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT Caused by: java.lang.IllegalAccessError: failed to access class net.jpountz.lz4.LZ4SafeUtils (class loader System@7452) from class net.jpountz.lz4.LZ4JavaSafeSafeDecompressor (class loader org.springframework.boot.loader.LaunchedURLClassLoader@11104)
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4JavaSafeSafeDecompressor.decompress(LZ4JavaSafeSafeDecompressor.java:71) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4SafeDecompressor.decompress(LZ4SafeDecompressor.java:74) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4Factory.<init>(LZ4Factory.java:214) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4Factory.instance(LZ4Factory.java:51) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4Factory.safeInstance(LZ4Factory.java:105) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4Factory.fastestJavaInstance(LZ4Factory.java:141) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at net.jpountz.lz4.LZ4Factory.fastestInstance(LZ4Factory.java:169) ~[lz4-java-1.7.1.jar:2.9.16-38a800d98fb18a51ecde14f570bbf28d81b66bbc]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.KafkaLZ4BlockInputStream.<clinit>(KafkaLZ4BlockInputStream.java:50) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.CompressionType$4.wrapForInput(CompressionType.java:110) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.DefaultRecordBatch.compressedIterator(DefaultRecordBatch.java:261) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.common.record.DefaultRecordBatch.streamingIterator(DefaultRecordBatch.java:346) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.nextFetchedRecord(Fetcher.java:1554) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1591) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1432) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:684) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:635) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1303) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1237) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210) ~[kafka-clients-2.6.2.jar:?]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1410) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1249) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1161) ~[spring-kafka-2.7.3.jar:2.7.3]
   2021-09-24T10:42:29.30+0300 [APP/PROC/WEB/1] OUT     ... 3 more
lz4 spring-kafka
1 Answers

I was facing the same issue like you. I faced the issue when running my program on Linux system. The root cause of the issue might be: your Linux OS kernel does not support method calling from .so file (not sure if you are using Linux or IBM AIX). The .so file is automatically loaded when lz4 jar file is loaded into the classpath based on the OS kernel version.

You can run the command line "uname -a" or "uname -mrs" to check the OS kernel version.

If you open the lz4 jar file, you will see .so files for each OS under "util" package. My thought is that your (and my) OS kernel was not added in the support list of the library. We might need to check the library's Jira issue list to see if they confirm that this is an issue and they are going to fix it.

user_22475670
Related