KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest#18889
KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest#18889chia7712 merged 1 commit intoapache:trunkfrom
Conversation
github-actions
Bot
added
core
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@FrankYang0529 thanks for this patch. one major comment is left.
|
|
||
| records.foreach(builder.append) | ||
|
|
||
| brokers.filter(_.config.brokerId == brokerId).foreach(b => b.replicaManager.appendRecords( |
There was a problem hiding this comment.
replicaManager.appendRecords performs the conversion if the magic of the batch is not equal to v2. Therefore, to handle older record formats, you should follow @ijuma's suggestion in the jira to utilize the Log class for writing records in the legacy format.
FrankYang0529
force-pushed
the
KAFKA-18784
branch
from
f8ccc83 to
77b62bc
Compare
dajac
added
the
Blocker
FrankYang0529
force-pushed
the
KAFKA-18784
branch
from
77b62bc to
f474a2d
Compare
| assertEquals(40, timestampTopic2P0.offset) | ||
| assertEquals(40, timestampTopic2P0.timestamp) | ||
| assertEquals(Optional.of(0), timestampTopic2P0.leaderEpoch) | ||
| assertEquals(Optional.empty, timestampTopic2P0.leaderEpoch) |
There was a problem hiding this comment.
The commit 8cdf1ab included KRaft support for ConsumerWithLegacyMessageFormatIntegrationTest. However, it didn't really set topic to use v0 format. The KRaft used inter broker protocol version 3.0-IV1, so the message format version is ignored. That's why we could get leader epoch for topic2.
kafka/core/src/main/scala/kafka/server/ConfigHandler.scala
Lines 167 to 172 in c455c10
There was a problem hiding this comment.
yes, we don't set the leader epoch for the old message format.
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@FrankYang0529 thanks for this patch. please take a look at following two minor comments
| // append legacy records to topic2 | ||
| appendLegacyRecords(100, tp, 0, part) | ||
| } else { | ||
| println("sendRecords") |
| assertEquals(40, timestampTopic2P0.offset) | ||
| assertEquals(40, timestampTopic2P0.timestamp) | ||
| assertEquals(Optional.of(0), timestampTopic2P0.leaderEpoch) | ||
| assertEquals(Optional.empty, timestampTopic2P0.leaderEpoch) |
There was a problem hiding this comment.
yes, we don't set the leader epoch for the old message format.
| new SimpleRecord(startingTimestamp + i, s"key $i".getBytes, s"value $i".getBytes) | ||
| } | ||
| val buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(RecordBatch.MAGIC_VALUE_V1, CompressionType.NONE, records.asJava)) | ||
| val builder = MemoryRecords.builder(buffer, RecordBatch.MAGIC_VALUE_V1, Compression.of(CompressionType.NONE).build, |
There was a problem hiding this comment.
Could you please add test for v0 also? in v0 we don't write the timestamp, so searching record by timestamp should return null.
FrankYang0529
force-pushed
the
KAFKA-18784
branch
from
f474a2d to
e9212a5
Compare
| new SimpleRecord(startingTimestamp + i, s"key $i".getBytes, s"value $i".getBytes) | ||
| } | ||
| val buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(RecordBatch.MAGIC_VALUE_V0, CompressionType.NONE, records.asJava)) | ||
| val builder = MemoryRecords.builder(buffer, RecordBatch.MAGIC_VALUE_V0, Compression.of(CompressionType.NONE).build, |
There was a problem hiding this comment.
Could you please keep tests for both v0 and v1?
Signed-off-by: PoAn Yang <payang@apache.org>
FrankYang0529
force-pushed
the
KAFKA-18784
branch
from
e9212a5 to
185b6f8
Compare
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
LGTM
the flaky is traced by https://issues.apache.org/jira/browse/KAFKA-15900
In PR #18267, we removed old message format for cases in ConsumerWithLegacyMessageFormatIntegrationTest. Although test cases can pass, they don't fulfill original purpose. We can't send old message format since 4.0, so I change cases to append old records by ReplicaManager directly. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
|
cherry-pick to 4.0 |
…e#18889) In PR apache#18267, we removed old message format for cases in ConsumerWithLegacyMessageFormatIntegrationTest. Although test cases can pass, they don't fulfill original purpose. We can't send old message format since 4.0, so I change cases to append old records by ReplicaManager directly. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
3 participants
In PR #18267, we removed old message format for cases in
ConsumerWithLegacyMessageFormatIntegrationTest. Although test cases can pass, they don't fulfill original purpose. We can't send old message format since 4.0, so I change cases to append old records byReplicaManagerdirectly.Committer Checklist (excluded from commit message)