J'écris une application Kafka Streams sur ma machine Windows de développement. Si j'essaie d'utiliser le leftJoin
y branch
de Kafka Streams Je reçois l'erreur suivante lors de l'exécution de l'application 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)
Il semble que Kafka ne trouve pas de DLL, mais attendez... je développe une application Java !
Quel pourrait être le problème ? Et pourquoi cette erreur ne se manifeste-t-elle pas si j'essaie d'effectuer des opérations de streaming plus simples, comme un simple filter
?
UPDATE :
Ce problème ne se pose que lorsqu'un message est présent dans le courtier. J'utilise Kafka Streams version 0.10.2.1.
Voici le morceau de code qui pose le problème
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));
}
}
J'ai ouvert le rocksdbjni-5.0.1.jar
téléchargée par Maven et qui comprend l'archive librocksdbjni-win64.dll
bibliothèque. Il semble qu'il essaie de récupérer la bibliothèque à partir de l'extérieur de la RocksDB au lieu de l'intérieur.
Je développe sur une machine Windows 7.
Avez-vous déjà rencontré ce problème ?