KAFKA-9169: fix standby checkpoint initialization #7681

vvcephei · 2019-11-11T22:50:21Z

Instead of caching the checkpoint map during StandbyTask initialization, use the latest checkpoints (which would have been updated during suspend).

Committer Checklist (excluded from commit message)

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

bbejeck

Thanks for the fix @vvcephei! LGTM. Should we add a test for this fix?

vvcephei · 2019-11-12T16:43:58Z

Thanks @bbejeck , not a bad idea...

I'll work on it.

omkreddy · 2019-11-12T18:37:34Z

@vvcephei Is this regression in 2.4 or existing bug from previous release? Normally we want to treat regression issues as blockers.

vvcephei · 2019-11-12T23:22:27Z

Thanks, @omkreddy ,

It's pretty hard to tell when the regression was introduced, because there are a large number of code paths involved.

However, I've just ported my test to the 2.3 branch, and it does look like it is broken there, too.

It sounds like it should not be a blocker, then. I'll downgrade the ticket.

vvcephei · 2019-11-13T06:44:47Z

FWIW, system tests from the original commit all passed:

Of course, I need to run them again, since I've changed a system test.

vvcephei

@bbejeck ,

Can you take another look? System tests are running already.

vvcephei · 2019-11-13T06:49:52Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java

-        log.trace("Checkpointable offsets updated with active acked offsets: {}", checkpointFileCache);
-
-        log.trace("Writing checkpoint: {}", checkpointFileCache);
+        log.debug("Writing checkpoint: {}", checkpointFileCache);


Removed a redundant log line and upgraded this log to DEBUG.

System tests run at DEBUG level, so this change makes this log visible to the system tests. Ideally, I could have just turned this class's logger to TRACE, but it happens to be a real pain with the way the system tests are set up.

I think this is reasonable, though, since this is not a high-volume log message, and it's pretty handy to know this information while debugging state restoration and initialization.

vvcephei · 2019-11-13T06:50:48Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java

