[
https://issues.apache.org/jira/browse/FLINK-40348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113608#comment-18113608
]
Bhanu Chander Vallabaneni commented on FLINK-40348:
---------------------------------------------------
I looked into this and can confirm the runtime failure, but the include
whitelist named above cannot be fixed by adding the three artifacts to it. Two
independent blockers, both verified against {{main}}.
First, evidence that the jar really is missing them. Building
{{flink-sql-connector-kafka}} from {{main}} and inspecting the shaded jar:
{code}
org/apache/flink/kafka/shaded/org/apache/kafka/common/compress/ZstdCompression.class
present
com/github/luben/zstd/**
0 entries
org/xerial/snappy/**
0 entries
net/jpountz/**
0 entries
*.so / *.dylib / *.dll
0 entries
{code}
{{ZstdCompression}}'s bytecode references
{{com/github/luben/zstd/ZstdOutputStreamNoFinalizer}},
{{ZstdInputStreamNoFinalizer}}, {{RecyclingBufferPool}} and {{BufferPool}},
none of which ship. So the caller is bundled and the callee is not, exactly as
described.
h2. Why only zstd fails in practice
{{flink-runtime}} declares {{org.xerial.snappy:snappy-java}} and
{{at.yawk.lz4:lz4-java}} at {{compile}} scope, so flink-dist puts them on every
cluster's classpath and Kafka's {{SnappyCompression}} /
{{Lz4BlockOutputStream}} resolve against those. {{zstd-jni}} is not among them.
The connector uber jar is therefore not self-contained for *any* codec; it
happens to work for two of the three because flink-dist supplies them.
h2. Why adding zstd-jni to the whitelist does not work
*Relocating it breaks JNI.* The bundled natives export symbols keyed to the
Java package:
{code}
_Java_com_github_luben_zstd_ZstdCompressCtx_compressByteArray0
_Java_com_github_luben_zstd_ZstdBufferDecompressingStreamNoFinalizer_decompressStreamNative
{code}
The existing relocation would move the classes to
{{org.apache.flink.kafka.shaded.com.github.luben.zstd}}, so the JVM would look
for {{Java_org_apache_flink_kafka_shaded_com_github_luben_zstd_*}} and fail
with {{UnsatisfiedLinkError}} on first use. zstd-jni cannot be shaded.
*Not relocating it breaks the packaging test.* {{PackagingITCase}} asserts the
jar contains only {{org/apache/flink/}} and {{META-INF/}}. The zstd-jni jar
carries 17 native libraries under top-level {{darwin/}}, {{freebsd/}},
{{linux/}} and {{win/}}, so bundling it unrelocated adds five forbidden
top-level entries, and leaves an unrelocated third-party native library in the
jar where it can collide with a user-supplied zstd-jni.
h2. Options
# Bundle unrelocated, widen {{PackagingITCase}} to permit the codec and native
paths, and add the BSD 2-Clause entry to {{NOTICE}} plus
{{META-INF/licenses/LICENSE.zstd-jni}}. Adds roughly 10 MB and deliberately
departs from the convention that test encodes.
# Document it: note in the Kafka SQL connector docs that
{{compression.type=zstd}} requires {{zstd-jni}} on the cluster classpath. Cheap
and honest, but does not deliver the parity with the Maven dependency that this
ticket asks for.
# Add {{zstd-jni}} to flink-dist alongside snappy and lz4, which would make
this an {{apache/flink}} ticket rather than a connector one, and arguably
addresses the real inconsistency.
I lean towards 3 as the correct fix and 2 as the stopgap, but this is a
packaging policy call rather than a code one. If a committer indicates a
preference I am happy to implement it.
> 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
> Priority: Major
> Labels: compression, kafka, packaging
>
> 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)
