KAFKA-12648: fix flaky #shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTopologyWithRepartitioning #11868

ableegoldman · 2022-03-09T08:54:04Z

This test has started to become flaky at a relatively low, but consistently reproducible, rate. Upon inspection, we find this is due to IOExceptions during the #cleanUpNamedTopology call -- specifically, most often a DirectoryNotEmptyException with an ocasional FileNotFoundException

Basically, signs pointed to having returned from/completed the #removeNamedTopology future prematurely, and moving on to try and clear out the topology's state directory while there was a streamthread somewhere that was continuing to process/close its tasks.

I believe this is due to updating the thread's topology version before we perform the actual topology update, in this case specifically the act of eg clearing out a directory. If one thread updates its version and then goes to perform the topology removal/cleanup when the second thread finishes its own topology removal, this other thread will check whether all threads are on the latest version and complete any waiting futures if so -- which means it can complete the future before the first thread has actually completed the corresponding action

wcarlson5

KafkaFuture.allOf is a nice find.

LGTM when you get a green build

guozhangwang

Just one comment about the fix itself.

guozhangwang · 2022-03-09T22:04:35Z

...apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java

-                        ex.printStackTrace();
+        final KafkaFutureImpl<Void> resetOffsetsFuture = new KafkaFutureImpl<>();
+        try {
+            removeTopologyFuture.get();


Why we have to wait on the first future before moving forward to construct the second future now? I thought the main fix is only in https://github.com/apache/kafka/pull/11868/files#diff-8baa5d7209fc00074bf3fe24d709c2dcf2a44c1623d7ced8c0e29c1d832a3bcbR1132 above, and with that we do not need to change behavior to wait for the removal of topology completes still?

Yeah that is the main fix, however I realized that we are currently in this awkward state of psueod-async-ness and I think we might ultimately want to scratch this whole RemoveNamedTopologyResult and just make it fully blocking. Though I didn't want to go ahead and change the method signatures just yet, so I just have it block on the named topology future and then perform the offset reset.

The actual advantage here is that before this, we were actually making the StreamThread who completed the future perform the offset reset, which of course means it gets stuck for a bit and can't continue processing until basically the whole group has dropped this named topology. Better to have the caller thread do the offset reset to let the StreamThreads keep processing the other topologies.

(When we get to finally doing a KIP maybe we can discuss having a blocking and non-blocking option for these, but my feeling is let's not complicate things unnecessarily and it may be that we only really need a blocking version)

Got it, that makes a lot of sense, thanks!

guozhangwang

Thanks @ableegoldman , lgtm!

wcarlson5 · 2022-03-10T03:35:52Z

shouldAllowRemovingAndAddingNamedTopologyToRunningApplicationWithMultipleNodesAndResetsOffsets is deadlocking. We need to be able to have its be async a little or else it can't make progress

ableegoldman · 2022-03-10T04:25:18Z

Ah good catch @wcarlson5 , I need to move the offset reset logic out of the actual removeNamedTopology call and make sure we don't block on the offsets being removed until the get() -- otherwise we get deadlock if we have two Streams clients and try to remove a named topology from both from a single thread. (To clarify for anyone else, this is not really a realistic/recommended usage pattern for real applications, but it helps keep the tests simple and makes for a more intuitive blocking behavior anyways)

Anyways I'll push a fix for this, and then we should be good to go 👍

guozhangwang · 2022-03-10T06:02:52Z

Anyways I'll push a fix for this, and then we should be good to go 👍

Thanks @wcarlson5 for the report! @ableegoldman please feel free to move on afterwards.

ableegoldman · 2022-03-10T07:29:50Z

...apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java

@@ -239,12 +238,16 @@ public RemoveNamedTopologyResult removeNamedTopology(final String topologyToRemo
        final boolean skipResetForUnstartedApplication =
            maybeCompleteFutureIfStillInCREATED(removeTopologyFuture, "removing topology " + topologyToRemove);

-        if (resetOffsets && !skipResetForUnstartedApplication) {
+        if (resetOffsets && !skipResetForUnstartedApplication && !partitionsToReset.isEmpty()) {


Moved the !partitionsToReset.isEmpty() check here to make sure we don't log the line about resetting offsets if we don't actually have any offsets to reset

ableegoldman · 2022-03-10T07:35:41Z

...apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java

-                        Thread.sleep(100);
-                    } catch (final InterruptedException ex) {
-                        ex.printStackTrace();
+    private void resetOffsets(final Set<TopicPartition> partitionsToReset) throws StreamsException {


Sorry for the large diff -- it's mainly due to spacing from having moved the 1!partitionsToReset.isEmpty(), plus one small stylistic change to use a while true loop with breaks because following the null status of the deleteOffsetsResult was a bit confusing.

The real change though is that this method now just performs the offset resets directly, rather than directing whoever completes the removeNamedTopology future to perform the offset reset (which is non-trivial and thus not appropriate for the StreamThreads to do).

We now invoke this directly when the user calls get() on the future returned from the RemoveNamedTopologyResult.

This is the main change since being approved @wcarlson5 @guozhangwang

There's also the

ableegoldman · 2022-03-10T07:37:00Z

...va/org/apache/kafka/streams/processor/internals/namedtopology/RemoveNamedTopologyResult.java

+        if (resetOffsetsFuture == null) {
+            return removeTopologyFuture;
+        } else {
+            return resetOffsetsFuture;


Basically we now have the caller thread perform the offset reset and block on it when it goes to call get() on the future returned by RemoveNamedTopologyResult#all (or #resetOffsetsFuture)

wcarlson5

These new changes make sense to me

ableegoldman · 2022-03-10T20:01:35Z

Test failures are unrelated. Merging

ableegoldman · 2022-03-10T20:02:15Z

Merged to trunk

guozhangwang · 2022-03-10T23:58:51Z

@wcarlson5 @ableegoldman I'm wondering if we could still have live lock scenarios like this: say we have two topologies A and B, and two threads a and b each on a different KS instance. And each thread tries to remove one topology at a time, getting the future to make sure it is cleaned up, BUT they did it in a different order.

Thread a calls remove(A) and gets a futureA.
Thread b calls remove(B) and gets a futureB.
Thread a calls futureA. get() trying to delete offsets, and after that it tries to remove(B), but gets blocked on topology A not removed by thread b yet.
Thread b calls futureB. get() trying to delete offsets, and after that it tries to remove(A), but gets blocked on topology B not removed by thread a yet.

Would that be a concern?

wcarlson5 · 2022-03-11T00:01:43Z

@guozhangwang yes that would be a concern.

I think we need to document that these calls need to be made in the same order with the same topologies for each client. That is what KSQL does to make sure it works

guozhangwang · 2022-03-13T04:13:01Z

I think we need to document that these calls need to be made in the same order with the same topologies for each client. That is what KSQL does to make sure it works

Sounds good, thanks for the clarification!

fixes

fbfdd32

ableegoldman requested a review from guozhangwang March 9, 2022 08:58

wcarlson5 approved these changes Mar 9, 2022

View reviewed changes

guozhangwang reviewed Mar 9, 2022

View reviewed changes

checkstyle

0980e8b

ableegoldman requested review from guozhangwang and vvcephei March 10, 2022 00:48

guozhangwang approved these changes Mar 10, 2022

View reviewed changes

A. Sophie Blee-Goldman added 3 commits March 9, 2022 22:52

move offset reset to RemoveNamedTopologyResult#get

836b843

fix

99838e4

cut down on line changes

eaa8c12

ableegoldman commented Mar 10, 2022

View reviewed changes

touch to trigger build

3aaf8a9

wcarlson5 approved these changes Mar 10, 2022

View reviewed changes

ableegoldman merged commit 113595c into apache:trunk Mar 10, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-12648: fix flaky #shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTopologyWithRepartitioning #11868

KAFKA-12648: fix flaky #shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTopologyWithRepartitioning #11868

ableegoldman commented Mar 9, 2022

wcarlson5 left a comment •

edited

guozhangwang left a comment

guozhangwang Mar 9, 2022

ableegoldman Mar 10, 2022

guozhangwang Mar 10, 2022

guozhangwang left a comment

wcarlson5 commented Mar 10, 2022

ableegoldman commented Mar 10, 2022

guozhangwang commented Mar 10, 2022

ableegoldman Mar 10, 2022

ableegoldman Mar 10, 2022

ableegoldman Mar 10, 2022

wcarlson5 left a comment

ableegoldman commented Mar 10, 2022

ableegoldman commented Mar 10, 2022

guozhangwang commented Mar 10, 2022

wcarlson5 commented Mar 11, 2022

guozhangwang commented Mar 13, 2022

KAFKA-12648: fix flaky #shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTopologyWithRepartitioning #11868

KAFKA-12648: fix flaky #shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTopologyWithRepartitioning #11868

Conversation

ableegoldman commented Mar 9, 2022

wcarlson5 left a comment • edited

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

guozhangwang Mar 9, 2022

Choose a reason for hiding this comment

ableegoldman Mar 10, 2022

Choose a reason for hiding this comment

guozhangwang Mar 10, 2022

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

wcarlson5 commented Mar 10, 2022

ableegoldman commented Mar 10, 2022

guozhangwang commented Mar 10, 2022

ableegoldman Mar 10, 2022

Choose a reason for hiding this comment

ableegoldman Mar 10, 2022

Choose a reason for hiding this comment

ableegoldman Mar 10, 2022

Choose a reason for hiding this comment

wcarlson5 left a comment

Choose a reason for hiding this comment

ableegoldman commented Mar 10, 2022

ableegoldman commented Mar 10, 2022

guozhangwang commented Mar 10, 2022

wcarlson5 commented Mar 11, 2022

guozhangwang commented Mar 13, 2022

wcarlson5 left a comment •

edited