Thanks for looking into this Alexey. I was able to get past this issue after I upgraded to beam 2.15.0. Let me know if you would still like a JIRA for this. That said, I am running into another issue which is that the producer within the KinesisIO is throwing a NullPointerException
More specifically https://github.com/apache/beam/blob/master/sdks/java/io/kinesis/src/main/java/org/apache/beam/sdk/io/kinesis/KinesisIO.java#L685 <https://github.com/apache/beam/blob/master/sdks/java/io/kinesis/src/main/java/org/apache/beam/sdk/io/kinesis/KinesisIO.java#L685> I do see that the producer is transient private static transient IKinesisProducer producer; I was wondering if that had do anything with the error below. org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.run(UnboundedSourceWrapper.java:297)\n\tat org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:93)\n\tat org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:57)\n\tat org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:97)\n\tat org.apache.flink.streaming.runtime.tasks.StoppableSourceStreamTask.run(StoppableSourceStreamTask.java:45)\n\tat org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)\n\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: java.lang.NullPointerException: null\n\tat org.apache.beam.sdk.io.kinesis.KinesisIO$Write$KinesisWriterFn.flushBundle(KinesisIO.java:685)\n\tat org.apache.beam.sdk.io.kinesis.KinesisIO$Write$KinesisWriterFn.finishBundle(KinesisIO.java:669 > On Sep 19, 2019, at 3:20 AM, Alexey Romanenko <[email protected]> > wrote: > > Hi Ankit, > > I haven’t seen this issue and I haven’t managed to reproduce it in my > environment. > We had similar issue [1] some time ago in Beam 2.6, related to Guava shading, > but it was already fixed. > Could you create a new Jira for this one? > Thanks. > > [1] https://issues.apache.org/jira/browse/BEAM-5060 > <https://issues.apache.org/jira/browse/BEAM-5060> > >> On 18 Sep 2019, at 22:51, Ankit Jhalaria <[email protected] >> <mailto:[email protected]>> wrote: >> >> Hey beam devs, >> >> I am trying to use KinesisIO.write() with beam 2.13, running on flink and >> its failing while trying to do >> Futures.addCallback(f, new UserRecordResultFutureCallback()); >> Its currently pulling in beam-vendor-guava-20_0-0.1.jar >> I have tried updating bringing in a current version but that doesn’t seem to >> help either. >> >> I was wondering if others have run into this issue and have recommendations >> for a fix. >> >> Relevant Part of the stack trace >> >> tat >> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)\n\tat >> >> org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollect(StreamSourceContexts.java:305)\n\tat >> >> org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collect(StreamSourceContexts.java:394)\n\tat >> >> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.emitElement(UnboundedSourceWrapper.java:341)\n\tat >> >> org.apache.beam.runners.flink.translation.wrappers.streaming.io.UnboundedSourceWrapper.run(UnboundedSourceWrapper.java:267)\n\tat >> >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:93)\n\tat >> >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:57)\n\tat >> >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:97)\n\tat >> >> org.apache.flink.streaming.runtime.tasks.StoppableSourceStreamTask.run(StoppableSourceStreamTask.java:45)\n\tat >> >> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)\n\tat >> org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)\n\tat >> java.lang.Thread.run(Thread.java:748)\nCaused by: >> java.lang.NoSuchMethodError: >> com.google.common.util.concurrent.Futures.addCallback(Lcom/google/common/util/concurrent/ListenableFuture;Lcom/google/common/util/concurrent/FutureCallback;)V\n\tat >> >> org.apache.beam.sdk.io.kinesis.KinesisIO$Write$KinesisWriterFn.processElement(KinesisIO.java:665)\n"} >> >> Thanks in advance >> >> Best, >> Ankit >
