KAFKA-7443: OffsetOutOfRangeException in restoring state store from changelog topic when start offset of local checkpoint is smaller than that of changelog topic

[linyli001]

See https://issues.apache.org/jira/browse/KAFKA-7443 for details.
The fix set this state partition to "NO_CHECKPOINT" when the offset in local checkpoint file has expired and older than the current start offset of changelog topic, thus making this task to restore local state from the current beginning offset of changelog topic, avoiding falling into the infinite loop caused by this exception.

[mjsax]

Thanks for the PR @linyli001. Can we add a unit (or integration) test for this case?

[mjsax]

There is a checkstyle error:

ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11@2/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:112:1: File contains tab characters (this is the first instance). [FileTabCharacter]

[vvcephei]

The test results are gone already, for some reason...

[vvcephei]

Retest this, please.

[mjsax]

@linyli001 Can we add a test for this case?

[bbejeck]

The change looks good to me, but I agree with @mjsax we'll need an integration test for this.

[linyli001]

@mjsax Ok, I'll add an integration test. But the scenario only triggers under certain conditions: the changelog topic deleted the expired segments and updated the new starting offset, which is greater than the offset recorded in local checkpoint file. I'm confused how to simulate this scenario in my test case, so could you give me some suggestions?

[bbejeck]

@linyli001 I see your point, just off the top of my head without looking at the code, could you simulate the condition better using mocks? In that case, you'd only need a unit test.

[mjsax]

I agree that it's a little tricky... A mocked unit test should be possible for sure as suggested by Bill.

For an integration test, you could create a topic with very small segment size, write data into it, and user purgeDataAPI to delete old records. Then, a checkpoint file with old offsets is written and afterwards Streams is started and we check if it recovers correctly.

Does this help? Thoughts?

[vvcephei]

I think we'd need to see the unit test to decide if it's complete enough, but it feels like enough to me...

The fundamental question is whether you trust the client to throw that exception when the offset is ... out of range. If that's the specified behavior of the client, then we don't need an integration test, we just need to make sure that Streams properly clears the offset when it gets the exception.

In other words, we should be able to test it satisfactorily with a mock.

[linyli001]

Ok. I will add the test case according to @mjsax. Thanks!

[linyli001]

The unit test has some problems, and I'll check and push a new commit latter.

[mjsax]

Thanks for adding the test. Some nits. Overall LGTM.

[vvcephei]

This looks good to me, @linyli001 .

Thanks!

[vvcephei]

@mjsax , is this ready to merge?

[mjsax]

Thanks for the PR @linyli001!

Merged to trunk and cherry-picked to 2.1, 2.0, and 1.1.

[linyli001]

@mjsax I have some problems when sync with the latest change. I can see my fix has been merged into the trunk branch, but when I folk the apache/kafka to my GitHub, I can't found the latest change. Any steps am I missing?

[vvcephei]

Hi @linyli001 ,

IIRC, you have to check out your trunk branch, then pull from apache/kafka's trunk, then push it to your fork's trunk.

Is that what you're already trying to do?

[linyli001]

@vvcephei Sorry I don't quite understand how to do it. I have cloned the git from my fork to local, but run "git pull origin trunk" not includes the latest change. I think it just pull from my fork's trunk, but I don't know how to pull from apache/kafka's trunk locally? Can you show me the git command? Thanks!

[mjsax]

If you fork the repo, your copy on github won't be updated automatically. You need to pull the latest changes from kafka repo into your local one (and than you can push those back to you copy on github).

Assume you have an alias called kafka-github (you can create one with git add remote... command) and assume you are on local branch trunk:

git fetch kafka-github              // get latest changes
git rebase kafka-github trunk       // update local trunk to latest changes
git push origin trunk               // push the changes to you github copy

[linyli001]

@mjsax Thanks for the guide! I can update my folk now.