Update
To provide some context - the Kafka topic is enabled with LZ4 compression.
When the compression type is recognized, the CompressionType enum tries to associate LZ4 with the KafkaLZ4BlockInputStream which it could find for some reason during runtime.
@Override
public InputStream wrapForInput(ByteBuffer inputBuffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
try {
return new KafkaLZ4BlockInputStream(inputBuffer, decompressionBufferSupplier,
messageVersion == RecordBatch.MAGIC_VALUE_V0);
} catch (Throwable e) {
throw new KafkaException(e);
}
}
Original Post
I have a JAR contains a Flink job to consume records from Kafka.
After submitting the JAR file to Flink, the task manager says that KafkaLZ4BlockInputStream class is missing:
Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:150)
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: org.apache.kafka.common.KafkaException: Received exception when fetching the next record from flight-cache-3. If needed, please seek past the record to continue consumption.
at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1553)
at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1374)
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:676)
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:631)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1313)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1240)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211)
at org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader.fetch(KafkaPartitionSplitReader.java:113)
at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)
... 6 more
Caused by: org.apache.kafka.common.KafkaException: java.lang.NoClassDefFoundError: Could not initialize class org.apache.kafka.common.record.KafkaLZ4BlockInputStream
at org.apache.kafka.common.record.CompressionType$4.wrapForInput(CompressionType.java:113)
at org.apache.kafka.common.record.DefaultRecordBatch.compressedIterator(DefaultRecordBatch.java:261)
at org.apache.kafka.common.record.DefaultRecordBatch.streamingIterator(DefaultRecordBatch.java:346)
at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.nextFetchedRecord(Fetcher.java:1496)
at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1533)
... 15 more
Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.kafka.common.record.KafkaLZ4BlockInputStream
at org.apache.kafka.common.record.CompressionType$4.wrapForInput(CompressionType.java:110)
... 19 more
I unzipped the JAR file to a folder and found that the class does exist:
$ ls -l org/apache/kafka/common/record/KafkaLZ4BlockInputStream.class
-rw-r--r-- 1 wxh wxh 6489 Mar 2 2020 org/apache/kafka/common/record/KafkaLZ4BlockInputStream.class
I can also confirm org.apache.kafka:kafka-clients:jar:2.4.1 that contains the class does exist in the dependency tree.
+- org.apache.flink:flink-core:jar:1.14.0:compile
| +- org.apache.flink:flink-annotations:jar:1.14.0:compile
| +- org.apache.flink:flink-metrics-core:jar:1.14.0:compile
| +- org.apache.flink:flink-shaded-asm-7:jar:7.1-14.0:compile
| +- org.apache.commons:commons-lang3:jar:3.3.2:compile
| +- com.esotericsoftware.kryo:kryo:jar:2.24.0:compile
| | +- com.esotericsoftware.minlog:minlog:jar:1.2:compile
| | \- org.objenesis:objenesis:jar:2.1:compile
| +- commons-collections:commons-collections:jar:3.2.2:compile
| +- org.apache.commons:commons-compress:jar:1.21:compile
| +- org.apache.flink:flink-shaded-guava:jar:30.1.1-jre-14.0:compile
| +- org.slf4j:slf4j-api:jar:1.7.15:compile
| +- com.google.code.findbugs:jsr305:jar:1.3.9:compile
| \- org.apache.flink:flink-shaded-force-shading:jar:14.0:compile
+- org.apache.flink:flink-clients_2.11:jar:1.14.0:compile
| +- org.apache.flink:flink-runtime:jar:1.14.0:compile
| | +- org.apache.flink:flink-rpc-core:jar:1.14.0:compile
| | +- org.apache.flink:flink-rpc-akka-loader:jar:1.14.0:compile
| | +- org.apache.flink:flink-queryable-state-client-java:jar:1.14.0:compile
| | +- org.apache.flink:flink-hadoop-fs:jar:1.14.0:compile
| | +- commons-io:commons-io:jar:2.8.0:compile
| | +- org.apache.flink:flink-shaded-netty:jar:4.1.65.Final-14.0:compile
| | +- org.apache.flink:flink-shaded-jackson:jar:2.12.4-14.0:compile
| | +- org.apache.flink:flink-shaded-zookeeper-3:jar:3.4.14-14.0:compile
| | +- org.javassist:javassist:jar:3.24.0-GA:compile
| | +- org.xerial.snappy:snappy-java:jar:1.1.8.3:compile
| | \- org.lz4:lz4-java:jar:1.8.0:compile
| +- org.apache.flink:flink-optimizer:jar:1.14.0:compile
| +- org.apache.flink:flink-java:jar:1.14.0:compile
| | \- org.apache.commons:commons-math3:jar:3.5:compile
| +- commons-cli:commons-cli:jar:1.3.1:compile
| \- org.apache.flink:flink-streaming-java_2.11:jar:1.14.0:compile
| \- org.apache.flink:flink-scala_2.11:jar:1.14.0:compile
| +- org.scala-lang:scala-reflect:jar:2.11.12:compile
| +- org.scala-lang:scala-library:jar:2.11.12:compile
| +- org.scala-lang:scala-compiler:jar:2.11.12:compile
| | +- org.scala-lang.modules:scala-xml_2.11:jar:1.0.5:compile
| | \- org.scala-lang.modules:scala-parser-combinators_2.11:jar:1.0.4:compile
| \- com.twitter:chill_2.11:jar:0.7.6:compile
| \- com.twitter:chill-java:jar:0.7.6:compile
+- org.apache.flink:flink-connector-files:jar:1.14.0:compile
| +- org.apache.flink:flink-file-sink-common:jar:1.14.0:compile
| \- org.apache.flink:flink-connector-base:jar:1.14.0:compile
+- org.apache.flink:flink-connector-kafka_2.11:jar:1.14.0:compile
| \- org.apache.kafka:kafka-clients:jar:2.4.1:compile # <--- IT DOES EXISTS
| \- com.github.luben:zstd-jni:jar:1.4.3-1:compile
+- org.projectlombok:lombok:jar:1.18.22:provided
Anyone has any experience with this strange issue?
Thanks!