Hello,
I am getting following exception while running Beam Samza Runner –
java.lang.UnsupportedOperationException: Cannot create a producer for an input
system
at
org.apache.beam.runners.samza.adapter.BoundedSourceSystem$Factory.getProducer(BoundedSourceSystem.java:411)
at
org.apache.samza.container.SamzaContainer$$anonfun$13.apply(SamzaContainer.scala:223)
at
org.apache.samza.container.SamzaContainer$$anonfun$13.apply(SamzaContainer.scala:220)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.Map$Map3.foreach(Map.scala:161)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at
org.apache.samza.container.SamzaContainer$.apply(SamzaContainer.scala:220)
at org.apache.samza.container.SamzaContainer.apply(SamzaContainer.scala)
at
org.apache.samza.processor.StreamProcessor.createSamzaContainer(StreamProcessor.java:198)
at
org.apache.samza.processor.StreamProcessor$1.onNewJobModel(StreamProcessor.java:290)
at
org.apache.samza.zk.ZkJobCoordinator.onNewJobModelConfirmed(ZkJobCoordinator.java:304)
at
org.apache.samza.zk.ZkJobCoordinator$ZkBarrierListenerImpl.lambda$onBarrierStateChanged$1(ZkJobCoordinator.java:394)
at
org.apache.samza.zk.ScheduleAfterDebounceTime.lambda$getScheduleableAction$0(ScheduleAfterDebounceTime.java:163)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
2019-01-04 21:39:11 ERROR SamzaContainer$:86 - Failed to create a producer for
0-KafkaIO_Read_Read_KafkaUnboundedSource__out__PCollection_, so skipping.
java.lang.UnsupportedOperationException: Cannot create a producer for an input
system
at
org.apache.beam.runners.samza.adapter.UnboundedSourceSystem$Factory.getProducer(UnboundedSourceSystem.java:452)
at
org.apache.samza.container.SamzaContainer$$anonfun$13.apply(SamzaContainer.scala:223)
at
org.apache.samza.container.SamzaContainer$$anonfun$13.apply(SamzaContainer.scala:220)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.Map$Map3.foreach(Map.scala:161)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at
org.apache.samza.container.SamzaContainer$.apply(SamzaContainer.scala:220)
at org.apache.samza.container.SamzaContainer.apply(SamzaContainer.scala)
at
org.apache.samza.processor.StreamProcessor.createSamzaContainer(StreamProcessor.java:198)
at
org.apache.samza.processor.StreamProcessor$1.onNewJobModel(StreamProcessor.java:290)
at
org.apache.samza.zk.ZkJobCoordinator.onNewJobModelConfirmed(ZkJobCoordinator.java:304)
at
org.apache.samza.zk.ZkJobCoordinator$ZkBarrierListenerImpl.lambda$onBarrierStateChanged$1(ZkJobCoordinator.java:394)
at
org.apache.samza.zk.ScheduleAfterDebounceTime.lambda$getScheduleableAction$0(ScheduleAfterDebounceTime.java:163)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
This exception does not stop the execution, however I would like to understand
the reason for this and possible resolution.
Thanks,
Omkar Deshpande
