Kumar Mallikarjuna created FLINK-40348:
------------------------------------------
Summary: ZSTD fails at runtime with the SQL Kafka Connector
Key: FLINK-40348
URL: https://issues.apache.org/jira/browse/FLINK-40348
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka
Reporter: Kumar Mallikarjuna
The {{flink-sql-connector-kafka}} uber jar bundles the Kafka classes that
_invoke_
the compression codecs, but none of the codec libraries themselves. Any table
using
{{'properties.compression.type' = 'zstd'}} fails at runtime.
The same connector consumed as a plain Maven dependency works: {{kafka-clients}}
declares {{zstd-jni}} as a mandatory ({{{}optional=false{}}}) runtime
dependency, so a
DataStream user gets it transitively. Only the SQL uber jar drops it. Since
that jar
exists to be the self-contained equivalent of the Maven dependency, the two
should not
diverge on which codecs work.
h2. Cause
The uber jar's shade configuration uses an include whitelist:
{code:xml}
<artifactSet>
<includes>
<include>org.apache.flink:flink-connector-kafka</include>
<include>org.apache.kafka:*</include>
</includes>
</artifactSet>
{code}
{{kafka-clients}} declares its three compression providers at {{runtime}} scope:
* {{com.github.luben:zstd-jni}}
* {{org.lz4:lz4-java}}
* {{org.xerial.snappy:snappy-java}}
An include whitelist is exhaustive, so none of the three are bundled, while
{{org.apache.kafka.common.compress.ZstdCompression}} (and {{ZstdFactory}} on
older lines), which reference them, are.
h2. Symptom
{code:java}
java.lang.NoClassDefFoundError: com/github/luben/zstd/BufferPool
at
org.apache.kafka.common.compress.ZstdCompression$Builder.build(ZstdCompression.java:138)
at
org.apache.kafka.clients.producer.KafkaProducer.configureCompression(KafkaProducer.java:559)
at
org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:410)
Caused by: java.lang.ClassNotFoundException: com.github.luben.zstd.BufferPool
{code}
On Kafka clients >= 3.8 this throws during producer construction.
h2. Why this only appears for zstd
* gzip: JDK provides
* snappy and lz4: {{flink-dist}} bundles them
* zstd: Flink doesn't need zstd, so zstd is the only codec with no provider
h2. Are codecs expected to be provided at runtime?
No such contract in the docs and Kafka expects this to be handled differently:
*1. Kafka declares them mandatory.* In {{kafka-clients}} 3.4.0/3.9.1/4.2.0, all
three (minus gzip) codecs are {{scope=runtime}} with {{{}optional=false{}}}. An
optional-false runtime dependency is delivered transitively to every Maven
consumer. {_}Had Kafka intended them to be user-supplied it would mark them
{{optional=true}}{_}.
*2. The same connector behaves differently depending only on packaging.* The
non-shaded {{org.apache.flink:flink-connector-kafka}} declares
{{kafka-clients}} at {{{}compile{}}}/{{{}optional=false{}}}, so a DataStream
user who depends on it via Maven gets {{zstd-jni}} transitively and ZSTD works.
_A SQL user who drops the uber jar into {{lib/}} gets the same connector code
with the codec silently removed._ The uber jar
exists precisely to be the self-contained equivalent of that Maven dependency,
so the
two should not diverge on which compression codecs function.
*3. Nothing in the build or the docs expresses such a contract.* The codecs are
lost
by omission from an include whitelist, not by an explicit {{{}<exclude>{}}}.
There is no
comment, and no reference to zstd/luben in any pom in the repository. _The
connector_
_documentation tells users to add the connector jar; it does not mention codec
jars._
h2. Steps to reproduce
# Start a Flink cluster and SQL client with {{flink-sql-connector-kafka}} in
{{{}lib/{}}}.
# Create a table against any Kafka topic with
{{'properties.compression.type' = 'zstd'}} in the WITH clause.
# {{INSERT INTO}} it. The job fails with the {{NoClassDefFoundError}} above.
h2. Proposed fix
Add the codec provider(s) to the shade {{{}<includes>{}}}, {*}unrelocated{*}:
{code:xml}
<include>com.github.luben:zstd-jni</include>
{code}
Two constraints:
* It must not be relocated. The shaded Kafka bytecode references the original
{{com.github.luben.zstd}} package, and zstd-jni's JNI loader resolves its native
library off that package name, relocating breaks it twice over.
* Flink's {{NoticeFileChecker}} requires bundled dependencies to be listed, so
{{META-INF/NOTICE}} needs the corresponding entry.
h3. Trade-off
{{zstd-jni}} is ~7 MB because it ships prebuilt natives for ~10 platforms. That
more
than doubles the 5.3 MB connector jar. Options:
# Bundle {{zstd-jni}} whole.
# Document that users must supply codec jars themselves. However, this
contradicts having an "uber" jar for SQL (users do not assemble a classpath).
h3. On lz4 and snappy
Bundle zstd only, the others are currently being shipped already with the Flink
build.
h2. Context
The whitelist pattern dates to FLINK-11026, which reworked SQL connector jar
creation.
It is correct for the compile-scope graph; the {{{}runtime{}}}-scope codec
providers were
simply never in scope for that change.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
