I am trying to run kafka-streams application on windows machine, however when the application starts it fails to give file permission and ends up with below stack trace.
I have given the custom state directory which has write permission. Has anyone encountered this issue before? I have also tried giving state.dir path under my local user directory but still getting same error.
state.dir = M:\streams
topology.optimization = none
upgrade.from = null
windowstore.changelog.additional.retention.ms = 86400000
2021-05-21 02:54:39.729 INFO 12060 --- [ main] o.a.k.s.KafkaStreams : stream-client [test-admin-stream-9bccb29d-b515-4e96-8e05-0fc20614be7e] Kafka Streams version: 2.6.1
2021-05-21 02:54:39.730 INFO 12060 --- [ main] o.a.k.s.KafkaStreams : stream-client [test-admin-stream-9bccb29d-b515-4e96-8e05-0fc20614be7e] Kafka Streams commit ID: 6b2021cd52659cef
2021-05-21 02:54:39.851 INFO 12060 --- [ main] ConditionEvaluationReportLoggingListener :
Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2021-05-21 02:54:39.912 ERROR 12060 --- [ main] o.s.b.SpringApplication : Application run
java.lang.IllegalStateException: Failed to execute CommandLineRunner
at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:822) [spring-boot-2.4.5.jar:2.4.5]
at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:803) [spring-boot-2.4.5.jar:2.4.5]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:346) [spring-boot-2.4.5.jar:2.4.5]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1340) [spring-boot-2.4.5.jar:2.4.5]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1329) [spring-boot-2.4.5.jar:2.4.5]
at com.pb.testadmin.testAdminApplication.main(testAdminApplication.java:10) [classes/:?]
Caused by: java.lang.UnsupportedOperationException
at java.nio.file.Files.setPosixFilePermissions(Files.java:2079) ~[?:?]
at org.apache.kafka.streams.processor.internals.StateDirectory.<init>(StateDirectory.java:115) ~[kafka-streams-2.6.1.jar:?]
at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:745) ~[kafka-streams-2.6.1.jar:?]
at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:657) ~[kafka-streams-2.6.1.jar:?]
at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:567) ~[kafka-streams-2.6.1.jar:?]
at com.pb.testadmin.consumer.ForexStream.run(ForexStream.java:55) ~[classes/:?]
at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:819) ~[spring-boot-2.4.5.jar:2.4.5]
Below is code snippet of streams.
props.put(StreamsConfig.STATE_DIR_CONFIG, "M:\\streams");
final StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> leftSource = builder.stream("REQ");
KStream<String, String> rightSource = builder.stream("RESP");
KStream<String, String> joined = leftSource.join(rightSource, (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue,
JoinWindows.of(Duration.ofMinutes(1)),
Joined.with(Serdes.String(), Serdes.String(), Serdes.String())
);
joined.to("test");
final Topology topology = builder.build();
streamsInnerJoin = new KafkaStreams(topology, props);
streamsInnerJoin.start();