使用 Kafka Streams 开发时，Lib 上的 UnsatisfiedLinkError 会影响 DB dll

问题描述 投票：0回答：7

我正在我的开发 Windows 机器上编写 Kafka Streams 应用程序。 如果我尝试使用 Kafka Streams 的 

leftJoin
和 
branch
功能，我在执行 jar 应用程序时会收到以下错误：

Exception in thread "StreamThread-1" java.lang.UnsatisfiedLinkError: C:\Users\user\AppData\Local\Temp\librocksdbjni325337723194862275.dll: Can't find dependent libraries
    at java.lang.ClassLoader$NativeLibrary.load(Native Method)
    at java.lang.ClassLoader.loadLibrary0(ClassLoader.java:1941)
    at java.lang.ClassLoader.loadLibrary(ClassLoader.java:1824)
    at java.lang.Runtime.load0(Runtime.java:809)
    at java.lang.System.load(System.java:1086)
    at org.rocksdb.NativeLibraryLoader.loadLibraryFromJar(NativeLibraryLoader.java:78)
    at org.rocksdb.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:56)
    at org.rocksdb.RocksDB.loadLibrary(RocksDB.java:64)
    at org.rocksdb.RocksDB.<clinit>(RocksDB.java:35)
    at org.rocksdb.Options.<clinit>(Options.java:22)
    at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:115)
    at org.apache.kafka.streams.state.internals.Segment.openDB(Segment.java:38)
    at org.apache.kafka.streams.state.internals.Segments.getOrCreateSegment(Segments.java:75)
    at org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:72)
    at org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.put(ChangeLoggingSegmentedBytesStore.java:54)
    at org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.put(MeteredSegmentedBytesStore.java:101)
    at org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:109)
    at org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:101)
    at org.apache.kafka.streams.kstream.internals.KStreamJoinWindow$KStreamJoinWindowProcessor.process(KStreamJoinWindow.java:65)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
    at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
    at org.apache.kafka.streams.kstream.internals.KStreamFlatMapValues$KStreamFlatMapValuesProcessor.process(KStreamFlatMapValues.java:43)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
    at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
    at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:44)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
    at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:70)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:197)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:641)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)

好像 Kafka 没有找到 DLL，但是等等...我正在开发一个 Java 应用程序！

可能是什么问题？如果我尝试执行更简单的流操作（例如仅 

filter
），为什么此错误不会显示出来？

更新：

仅当代理中存在消息时才会出现此问题。我正在使用 Kafka Streams 版本 0.10.2.1。

这是引发问题的代码片段

public class KafkaStreamsMainClass {

    private KafkaStreamsMainClass() {
    }

    public static void main(final String[] args) throws Exception {
        Properties streamsConfiguration = new Properties();
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams");
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-server:9092");
        streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "schema-registry:8082");
        streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
        streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
        streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
        streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
        KStreamBuilder builder = new KStreamBuilder();
        KStream<GenericRecord, GenericRecord> sourceStream = builder.stream(SOURCE_TOPIC);

        KStream<GenericRecord, GenericRecord> finishedFiltered = sourceStream
                .filter((GenericRecord key, GenericRecord value) -> value.get("endTime") != null);

        KStream<GenericRecord, GenericRecord>[] branchedStreams = sourceStream
                .filter((GenericRecord key, GenericRecord value) -> value.get("endTime") == null)
                .branch((GenericRecord key, GenericRecord value) -> value.get("firstField") != null,
                        (GenericRecord key, GenericRecord value) -> value.get("secondField") != null);

        branchedStreams[0] = finishedFiltered.join(branchedStreams[0],
                (GenericRecord value1, GenericRecord value2) -> {
                    return value1;
                }, JoinWindows.of(TimeUnit.SECONDS.toMillis(2)));

        branchedStreams[1] = finishedFiltered.join(branchedStreams[1],
                (GenericRecord value1, GenericRecord value2) -> {
                    return value1;
                }, JoinWindows.of(TimeUnit.SECONDS.toMillis(2)));

        KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
        streams.setUncaughtExceptionHandler((Thread thread, Throwable throwable) -> {
            throwable.printStackTrace();
        });
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }

}

我打开了 Maven 下载的 

rocksdbjni-5.0.1.jar
存档，其中包含 
librocksdbjni-win64.dll
库。看起来它正在尝试从 RocksDB 外部而不是内部检索库。

我正在 Windows 7 机器上进行开发。

你有遇到过这个问题吗？

java apache-kafka apache-kafka-streams rocksdb
7个回答
15
投票

最近我也遇到了这个问题。我设法分两步解决这个问题：

  1. 删除 
    librocksdbjni[...].dll
    文件夹中的所有 
    C:\Users\[your_user]\AppData\Local\Temp
    文件。
  2. 在项目中添加 
    rocksdb
    的 Maven 依赖项，这对我有用：https://mvnrepository.com/artifact/org.rocksdb/rocksdbjni/5.0.1

编译您的 Kafka Stream 应用程序并运行它。应该有效！

3
投票

我将我的 kafka-streams 项目更新到最新发布的版本 1.0.0。

这个版本存在 this bug，但是在对其进行修补并将此修补版本上传到内部 Artifactory 服务器上之后，我们能够在 Windows 和 Linux 上执行我们的 kafka-streams 代理。 下一个版本 1.0.1 和 1.1.0 将修复此错误，因此一旦发布其中一个版本，我们将切换到它们而不是修补版本。

总而言之，Kafka 人员在 1.0.0 版本中解决了这个错误。

1
投票

我的问题是

/tmp/
目录中的权限（CentOS）

rockdb 使用 

java.io.tmpdir

系统属性内部决定放置

librocksdbjni
文件的位置，通常是这样的
/tmp/librocksdbjni2925599838907625983.so

通过在 kafka-streams 应用程序中设置不同的 tempdir 属性和适当的权限来解决。

System.setProperty("java.io.tmpdir", "/opt/kafka-streams/tmp");
0
投票

您缺少一些rocksdb dll依赖的本机库。请参阅https://github.com/facebook/rocksdb/issues/1302

0
投票

我在使用 jdk 1.8 时遇到了同样的问题。当我把它改成jre后就解决了。

0
投票

在 Mac 中遇到类似的问题。根据此链接，https://github.com/facebook/rocksdb/issues/5064问题与我的 Mac OS 版本 (10.11.6) 中安装的旧版 libc 有关。

0
投票

我在为 KStreams 容器应用程序使用 distroless JRE 映像时遇到了此问题。

我从

mcr.microsoft.com/openjdk/jdk:11-distroless
切换到
mcr.microsoft.com/openjdk/jdk:11-ubuntu
，然后它就消失了

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.