You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/18 07:22:46 UTC
[GitHub] [kafka] ijuma commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
ijuma commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559357615
##########
File path: core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
##########
@@ -794,9 +797,12 @@ class PlaintextConsumerTest extends BaseConsumerTest {
awaitAssignment(consumer, partitions.toSet)
val producer = createProducer()
- val producerRecords = partitions.flatMap(sendRecords(producer, numRecords = 15, _))
+ // it'll send 30 records to each topic partition (3 topics and 30 partitions each topic)
+ val producerRecords = partitions.flatMap(sendRecords(producer, numRecords = partitionCount, _, needCurrentTime = true))
Review comment:
Why did we change the number of records to match the partition count?
##########
File path: core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala
##########
@@ -103,9 +103,10 @@ abstract class AbstractConsumerTest extends BaseRequestTest {
}
protected def sendRecords(producer: KafkaProducer[Array[Byte], Array[Byte]], numRecords: Int,
- tp: TopicPartition): Seq[ProducerRecord[Array[Byte], Array[Byte]]] = {
+ tp: TopicPartition, needCurrentTime: Boolean): Seq[ProducerRecord[Array[Byte], Array[Byte]]] = {
Review comment:
Do we need this boolean? Could we always set the current time? It seems like using the existing behavior may cause problems to many other tests too.
##########
File path: core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
##########
@@ -794,9 +797,12 @@ class PlaintextConsumerTest extends BaseConsumerTest {
awaitAssignment(consumer, partitions.toSet)
val producer = createProducer()
- val producerRecords = partitions.flatMap(sendRecords(producer, numRecords = 15, _))
+ // it'll send 30 records to each topic partition (3 topics and 30 partitions each topic)
+ val producerRecords = partitions.flatMap(sendRecords(producer, numRecords = partitionCount, _, needCurrentTime = true))
+
val consumerRecords = consumeRecords(consumer, producerRecords.size)
+ // the expected records should be 30 records * 3 topics * 30 partitions = 2700
Review comment:
What is the purpose of this and the other comment?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org