KAFKA-7672 : force write checkpoint during StreamTask #suspend

[abbccdda]

This fix is aiming for #2 issue pointed out within https://issues.apache.org/jira/browse/KAFKA-7672
In the current setup, we do offset checkpoint file write when EOS is turned on during #suspend, which introduces the potential race condition during StateManager #closeSuspend call. To mitigate the problem, we attempt to always write checkpoint file in #suspend call.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[abbccdda]

@mjsax @guozhangwang Could you take another look to see if this makes sense?

[abbccdda]

@guozhangwang @mjsax Maybe another look?

[abbccdda]

@guozhangwang @mjsax Take a look when you got time?

[abbccdda]

@mjsax Mind take another look?

[mjsax]

The change itself LGTM.

I would recommend to update the comment a little bit.

Call for second review @guozhangwang @bbejeck @vvcephei @ableegoldman

Can we add a test that exposes the issue, ie, force the race condition? (Could maybe also be done as a follow up) -- this is a critical bug fix and we should get it into 2.2 (ie, need to be merged by 2/15).

[mjsax]

Which check? I don't see any if clause.

[bbejeck]

I was thinking the same. Did you mean that the other change on this PR eliminates the chance for a double checkpoint file writes?

[abbccdda]

Oh, the check is just pointing to eosEnabled. Let me update the comment.

[mjsax]

@abbccdda One more thing: can we also add

completedRestorers.clear();

to StoreChangelogReader#reset() as pointed out on the ticket (and add a test if possible).

[mjsax]

Another thought that just crossed my mind while digging into this further: Atm, there is a race condition between writing and reading the checkpoint file. Wouldn't it be simpler to avoid this race condition by moving

        // load the checkpoint information
        checkpointableOffsets.putAll(checkpoint.read());
        if (eosEnabled) {
            // delete the checkpoint file after finish loading its stored offsets
            checkpoint.delete();
            checkpoint = null;
        }

from ProcessorStateManager constructor to ProcessorStateManager#register() ?

This would simplify the logic IMHO.

[vvcephei]

Actually, @mjsax , after reviewing the ticket, I think 6113 (#6113) is the actual bugfix that must go in. So I don't think that @abbccdda needs to add the clear in this PR.

If we can merge 6115 as well, which contains a significant performance improvement for restore in some cases, then it would be ideal.

[bbejeck]

If we can merge 6115 as well, which contains a significant performance improvement for restore in some cases, then it would be ideal.

#6115 includes the completedRestorers.clear() call so I think we just need to make sure both PRs get in.

[vvcephei]

Hi @abbccdda ,

Thanks for the PR.

I had one comment: it seems like we should update more of the state manager APIs to indicate that commit is no longer a guaranteed part of close. Otherwise, we could easily introduce a bug later, calling the method with acked offsets, not realizing that they will actually just be ignored.

Thanks,
-John

[vvcephei]

If we're going to ignore the ackedOffsets argument here, it seems like we should revisit the interface.

There are multiple paths that lead to this method, and it's not analytically obvious why it's ok to just ignore the argument and skip checkpointing here.

If we want to move the checkpoint up in the lifecycle to suspend, then it seems like we should do so holistically and remove the checkpointable offsets from the paramters of close. What do you think?

[abbccdda]

@vvcephei @bbejeck I checked the callers of close(ackedOffsets) function,

1. Abstract task is ok since our fix is to make sure state manager could checkpoint when suspending
2. GlobalStateUpdateTask is also fine since the global state manager (using GlobalStateManagerImpl) is overriding this function who will do offset checkpoint.

So I think we should be safe to ignore this in the base class, but keep the acked offset parameter in order for subclass to checkpoint for now, which minimizing the risk of this PR. Thank you!

[mjsax]

What "subclasses" are you referring to? Both classes you mention are internal and not extended. I don't see any risk that we need to minimize?

[abbccdda]

I mean GlobalStateManagerImpl implements the close(offsets) @mjsax

[mjsax]

Could wel update the code in GlobalStateUpdateTask to:

    public void close() throws IOException {
        stateMgr.checkpoint(offsets);
        stateMgr.close();
    }

[abbccdda]

Oh, my point is that stateMgr in GlobalStateUpdateTask is taking in GlobalStateManagerImpl whose close(offsets) will do the checkpoint operation. So I guess we don't need to call checkpoint explicitly here right? @mjsax

[abbccdda]

@mjsax Thoughts?

[guozhangwang]

@mjsax @abbccdda @bbejeck I left a comment below for this issue. Please take a look.

[bbejeck]

LGTM, but I've left some comments regarding un-used method parameters as a result of this change.

[bbejeck]

I was thinking the same. Did you mean that the other change on this PR eliminates the chance for a double checkpoint file writes?

[bbejeck]

One note, with the removal of this line, the ackedOffsets is no longer used at all in this method. However, it's part of the StateManager#close interface, but it's not safe to remove as GlobalStateManager also implements the StateManager interface. Which to me, brings up two questions

1. Do we need to apply the same approach for global state stores?
2. Should we refactor the StateManager interface to have a no-args close() method?

Given the time constraints, I don't think should hold up this PR but it's something that IMHO we shouldn't let slip.

[guozhangwang]

Just laying out the context here (admittedly, this piece of logic is a bit hard to understand due to messy code structures, and we have a TODO task to clean up this tech debt soon):

StreamTask:

1. With EOS turned off, today we are actually double-checkpointing necessarily, once in suspend, and once in close. In fact, only one checkpoint is needed. This is known issue but did not incur any correctness issue, just unnecessary overhead.

2. With EOS turned off, we do not checkpoint on suspend, and only on close.

StandbyTask:

We always write the checkpoint in flushAndCheckpointState, which is called in both commit and suspend. Note that although we pass empty map into the checkpoint call, it is okay since it will be updated with the committed offsets internally. However, in fact in commit it is okay to just pass in null to NOT checkpoint at all, since there should be no offset change between the latest flushAndCheckpointState and the close call during normal processing (i.e. we are also duplicate-checkpointing here).

GlobalUpdateTask:

We pass in offsets in both stateConsumer.pollAndUpdate() and stateConsumer.close() but again this is the same duplicate-checkpointing issue as in StreamsTask case 1), because the offsets would never changed between the previous checkpoint call during normal run, and if there is an exception in between -- e.g. you checkpointed offset 100 in the latest pollAndUpdate, and in the next pollAndUpdate call you get an exception at offset 105, and hence jump into the finally blocked to call close -- we should not checkpoint 105 in this case since the data may not be flushed to the store at all, but rather just keep the checkpoint file with 100 to maintain at least once semantics.

