Correct way to cleanly shut down StateFun Harness in test code

[Ken Krugler <kk...@transpac.com>]

Hi all,

I’m using org.apache.flink.statefun.flink.harness.Harness in some unit test code, where I control the sources so that they are finite.

This is similar to what I found at https://stackoverflow.com/questions/61939681/is-it-possible-to-write-a-unit-test-which-terminates-using-flink-statefun-harnes

The problem is that if I shut down all of the sources, it looks like StateFun then starts shutting down some resources prematurely, which results in this error:

Caused by: java.lang.IllegalStateException: Mailbox is in state CLOSED, but is required to be in state OPEN for put operations.
	at org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl.checkPutStateConditions(TaskMailboxImpl.java:256) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl.put(TaskMailboxImpl.java:184) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.execute(MailboxExecutorImpl.java:73) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.api.operators.MailboxExecutor.execute(MailboxExecutor.java:98) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.statefun.flink.core.common.MailboxExecutorFacade.execute(MailboxExecutorFacade.java:35) ~[statefun-flink-datastream-2.2.2.jar:2.2.2]
	at org.apache.flink.statefun.flink.core.feedback.FeedbackChannel$ConsumerTask.scheduleDrainAll(FeedbackChannel.java:114) ~[statefun-flink-datastream-2.2.2.jar:2.2.2]
	at org.apache.flink.statefun.flink.core.feedback.FeedbackChannel.put(FeedbackChannel.java:64) ~[statefun-flink-datastream-2.2.2.jar:2.2.2]
	at org.apache.flink.statefun.flink.core.feedback.FeedbackSinkOperator.processElement(FeedbackSinkOperator.java:58) ~[statefun-flink-datastream-2.2.2.jar:2.2.2]
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:158) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:191) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:162) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:374) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:190) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:608) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:574) ~[flink-streaming-java_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:752) ~[flink-runtime_2.12-1.11.4.jar:1.11.4]
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:569) ~[flink-runtime_2.12-1.11.4.jar:1.11.4]
	at java.lang.Thread.run(Thread.java:834) ~[?:?]

I believe the issue is that messages are still being generated but the target for the feedback channel has been terminated.

I currently work around this by turning off just the sources that would trigger new egress results, then I wait for the egresses to all be idle for > 2 seconds, and then I shut down the rest of the sources.

But it feels fragile...

Thanks,

— Ken

PS - using StateFun 2.2.2

--------------------------
Ken Krugler
http://www.scaleunlimited.com
Custom big data solutions
Flink, Pinot, Solr, Elasticsearch