Hi Avi,
you need to complete the given resultFuture and not return a future. You
can do this via resultFuture.complete(r).
Cheers,
Till
On Tue, May 7, 2019 at 8:30 PM Avi Levi <avi.l...@bluevoyant.com> wrote:
> Hi,
> We are using flink 1.8.0 (but the flowing also happens in 1.7.2) I tried
> very simple unordered async call
> override def asyncInvoke(input: Foo, resultFuture:
> ResultFuture[ScoredFoo]) : Unit = {
> val r = ScoredFoo(Foo("a"), 80)
> Future.successful(r)
> }
>
> Running this stream seem to be stuck in some infinite loop until it
> crashes on timeout exception.:
>
> *java.lang.Exception: An async function call terminated with an exception.
> Failing the AsyncWaitOperator.*
> * at
> org.apache.flink.streaming.api.operators.async.Emitter.output(Emitter.java:137)*
> * at
> org.apache.flink.streaming.api.operators.async.Emitter.run(Emitter.java:85)*
> * at java.base/java.lang.Thread.run(Thread.java:844)*
> *Caused by: java.util.concurrent.ExecutionException:
> java.util.concurrent.TimeoutException: Async function call has timed out.*
> * at
> java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)*
> * at
> java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)*
> * at
> org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry.get(StreamRecordQueueEntry.java:68)*
> * at
> org.apache.flink.streaming.api.operators.async.Emitter.output(Emitter.java:129)*
> * ... 2 common frames omitted*
> *Caused by: java.util.concurrent.TimeoutException: Async function call has
> timed out.*
> * at
> org.apache.flink.streaming.api.scala.async.AsyncFunction.timeout(AsyncFunction.scala:60)*
> * at
> org.apache.flink.streaming.api.scala.async.AsyncFunction.timeout$(AsyncFunction.scala:59)*
> * at
> com.lookalike.analytic.utils.LookalikeScoreEnrich.timeout(LookalikeScoreEnrich.scala:18)*
> * at
> org.apache.flink.streaming.api.scala.AsyncDataStream$$anon$3.timeout(AsyncDataStream.scala:301)*
> * at
> org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$1.onProcessingTime(AsyncWaitOperator.java:211)*
> * at
> org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:285)*
> * at
> java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:514)*
> * at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)*
> * at
> java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:299)*
> * at
> java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)*
> * at
> java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)*
> * ... 1 common frames omitted*
>
> Please advise , Thanks
> Avi
>
>
