You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2025/02/13 10:58:34 UTC
[PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
FrankYang0529 opened a new pull request, #18889:
URL: https://github.com/apache/kafka/pull/18889
In PR https://github.com/apache/kafka/pull/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.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18889:
URL: https://github.com/apache/kafka/pull/18889#discussion_r1955302461
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -18,16 +18,41 @@ package kafka.api
import kafka.utils.TestInfoUtils
import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.compress.Compression
+import org.apache.kafka.common.record.{AbstractRecords, CompressionType, MemoryRecords, RecordBatch, SimpleRecord, TimestampType}
+import org.apache.kafka.storage.internals.log.AppendOrigin
import org.junit.jupiter.api.Assertions.{assertEquals, assertNull, assertThrows}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.MethodSource
+import java.nio.ByteBuffer
import java.util
import java.util.{Collections, Optional}
import scala.jdk.CollectionConverters._
class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTest {
+ private def appendLegacyRecords(numRecords: Int, tp: TopicPartition, startingTimestamp: Long, brokerId: Int): Unit = {
+ val records = (0 until numRecords).map { i =>
+ 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,
+ TimestampType.CREATE_TIME, 0L, RecordBatch.NO_TIMESTAMP, RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH,
+ 0, false, RecordBatch.NO_PARTITION_LEADER_EPOCH)
+
+ records.foreach(builder.append)
+
+ brokers.filter(_.config.brokerId == brokerId).foreach(b => b.replicaManager.appendRecords(
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18889:
URL: https://github.com/apache/kafka/pull/18889#discussion_r1957730744
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -18,16 +18,43 @@ package kafka.api
import kafka.utils.TestInfoUtils
import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.compress.Compression
+import org.apache.kafka.common.record.{AbstractRecords, CompressionType, MemoryRecords, RecordBatch, RecordVersion, SimpleRecord, TimestampType}
import org.junit.jupiter.api.Assertions.{assertEquals, assertNull, assertThrows}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.MethodSource
+import java.nio.ByteBuffer
import java.util
import java.util.{Collections, Optional}
import scala.jdk.CollectionConverters._
class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTest {
+ private def appendLegacyRecords(numRecords: Int, tp: TopicPartition, startingTimestamp: Long, brokerId: Int): Unit = {
+ val records = (0 until numRecords).map { i =>
+ 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,
Review Comment:
Could you please keep tests for both v0 and v1?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #18889:
URL: https://github.com/apache/kafka/pull/18889
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18889:
URL: https://github.com/apache/kafka/pull/18889#discussion_r1957459755
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -18,16 +18,43 @@ package kafka.api
import kafka.utils.TestInfoUtils
import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.compress.Compression
+import org.apache.kafka.common.record.{AbstractRecords, CompressionType, MemoryRecords, RecordBatch, RecordVersion, SimpleRecord, TimestampType}
import org.junit.jupiter.api.Assertions.{assertEquals, assertNull, assertThrows}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.MethodSource
+import java.nio.ByteBuffer
import java.util
import java.util.{Collections, Optional}
import scala.jdk.CollectionConverters._
class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTest {
+ private def appendLegacyRecords(numRecords: Int, tp: TopicPartition, startingTimestamp: Long, brokerId: Int): Unit = {
+ val records = (0 until numRecords).map { i =>
+ 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,
Review 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.
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -80,12 +114,12 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
val timestampTopic2P0 = timestampOffsets.get(new TopicPartition(topic2, 0))
assertEquals(40, timestampTopic2P0.offset)
assertEquals(40, timestampTopic2P0.timestamp)
- assertEquals(Optional.of(0), timestampTopic2P0.leaderEpoch)
+ assertEquals(Optional.empty, timestampTopic2P0.leaderEpoch)
Review Comment:
yes, we don't set the leader epoch for the old message format.
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -51,8 +78,15 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
for (topic <- List(topic1, topic2, topic3)) {
for (part <- 0 until numParts) {
val tp = new TopicPartition(topic, part)
- // In sendRecords(), each message will have key, value and timestamp equal to the sequence number.
- sendRecords(producer, numRecords = 100, tp, startingTimestamp = 0)
+ // In sendRecords() and sendLegacyRecords(), each message will have key, value and timestamp equal to the sequence number.
+ if (topic == topic2) {
+ // append legacy records to topic2
+ appendLegacyRecords(100, tp, 0, part)
+ } else {
+ println("sendRecords")
Review Comment:
please remove this line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18889:
URL: https://github.com/apache/kafka/pull/18889#discussion_r1957057125
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -80,12 +114,12 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
val timestampTopic2P0 = timestampOffsets.get(new TopicPartition(topic2, 0))
assertEquals(40, timestampTopic2P0.offset)
assertEquals(40, timestampTopic2P0.timestamp)
- assertEquals(Optional.of(0), timestampTopic2P0.leaderEpoch)
+ assertEquals(Optional.empty, timestampTopic2P0.leaderEpoch)
Review Comment:
The commit https://github.com/apache/kafka/commit/8cdf1abb0b89f1f9f85fb1acb4cc8b83f527d16a 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.
https://github.com/apache/kafka/blob/c455c10405a4a201912e35a52ca1e6aef36e5af9/core/src/main/scala/kafka/server/ConfigHandler.scala#L173-L177
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18889:
URL: https://github.com/apache/kafka/pull/18889#issuecomment-2663020019
cherry-pick to 4.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18784: Fix ConsumerWithLegacyMessageFormatIntegrationTest [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18889:
URL: https://github.com/apache/kafka/pull/18889#discussion_r1957057125
##########
core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala:
##########
@@ -80,12 +114,12 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes
val timestampTopic2P0 = timestampOffsets.get(new TopicPartition(topic2, 0))
assertEquals(40, timestampTopic2P0.offset)
assertEquals(40, timestampTopic2P0.timestamp)
- assertEquals(Optional.of(0), timestampTopic2P0.leaderEpoch)
+ assertEquals(Optional.empty, timestampTopic2P0.leaderEpoch)
Review Comment:
The commit https://github.com/apache/kafka/commit/8cdf1abb0b89f1f9f85fb1acb4cc8b83f527d16a 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.
https://github.com/apache/kafka/blob/c455c10405a4a201912e35a52ca1e6aef36e5af9/core/src/main/scala/kafka/server/ConfigHandler.scala#L167-L172
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org