Draining mode doesn't work properly

[Alessandrovito]

Description of my configuration for parallel consumers:

• created new JStreamVertxParallelEoSStreamProcessor<>(vertx, wc, options) to permit to invoke vertxHttpWebClientStream to - send a http POST request to the endpoint
• Use Vertx as HTTP engine
• Set commit mode to CommitMode.PERIODIC_CONSUMER_SYNC and setTimeBetweenCommits to 5 seconds
• Ordering by ProcessingOrder.KEY
• version 0.5.1.0

I have created an integration test to push events with specific sequence: events are pushed into a topic consumed by parallel consumers with key conversationId and finally events are sent to an endpoint in charge of recognizing "out of sequence" of the events, data duplication, data loss per conversationId . During the test we destroy the bean into the context of spring simulating the shutdown and after a while inject the bean and recreate the parallel consumers entities.

I'm reporting the  behaviours about the draining mode exposed by following the APIs:

• parallelStreamProcessor.closeDontDrainFirst();
• parallelStreamProcessor.closeDrainFirst();
• parallelStreamProcessor.closeDrainFirst(Duration.ofSeconds(XXX));

While the closeDrainFirst sends all the events present in the queue before to commit the offset map, on the opposite the closeDontDrainFirst seems to close the connection and doesn't commit at all the current offset map (The error message raised is "Connection was closed " caught in the onFailure handler for Vert.x module)
Here my observation:

• invoking the API closeDontDrainFirst, it closes the HTTP connection to the endpoint and doesn't commit offset map. Then when the parallel consumer starts again, the endpoint receives duplicate events. I thought, It is supposed to wait to send the inflighs events before to commit the map offset and finally to close the parallel consumer.
• invoking the API closeDrainFirst works, but it waits to drain all the events in the queue and this could take longer. Since our application is running in k8s env, when the pod enter into TERMINATE state, k8s sends the SIGTERM signalling to the pod waiting at most 30 seconds before definitely shutdown the pod and then the java application.

Further :

parallelStreamProcessor.closeDrainFirst(Duration.ofSeconds(XXX));

We can see the following error when timeout expires :

09:17:04.840 [main] ERROR io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor - Execution or timeout exception while waiting for the control thread to close cleanly (state was draining). Try increasing your time-out to allow the system to drain, or close without draining.
java.util.concurrent.TimeoutException: null
at java.util.concurrent.FutureTask.get(FutureTask.java:204) ~[?:?]

   
But the events are still sending and no commit offset happens.
I'm a bit lost with different behaviour.
Could you please explain in detail what should be the exact behaviour of closeDontDrainFirst, closeDrainFirst and closeDrainFirst(Duration.ofSeconds(XXX)) ?
Could you explain why closeDontDrainFirst doesn't work as expected ?
Do you have any tests around that ?

[astubbs]

[note:] This was a symptom of #365.

• Received duplicated records when rebalance occurs #365

Hello again! Thanks for submitting your feedback and your rigorous testing! 🤗

Draining hasn't been visited in a while. Upon refreshing myself - it is very aggressive. So what the draining system is currently, is that it basically stops polling for more records, but doesn't close until all previously polled and queued records get finished processing. So - that could be a very long time yes (thousands or more records).

I'm not sure how popular this feature is.

With closeDontDrainFirst, if you have concurrency set to 10 - this would only wait until (up to) those 10 records are completed, commit, then close.

The closeDontDrainFirst system should wait for inflight to complete, and commit before closing. Perhaps this isn't functioning as intended in the Vertx module. Looking it - yes, it does immediately close the WebClient - that's probably in the wrong order. - Because of this, there may end up being no committable offsets.

May I ask - what is your main driver for using the Vert.x module?

[astubbs]

Actually, looking at the Vert.x close method, it does wait for the base system to close, before closing the web client:

parallel-consumer/parallel-consumer-vertx/src/main/java/io/confluent/parallelconsumer/vertx/VertxParallelEoSStreamProcessor.java

Lines 321 to 322 in 7fd051d

	super.close(timeout, drainMode);
	webClient.close();

Perhaps the dupes you are seeing is caused by the truncation issue in your other discussion, and not by any issues with close? #360

[astubbs]

Unless I get further information, I'm going to conclude this was a symptom of #365 . Thanks of your rigorous testing!

• Received duplicated records when rebalance occurs #365

[Alessandrovito]

Really appreciate your work and support. Pretty sure the issue is a symptom of #365. I need to setup all my environment to run again the IT against your new version, but it will take sometime.

[antonmos]

This is a very helpful explanation of what closeDontDrainFirst and closeDrainFirst do. Could you please add it to the documentation?