@@ -195,7 +193,7 @@ public void close(final boolean clean,
    }

    Map<TopicPartition, Long> checkpointedOffsets() {
-        return checkpointedOffsets;
+        return Collections.unmodifiableMap(stateMgr.checkpointed());


The main fix for this PR. Report the current checkpointedOffsets, not the ones from when the standby task was first created.

vvcephei · 2019-11-13T06:52:34Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

@@ -440,6 +440,7 @@ private void assignStandbyPartitions() {
            checkpointedOffsets.putAll(standbyTask.checkpointedOffsets());
        }

+        log.debug("Assigning and seeking restoreConsumer to {}", checkpointedOffsets);


Adding a DEBUG log. This is also in support of the new system tests, but I think it's again nice to have this information available. I actually added this log while debugging the issue, before I wrote the test, to manually verify the bug.

vvcephei · 2019-11-13T06:52:46Z

tests/kafkatest/tests/streams/streams_standby_replica_test.py

+                                                     self.streams_sink_topic_1: {'partitions': 1,
+                                                                                 'replication-factor': 1},
+                                                     self.streams_sink_topic_2: {'partitions': 1,
+                                                                                 'replication-factor': 1}


lines were too long.

vvcephei · 2019-11-13T06:53:05Z

tests/kafkatest/tests/streams/streams_standby_replica_test.py

+        self.assert_consume(self.client_id, "assert all messages consumed from %s" % self.streams_sink_topic_1,
+                            self.streams_sink_topic_1, self.num_messages)
+        self.assert_consume(self.client_id, "assert all messages consumed from %s" % self.streams_sink_topic_2,
+                            self.streams_sink_topic_2, self.num_messages)


lines were too long

vvcephei · 2019-11-13T06:53:30Z

tests/kafkatest/tests/streams/streams_standby_replica_test.py


        wait_until(lambda: producer.num_acked >= self.num_messages,
                   timeout_sec=60,
                   err_msg="Failed to send all %s messages" % str(self.num_messages))

        producer.stop()

+        processor_1.stop()
+        processor_2.stop()
+        processor_3.stop()


There's no reason not to stop the processors at this point.

vvcephei · 2019-11-13T06:54:06Z

tests/kafkatest/tests/streams/streams_standby_replica_test.py

+
+        # Validate the checkpoint/restore logs for monotonicity
+        # This was added to ensure that standby restoration resumes from the checkpoint
+        # rather than the beginning of the changelog, as part of KAFKA-9169


This comment should explain the new verification I've added.

vvcephei · 2019-11-13T14:08:07Z

System tests are still passing:

vvcephei · 2019-11-13T14:16:38Z

Filed https://issues.apache.org/jira/browse/KAFKA-9179 for the unrelated JDK8/Scala11 test failure.

bbejeck

Thanks for the test @vvcephei. Overall I'm ok with the change to the system test, but I'm thinking a unit test is helpful as well.

I looked at the StandbyTaskTest and I updated the testUpdate() test so it will fail without your change.
After line 257 add this

final Map<TopicPartition, Long> originalCheckpointOffsets = task.checkpointedOffsets();
assertThat(originalCheckpointOffsets.get(partition2), equalTo(-1L));

Then after line 296 add this block

final Map<TopicPartition, Long> updatedCheckpointOffsets = task.checkpointedOffsets();
final long originalCheckpoint = originalCheckpointOffsets.get(partition2);
final long updatedCheckpoint = updatedCheckpointOffsets.get(partition2);
assertTrue("Original checkpoint should be less than updated checkpoint", originalCheckpoint < updatedCheckpoint);
assertThat(updatedCheckpoint, equalTo((31L)));

WDYT?

vvcephei · 2019-11-13T21:25:11Z

Thanks, @bbejeck , I think that unit test assertion is a good idea. I've added it.

I think it still makes sense to keep the system test, though, since the reasoning on the mailing list thread about this was that, although it's true the checkpoint wasn't updated on this path, it would be fine because the restore consumer wouldn't actually be re-assigned after suspend resumes, just un-paused. Verifying that, though requires testing almost all of Streams, since it includes the code paths of StreamThread, as well as StreamPartitionAssignor/TaskManager/etc.

I know you didn't object to the system test, just felt like presenting the argument to keep it, for posterity.

vvcephei · 2019-11-13T21:26:46Z

I'll wait for a green build and then merge. Thanks again for the review, @bbejeck !

KAFKA-9169: fix standby checkpoint initialization

f25ded9

bbejeck added the streams label Nov 12, 2019

bbejeck approved these changes Nov 12, 2019

View reviewed changes

vvcephei commented Nov 13, 2019

View reviewed changes

Add check to system test

b76a99a

vvcephei requested a review from bbejeck November 13, 2019 07:10

bbejeck reviewed Nov 13, 2019

View reviewed changes

add unit test

7413065

vvcephei merged commit cac8560 into apache:trunk Nov 14, 2019

vvcephei deleted the KAFKA-9169 branch November 14, 2019 04:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-9169: fix standby checkpoint initialization #7681

KAFKA-9169: fix standby checkpoint initialization #7681

vvcephei commented Nov 11, 2019

bbejeck left a comment

vvcephei commented Nov 12, 2019

omkreddy commented Nov 12, 2019 •

edited

Loading

vvcephei commented Nov 12, 2019

vvcephei commented Nov 13, 2019

vvcephei left a comment

vvcephei Nov 13, 2019

vvcephei Nov 13, 2019

vvcephei Nov 13, 2019

vvcephei Nov 13, 2019

vvcephei Nov 13, 2019

vvcephei Nov 13, 2019

vvcephei Nov 13, 2019

vvcephei commented Nov 13, 2019

vvcephei commented Nov 13, 2019

bbejeck left a comment

vvcephei commented Nov 13, 2019

vvcephei commented Nov 13, 2019

KAFKA-9169: fix standby checkpoint initialization #7681

KAFKA-9169: fix standby checkpoint initialization #7681

Conversation

vvcephei commented Nov 11, 2019

Committer Checklist (excluded from commit message)

bbejeck left a comment

Choose a reason for hiding this comment

vvcephei commented Nov 12, 2019

omkreddy commented Nov 12, 2019 • edited Loading

vvcephei commented Nov 12, 2019

vvcephei commented Nov 13, 2019

vvcephei left a comment

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei Nov 13, 2019

Choose a reason for hiding this comment

vvcephei commented Nov 13, 2019

vvcephei commented Nov 13, 2019

bbejeck left a comment

Choose a reason for hiding this comment

vvcephei commented Nov 13, 2019

vvcephei commented Nov 13, 2019

omkreddy commented Nov 12, 2019 •

edited

Loading