KAFKA-9080: Addresses MessageFormatChangeTest.testCompatibilty with version 0.9.0.1

[tuvtran]

#7167 added a check for non-incremental offsets in assignOffsetsNonCompressed, which is not applicable for message format V0 and V1. Therefore, I added a condition to disable the check if the record version precedes V2.

Committer Checklist (excluded from commit message)

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

[tuvtran]

All MessageFormatChangeTest tests passed.

[tuvtran]

@omkreddy @guozhangwang ready for reviews

[guozhangwang]

@tuvtran I took a look at the PR but still cannot understand why for V0/V1 continuous offsets no longer hold true: with V0/V1 the inner records use the base offset + relative offset which should also be continuous, right?

[junrao]

@tuvtran : Looking at the logic in DeepRecordsIterator, it seems there is a difference btw V0 and V1 message format. In V1, the wrapper message has the absolute base offset and each inner message has the relative offset. In V0, each inner message has the absolute offset and the offset in the wrapper message is ignored. So, we need to distinguish between these 2 cases during offset validation.

[omkreddy]

Can we revert the check added in assignOffsetsNonCompressed ?, so that we can continue with existing behavior in 2.4. We can add complete fix in next release.

[tuvtran]

Yeah I can create a separate JIRA for V0/V1 offset validation.

[tuvtran]

retest this please

[guozhangwang]

@tuvtran : Looking at the logic in DeepRecordsIterator, it seems there is a difference btw V0 and V1 message format. In V1, the wrapper message has the absolute base offset and each inner message has the relative offset. In V0, each inner message has the absolute offset and the offset in the wrapper message is ignored. So, we need to distinguish between these 2 cases during offset validation.

What confused me here is that, in either V0, V1 or V2 case, the inner message offset (either it is inferred as base + relative, or it is encoded as absolute) should always be continuous right? @junrao @tuvtran

[junrao]

@guozhangwang : It seems the issue is that expectedInnerOffset always starts at offset 0. This will fail the test if (record.offset != expectedOffset) for V0 message since each inner offset is the absolute offset.

[omkreddy]

@tuvtran Thanks for the PR. LGTM. Please raise JIRA to handle V0/V1 offset validation.

[ijuma]

Can we make sure to include a unit test that verifies this fix? Generally, we should aim to do that before merging.

[guozhangwang]

@guozhangwang : It seems the issue is that expectedInnerOffset always starts at offset 0. This will fail the test if (record.offset != expectedOffset) for V0 message since each inner offset is the absolute offset.

That makes sense, thanks!