[
https://issues.apache.org/jira/browse/FLINK-40348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18106088#comment-18106088
]
Bhanu Chander Vallabaneni commented on FLINK-40348:
---------------------------------------------------
I looked into this on {{main}} (kafka.version 4.2.0) and can confirm the
diagnosis, but I think
bundling the codecs runs into a packaging constraint that is worth settling
before anyone writes a
patch.
h3. Confirming the cause
{{kafka-clients:4.2.0}} declares all three codec providers at
{{scope=runtime}}, {{optional=false}}:
* {{com.github.luben:zstd-jni:1.5.6-10}}
* {{at.yawk.lz4:lz4-java:1.10.1}}
* {{org.xerial.snappy:snappy-java:1.1.10.7}}
(the lz4 coordinates moved to {{at.yawk.lz4}} on the 4.2.0 line, so the
ticket's {{org.lz4}} is
slightly out of date, but the scoping point stands unchanged)
And {{flink-connector-kafka}} declares {{kafka-clients}} at compile scope,
non-optional. So a Maven
consumer does receive zstd-jni transitively, while the uber jar's include
whitelist in
{{flink-sql-connector-kafka/pom.xml}} drops all three. The divergence is real.
h3. The constraint
{{flink-sql-connector-kafka}} has an explicit packaging test,
{{PackagingITCase#testPackaging}}:
{code:java}
PackagingTestUtils.assertJarContainsOnlyFilesMatching(
jar, Arrays.asList("org/apache/flink/", "META-INF/"));
{code}
That encodes a deliberate policy: everything in the jar is either
Flink-namespaced or metadata,
which is why {{org.apache.kafka}} is relocated to
{{org.apache.flink.kafka.shaded.org.apache.kafka}}. Adding zstd-jni collides
with it in a way
relocation does not solve:
* zstd-jni ships its native libraries at *top-level* paths --
{{linux/amd64/libzstd-jni-1.5.6-10.so}},
{{darwin/aarch64/...}}, {{win/amd64/...}}, and so on for roughly twenty
platform/arch combinations.
Shade relocation rewrites the {{com.github.luben.zstd}} classes but does not
move those resource
paths, and the loader resolves them by OS/arch at runtime. So the test fails
whether or not the
classes are relocated.
* the artifact is 7.0 MB, essentially all native binaries.
* zstd-jni is BSD 2-Clause rather than ASL2, so it needs its own NOTICE entry
and bundled license
text rather than being folded into the existing ASL2 section.
None of that makes bundling wrong -- BSD 2-Clause is Category A and other ASF
uber jars carry native
dependencies. But it turns a whitelist edit into a decision about what this jar
is allowed to
contain, which is not mine to make unilaterally.
h3. Two directions
# *Bundle it.* Add the codec artifacts to the whitelist, relax
{{PackagingITCase}} to permit the
native library paths, and add the NOTICE and license entries. Users get
parity with the Maven
dependency, at +7 MB and a precedent for non-Flink-namespaced content in the
jar.
# *Fail fast instead.* Leave packaging alone and validate at table-factory
level: if
{{properties.compression.type}} names a codec whose provider is absent, fail
with a message naming
the missing artifact, and document the requirement. Keeps the jar as-is and
replaces a
{{NoClassDefFoundError}} deep inside {{KafkaProducer}} construction with
something actionable.
I lean towards 2 on its own being insufficient if the goal really is
uber-jar/Maven parity, and
towards 1 needing a committer's sign-off on the packaging precedent. Happy to
implement either --
could a committer assign this to me, or say which direction is preferred?
> 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)
