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/13 10:04:00 UTC
[GitHub] [kafka] showuon opened a new pull request #9877: KAFKA-8460: consume records right after awaitAssignment
showuon opened a new pull request #9877:
URL: https://github.com/apache/kafka/pull/9877
Originally, we make sure consumer `awaitAssignment`, and then produce records. We total send 30 records to 3 topics, and each topic has 30 partitions, so it takes some time to process it. If it exceeds 6 seconds, it'll make the consumer left due to the `max.poll.interval.ms` is set to 6 secs. And then it will delete the logs and then increment the start offset. So, later when the consumer is back, we'll first listOffset, and got the new and unexpected start offset. That's why we sometimes cannot receive expected amount of records.
To fix it, I produce records before `awaitAssignment`, and then, I make sure we consume records right after awaitAssignemnt. This will make this test much more reliable.
### 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-763394472
failed tests:
```
Build / JDK 8 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddCurSizeAllMemTablesMetric
Build / JDK 8 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddNumLiveVersionMetric
Build / JDK 8 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddEstimateNumKeysMetric
Build / JDK 8 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddCurSizeAllMemTablesMetric
Build / JDK 8 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddNumLiveVersionMetric
Build / JDK 8 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddEstimateNumKeysMetric
Build / JDK 15 / org.apache.kafka.clients.consumer.KafkaConsumerTest.testCloseWithTimeUnit()
Build / JDK 15 / org.apache.kafka.clients.consumer.internals.FetcherTest.testEarlierOffsetResetArrivesLate()
Build / JDK 15 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddNumLiveVersionMetric
Build / JDK 15 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddEstimateNumKeysMetric
Build / JDK 15 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddCurSizeAllMemTablesMetric
Build / JDK 15 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddCurSizeAllMemTablesMetric
Build / JDK 15 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddNumLiveVersionMetric
Build / JDK 15 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddEstimateNumKeysMetric
Build / JDK 11 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddCurSizeAllMemTablesMetric
Build / JDK 11 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddNumLiveVersionMetric
Build / JDK 11 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddEstimateNumKeysMetric
Build / JDK 11 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddCurSizeAllMemTablesMetric
Build / JDK 11 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddNumLiveVersionMetric
Build / JDK 11 / org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsTest.shouldAddEstimateNumKeysMetric
```
Mostly they are failed in `RocksDBMetricsTest`, which is addressed in #9935 . The other tests are not related to my change. Thanks.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762730140
Thanks. I tweaked things a bit in the last commit here:
https://github.com/apache/kafka/pull/9927/commits/1ecdd58e682a2572fe798d6249cd185655978da2
If you think it looks good, please add it to your PR and please merge trunk.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762065711
Let's see if it'll break other tests.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-763055332
@showuon This test seems to be failing a lot, so it would be good to integrate the proposed changes (if you agree) so that we can merge it.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559364510
##########
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:
Sounds good. Yes, if some tests fail after this change, we can use the previous behavior in those tests. And the new behavior should be the default.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-763303344
@ijuma , Thanks for helping refactor the codes. I've fix the conflict and make sure every tests extends `BaseConsumerTest` all passed. Let's wait the jenkins test results. Thank you.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-759343202
@hachikuji , could you help review this PR? Thanks.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-760953842
Can we just bump `max.poll.interval.ms`? It doesn't seem like there's any value in having that be low in this test.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559364510
##########
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:
Sounds good. Yes, if some tests fail after this change, we can use the previous behavior in those tests (via a default argument instead of method overload). And the new behavior should be the default.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon edited a comment on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon edited a comment on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762020474
@ijuma , thanks for the comments. I agree with you that it might changes what it was testing. In this commit(https://github.com/apache/kafka/pull/9877/commits/a43e72a13efbebc82dd436659f7ff62ec8590d52), I updated:
1. revert what I did
2. increase `max.poll.interval.ms` to avoid unnecessary consumer rejoin
3. send records with current timestamp, so the log cleaner won't clean the records if the record sending takes too much of time.
I found the (3) is the main reason the test will fail because of log cleaner will delete the logs too old, and we'll get nothing to consume. So we need to set the records to the current timestamp ( or set to `compact` only, of course, I prefer to set timestamp :) ). I still increase the `max.poll.interval.ms` because it'll avoid the consumer rejoin to improve the test performance.
Thank you.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559364033
##########
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:
Sounds good. Worth mentioning it in the PR description.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559366732
##########
File path: core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
##########
@@ -772,6 +772,9 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// this behaves a little different than when remaining limit bytes is 0 and it's important to test it
this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "500")
this.consumerConfig.setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "100")
+
+ // set the 'max.poll.interval.ms' larger (default is 6 seconds), to avoid the consumer got kicked out during sending records
Review comment:
This comment can be more concise. Something like:
`Avoid a rebalance while the records are being sent (the default is 6 seconds)`
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559378385
##########
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:
OK, now, the default `sendRecord(producer, numRecords, tp)` will send with current timestamp. Some tests will verity the timestamp after consuming, so we should keep as the old behavior.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-760952738
Not sure what the original purpose of the test was, but this changes what it's testing, right? Before producing was happening after subscription and now before.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon removed a comment on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon removed a comment on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-759343202
@hachikuji , could you help review this PR? Thanks.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-763228241
@ijuma , sorry I was busy yesterday. I'll work on it today. Thanks
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762020474
@ijuma , thanks for the comments. I agree with you that it might changes what it was testing. In this commit(https://github.com/apache/kafka/pull/9877/commits/a43e72a13efbebc82dd436659f7ff62ec8590d52), I updated:
1. increase `max.poll.interval.ms` to avoid unnecessary consumer rejoin
2. send records with current timestamp, so the log cleaner won't clean the records if the record sending takes too much of time.
Thank you.
----------------------------------------------------------------
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
[GitHub] [kafka] chia7712 merged pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
chia7712 merged pull request #9877:
URL: https://github.com/apache/kafka/pull/9877
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: produce records with current timestamp
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762200358
test jdk 11/15 passed. jdk8 failed test is addressed in #9888
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-759344449
@hachikuji , could you help review this PR? Thanks.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559361106
##########
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:
Let me remove it.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559376682
##########
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:
Thanks for reminding. Updated in the PR description.
##########
File path: core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
##########
@@ -772,6 +772,9 @@ class PlaintextConsumerTest extends BaseConsumerTest {
// this behaves a little different than when remaining limit bytes is 0 and it's important to test it
this.consumerConfig.setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "500")
this.consumerConfig.setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "100")
+
+ // set the 'max.poll.interval.ms' larger (default is 6 seconds), to avoid the consumer got kicked out during sending records
Review comment:
Updated.
----------------------------------------------------------------
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
[GitHub] [kafka] ijuma commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
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
[GitHub] [kafka] ijuma commented on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-760954112
Also, thanks so much for diagnosing this! Very awesome. :)
----------------------------------------------------------------
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
[GitHub] [kafka] showuon edited a comment on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon edited a comment on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762020474
@ijuma , thanks for the comments. I agree with you that it might changes what it was testing. In this commit(https://github.com/apache/kafka/pull/9877/commits/a43e72a13efbebc82dd436659f7ff62ec8590d52), I updated:
1. revert what I did
2. increase `max.poll.interval.ms` to avoid unnecessary consumer rejoin
3. send records with current timestamp, so the log cleaner won't clean the records if the record sending takes too much of time.
I found the (3) is the main reason the test will fail because of log cleaner will delete the logs too old, and we'll get nothing to consume. So we need to set the records to the current timestamp ( or set to log.cleanup.policy to `compact` only, of course, I prefer to set timestamp :) ). I still increase the `max.poll.interval.ms` because it'll avoid the consumer rejoin to improve the test performance.
Thank you.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559361405
##########
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:
Of course, I'm just afraid of some tests relying on the original "fake time". Let me test it.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon edited a comment on pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon edited a comment on pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#issuecomment-762020474
@ijuma , thanks for the comments. I agree with you that it might changes what it was testing. In this commit(https://github.com/apache/kafka/pull/9877/commits/a43e72a13efbebc82dd436659f7ff62ec8590d52), I updated:
1. revert what I did
2. increase `max.poll.interval.ms` to avoid unnecessary consumer rejoin
3. send records with current timestamp, so the log cleaner won't clean the records if the record sending takes too much of time.
Thank you.
----------------------------------------------------------------
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
[GitHub] [kafka] showuon commented on a change in pull request #9877: KAFKA-8460: consume records right after awaitAssignment
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9877:
URL: https://github.com/apache/kafka/pull/9877#discussion_r559361000
##########
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:
Actually, the `partitionCount` is the original numRecords. And I tried to reduce the records number months ago, and still found it failed. So that's why I deep dive this flaky test. :) So, I just revert the previous change.
----------------------------------------------------------------
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