CODEWITHSUNDEEP
lambdaapache-kafkakafka-consumer-apiapache-kafka-streams
dijeah
dijeah

Reputation: 303

RocksDB exception in Kafka Streams

In a simple Kafka Stream program when I am using the below code, it works without throwing any error:

      KTable<String, Long> result= source.mapValues(textLine
      ->textLine.toLowerCase()) .flatMapValues(lowercasedTextLine ->
      Arrays.asList(lowercasedTextLine.split(" "))) .selectKey((ignoredKey,word) ->
      word) .groupByKey() .count("Counts");

      result.to(Serdes.String(), Serdes.Long(), "wc-output");

However when I am using the below code I am getting error:

    KStream<String, String> source = builder.stream("wc-input");
    source.groupBy((key, word) -> word).windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(5000))).count()
            .toStream().map((key, value) -> new KeyValue<>(key.key(), value))
            .to("wc-output", Produced.with(Serdes.String(), Serdes.Long()));

Exception in thread "streams-wordcount-b160d715-f0e0-42ee-831e-0e4eed7e9424-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=1_0, processor=KSTREAM-SOURCE-0000000006, topic=streams-wordcount-KSTREAM-AGGREGATE-STATE-STORE-0000000002-repartition, partition=0, offset=0 at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:232) at org.apache.kafka.streams.processor.internals.AssignedTasks.process(AssignedTasks.java:403) at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:317) at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:942) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:822) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744) Caused by: org.apache.kafka.streams.errors.ProcessorStateException: Error opening store KSTREAM-AGGREGATE-STATE-STORE-0000000002:1553472000000 at location \tmp\kafka-streams\streams-wordcount\1_0\KSTREAM-AGGREGATE-STATE-STORE-0000000002\KSTREAM-AGGREGATE-STATE-STORE-0000000002:1553472000000 at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:204) at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:174) at org.apache.kafka.streams.state.internals.Segment.openDB(Segment.java:40) at org.apache.kafka.streams.state.internals.Segments.getOrCreateSegment(Segments.java:89) at org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:81) at org.apache.kafka.streams.state.internals.RocksDBWindowStore$RocksDBWindowBytesStore.put(RocksDBWindowStore.java:43) at org.apache.kafka.streams.state.internals.RocksDBWindowStore$RocksDBWindowBytesStore.put(RocksDBWindowStore.java:34) at org.apache.kafka.streams.state.internals.ChangeLoggingWindowBytesStore.put(ChangeLoggingWindowBytesStore.java:67) at org.apache.kafka.streams.state.internals.ChangeLoggingWindowBytesStore.put(ChangeLoggingWindowBytesStore.java:33) at org.apache.kafka.streams.state.internals.CachingWindowStore$1.apply(CachingWindowStore.java:100) at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:141) at org.apache.kafka.streams.state.internals.NamedCache.evict(NamedCache.java:232) at org.apache.kafka.streams.state.internals.ThreadCache.maybeEvict(ThreadCache.java:245) at org.apache.kafka.streams.state.internals.ThreadCache.put(ThreadCache.java:153) at org.apache.kafka.streams.state.internals.CachingWindowStore.put(CachingWindowStore.java:157) at org.apache.kafka.streams.state.internals.CachingWindowStore.put(CachingWindowStore.java:36) at org.apache.kafka.streams.state.internals.MeteredWindowStore.put(MeteredWindowStore.java:96) at org.apache.kafka.streams.kstream.internals.KStreamWindowAggregate$KStreamWindowAggregateProcessor.process(KStreamWindowAggregate.java:122) at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46) at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:85) at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216) ... 6 more Caused by: org.rocksdb.RocksDBException: Failed to create dir: H:\tmp\kafka-streams\streams-wordcount\1_0\KSTREAM-AGGREGATE-STATE-STORE-0000000002\KSTREAM-AGGREGATE-STATE-STORE-0000000002:1553472000000: Invalid argument at org.rocksdb.RocksDB.open(Native Method) at org.rocksdb.RocksDB.open(RocksDB.java:231) at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:197)

Upvotes: 1

Views: 2533

Answers (1)

Matthias J. Sax
Matthias J. Sax

Reputation: 62360

When you use a windowed aggregation, stores a named differently, and there is a bug in Kafka 1.0.0 affecting Windows OS: the name for windowed stores contains a : that is not allowed on Windows OS. The bug is fixed in version 1.0.1 and 1.1.0

Cf. https://issues.apache.org/jira/browse/KAFKA-6167

Upvotes: 2

Related Questions