With all this, let's do the following:

0. in StreamTask, the principle is that we always checkpoint on flush, and never checkpoint on close any more, regardless of EOS. And because of that:
1. Remove final Map<TopicPartition, Long> ackedOffsets from the close call, as @bbejeck suggested. In the StreamTask caller, we can safely remove the parameter since we are now never checkpointing on close.
2. Remove the parameter in AbstractTask#void closeStateManager(final boolean writeCheckpoint) as well and also remove the condition that writes checkpoint or not also. Since, again, no matter if we are closing cleanly or committing successfully, we would not write checkpoint files any more. And we can also remove the corresponding parameters like clean and commitSuccesfully in its ancestor call trace as well.
3. remove the parameter in stateMgr.close(offsets); in GlobalUpdateTask#close().

WDYT?

[guozhangwang]

Actually, about the action 0) above: since we now write checkpoint file at suspend as well when EOS is turned on, we should delete the file upon resumption we should also delete the checkpoint file as we did at the construction time:

if (eosEnabled) {
            // delete the checkpoint file after finish loading its stored offsets
            checkpoint.delete();
            checkpoint = null;
        }

so that the semantics is guaranteed: after resumption, if we get a crash, we should enforce bootstrapping from the beginning.

[guozhangwang]

EosIntegrationTest. shouldNotViolateEosIfOneTaskFailsWithState
GlobalStateManagerImplTest. shouldCheckpointRestoredOffsetsToFile
GlobalStateManagerImplTest. shouldWriteCheckpointsOnClose
GlobalStateTaskTest. shouldCloseStateManagerWithOffsets
StreamTaskTest. shouldNotCheckpointOffsetsOnCommitIfEosIsEnabled

Those tests failed locally.

[abbccdda]

Among the tests, I think
shouldCheckpointRestoredOffsetsToFile, shouldWriteCheckpointsOnClose,shouldCloseStateManagerWithOffsets,shouldNotCheckpointOffsetsOnCommitIfEosIsEnabled could be removed, because our change basically breaks the fundamental assumption behind them.
I will keep looking at EosIntegrationTest to figure out whether we should do sth to make it work.

[abbccdda]

Thanks @guozhangwang for the bug fix! I also rebased my test removal changes on top

[mjsax]

chechstyle error -- can you update the pr to fix it? @abbccdda

[mjsax]

nit: remove this

[mjsax]

To make the semantics clearer, I am wondering if we should use two nested if:

if(eosEnabled && !clean) {
  try {
    if (checkpoint != null) { ... }
  } catch(...) {...}
}

This make it clear, that it's a EOS condition (first var to be checked) and for the EOS, do something if !clean.

The checkpoint != null if just a guard against NPE and has nothing to do with the actual logic

[mjsax]

Why do we need to include clean here? (Was this another bug?) I thought for standbys this does not matter?

[guozhangwang]

Good point, since standby tasks do not have eos anyways today.

[mjsax]

nit: "Did not find checkpoint..."

[mjsax]

Thanks for the follow ups @guozhangwang LGTM.

[guozhangwang]

Cherry-picked to 2.2 as well.

[abbccdda]

@guozhangwang Thanks a lot for making the fix work!