HI Qi, You are right. I forgot to update the other thread on this. I suspected Kafka 1.0 was the reason for NPE since it had trouble instantiating the logger. Here is the Kafka client support matrix for our releases
Dependency Supported versions Not supported version Samza version Kafka 0.11.* 1.*, 2.* 1.0 1.1 2.0.* 1.* and below 1.2(Upcoming release) Hope it helps. Thanks, Bharath On Tue, Jun 4, 2019 at 11:24 PM QiShu <sh...@eefung.com> wrote: > Hi, > > Running environment: > Samza version: 2.12-1.1.0 > Kafka cluster version: 2.12-1.1.0 > Hadoop version: 3.1.0 > > When using Kafka_2.12-1.1.0.jar, Samza job failed to run when > retrieving coordinator stream meta info from Kafka: > 2019-06-03 17:26:11.176 [main] KafkaSystemAdmin [INFO] Fetching > SystemStreamMetadata for topics [__samza_coordinator_canal-metrics-test_1] > on system kafka > 2019-06-03 17:26:11.179 [main] KafkaSystemAdmin [ERROR] Fetching system > stream metadata for: [__samza_coordinator_canal-metrics-test_1] threw an > exception. > java.lang.NullPointerException > at > java.util.concurrent.ConcurrentHashMap.get(ConcurrentHashMap.java:936) > at > org.slf4j.impl.Log4jLoggerFactory.getLogger(Log4jLoggerFactory.java:75) > at org.slf4j.LoggerFactory.getLogger(LoggerFactory.java:253) > at kafka.utils.Logging.logger(Logging.scala:43) > at kafka.utils.Logging.debug(Logging.scala:62) > at kafka.utils.Logging.debug$(Logging.scala:62) > at > org.apache.samza.system.kafka.KafkaSystemAdminUtilsScala$.debug(KafkaSystemAdminUtilsScala.scala:45) > at > org.apache.samza.system.kafka.KafkaSystemAdminUtilsScala$.assembleMetadata(KafkaSystemAdminUtilsScala.scala:74) > at > org.apache.samza.system.kafka.KafkaSystemAdminUtilsScala.assembleMetadata(KafkaSystemAdminUtilsScala.scala) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.fetchSystemStreamMetadata(KafkaSystemAdmin.java:461) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.access$400(KafkaSystemAdmin.java:76) > at > org.apache.samza.system.kafka.KafkaSystemAdmin$4.apply(KafkaSystemAdmin.java:340) > at > org.apache.samza.system.kafka.KafkaSystemAdmin$4.apply(KafkaSystemAdmin.java:337) > at > org.apache.samza.util.ExponentialSleepStrategy.run(ExponentialSleepStrategy.scala:90) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.getSystemStreamMetadata(KafkaSystemAdmin.java:374) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.getSystemStreamMetadata(KafkaSystemAdmin.java:292) > at > org.apache.samza.coordinator.stream.CoordinatorStreamSystemConsumer.register(CoordinatorStreamSystemConsumer.java:109) > at org.apache.samza.job.JobRunner.run(JobRunner.scala:105) > at > org.apache.samza.runtime.RemoteApplicationRunner.lambda$run$0(RemoteApplicationRunner.java:76) > at java.util.ArrayList.forEach(ArrayList.java:1255) > at > org.apache.samza.runtime.RemoteApplicationRunner.run(RemoteApplicationRunner.java:73) > at > org.apache.samza.runtime.ApplicationRunnerUtil.invoke(ApplicationRunnerUtil.java:49) > at > org.apache.samza.runtime.ApplicationRunnerMain.main(ApplicationRunnerMain.java:53) > Exception in thread "main" org.apache.samza.SamzaException: Failed to run > application > at > org.apache.samza.runtime.RemoteApplicationRunner.run(RemoteApplicationRunner.java:79) > at > org.apache.samza.runtime.ApplicationRunnerUtil.invoke(ApplicationRunnerUtil.java:49) > at > org.apache.samza.runtime.ApplicationRunnerMain.main(ApplicationRunnerMain.java:53) > Caused by: org.apache.samza.SamzaException: java.lang.NullPointerException > at > org.apache.samza.system.kafka.KafkaSystemAdmin$5.apply(KafkaSystemAdmin.java:358) > at > org.apache.samza.system.kafka.KafkaSystemAdmin$5.apply(KafkaSystemAdmin.java:347) > at > org.apache.samza.util.ExponentialSleepStrategy.run(ExponentialSleepStrategy.scala:97) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.getSystemStreamMetadata(KafkaSystemAdmin.java:374) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.getSystemStreamMetadata(KafkaSystemAdmin.java:292) > at > org.apache.samza.coordinator.stream.CoordinatorStreamSystemConsumer.register(CoordinatorStreamSystemConsumer.java:109) > at org.apache.samza.job.JobRunner.run(JobRunner.scala:105) > at > org.apache.samza.runtime.RemoteApplicationRunner.lambda$run$0(RemoteApplicationRunner.java:76) > at java.util.ArrayList.forEach(ArrayList.java:1255) > at > org.apache.samza.runtime.RemoteApplicationRunner.run(RemoteApplicationRunner.java:73) > ... 2 more > Caused by: java.lang.NullPointerException > at > java.util.concurrent.ConcurrentHashMap.get(ConcurrentHashMap.java:936) > at > org.slf4j.impl.Log4jLoggerFactory.getLogger(Log4jLoggerFactory.java:75) > at org.slf4j.LoggerFactory.getLogger(LoggerFactory.java:253) > at kafka.utils.Logging.logger(Logging.scala:43) > at kafka.utils.Logging.debug(Logging.scala:62) > at kafka.utils.Logging.debug$(Logging.scala:62) > at > org.apache.samza.system.kafka.KafkaSystemAdminUtilsScala$.debug(KafkaSystemAdminUtilsScala.scala:45) > at > org.apache.samza.system.kafka.KafkaSystemAdminUtilsScala$.assembleMetadata(KafkaSystemAdminUtilsScala.scala:74) > at > org.apache.samza.system.kafka.KafkaSystemAdminUtilsScala.assembleMetadata(KafkaSystemAdminUtilsScala.scala) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.fetchSystemStreamMetadata(KafkaSystemAdmin.java:461) > at > org.apache.samza.system.kafka.KafkaSystemAdmin.access$400(KafkaSystemAdmin.java:76) > at > org.apache.samza.system.kafka.KafkaSystemAdmin$4.apply(KafkaSystemAdmin.java:340) > at > org.apache.samza.system.kafka.KafkaSystemAdmin$4.apply(KafkaSystemAdmin.java:337) > at > org.apache.samza.util.ExponentialSleepStrategy.run(ExponentialSleepStrategy.scala:90) > ... 9 more > > > > When using Kafka_2.12-0.11.0.2.jar，coordinator stream meta info > can be retrieved successfully. > > So I want to know the supported Kafka version by Samza, thanks! > > > Qi Shu
