You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/03/28 13:16:02 UTC
(kafka) branch trunk updated: KAFKA-16406 [2] : Split consumer commit tests (#15612)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new b411ac0c700 KAFKA-16406 [2] : Split consumer commit tests (#15612)
b411ac0c700 is described below
commit b411ac0c700fcb74639502cf38181de6096b6404
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Thu Mar 28 14:15:56 2024 +0100
KAFKA-16406 [2] : Split consumer commit tests (#15612)
Follow-up to #15535, splitting consumer integration tests defined in the long-running PlainTextConsumerTest. This PR extracts the tests that directly relate to committing offsets. No changes in logic.
Reviewers: Lucas Brutschy <lb...@confluent.io>
---
.../kafka/api/PlaintextConsumerCommitTest.scala | 320 +++++++++++++++++++++
.../kafka/api/PlaintextConsumerTest.scala | 279 +-----------------
2 files changed, 321 insertions(+), 278 deletions(-)
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerCommitTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerCommitTest.scala
new file mode 100644
index 00000000000..db5ef480d14
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerCommitTest.scala
@@ -0,0 +1,320 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
+ * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
+ * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package kafka.api
+
+import kafka.utils.{TestInfoUtils, TestUtils}
+import org.apache.kafka.clients.consumer._
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.test.MockConsumerInterceptor
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Timeout
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.{Arguments, MethodSource}
+
+import java.time.Duration
+import java.util
+import java.util.Optional
+import java.util.stream.Stream
+import scala.jdk.CollectionConverters._
+
+/**
+ * Integration tests for the consumer that covers the logic related to committing offsets.
+ */
+@Timeout(600)
+class PlaintextConsumerCommitTest extends AbstractConsumerTest {
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testAutoCommitOnClose(quorum: String, groupProtocol: String): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
+ val consumer = createConsumer()
+
+ val numRecords = 10000
+ val producer = createProducer()
+ sendRecords(producer, numRecords, tp)
+
+ consumer.subscribe(List(topic).asJava)
+ awaitAssignment(consumer, Set(tp, tp2))
+
+ // should auto-commit sought positions before closing
+ consumer.seek(tp, 300)
+ consumer.seek(tp2, 500)
+ consumer.close()
+
+ // now we should see the committed positions from another consumer
+ val anotherConsumer = createConsumer()
+ assertEquals(300, anotherConsumer.committed(Set(tp).asJava).get(tp).offset)
+ assertEquals(500, anotherConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testAutoCommitOnCloseAfterWakeup(quorum: String, groupProtocol: String): Unit = {
+ this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
+ val consumer = createConsumer()
+
+ val numRecords = 10000
+ val producer = createProducer()
+ sendRecords(producer, numRecords, tp)
+
+ consumer.subscribe(List(topic).asJava)
+ awaitAssignment(consumer, Set(tp, tp2))
+
+ // should auto-commit sought positions before closing
+ consumer.seek(tp, 300)
+ consumer.seek(tp2, 500)
+
+ // wakeup the consumer before closing to simulate trying to break a poll
+ // loop from another thread
+ consumer.wakeup()
+ consumer.close()
+
+ // now we should see the committed positions from another consumer
+ val anotherConsumer = createConsumer()
+ assertEquals(300, anotherConsumer.committed(Set(tp).asJava).get(tp).offset)
+ assertEquals(500, anotherConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testCommitMetadata(quorum: String, groupProtocol: String): Unit = {
+ val consumer = createConsumer()
+ consumer.assign(List(tp).asJava)
+
+ // sync commit
+ val syncMetadata = new OffsetAndMetadata(5, Optional.of(15), "foo")
+ consumer.commitSync(Map((tp, syncMetadata)).asJava)
+ assertEquals(syncMetadata, consumer.committed(Set(tp).asJava).get(tp))
+
+ // async commit
+ val asyncMetadata = new OffsetAndMetadata(10, "bar")
+ sendAndAwaitAsyncCommit(consumer, Some(Map(tp -> asyncMetadata)))
+ assertEquals(asyncMetadata, consumer.committed(Set(tp).asJava).get(tp))
+
+ // handle null metadata
+ val nullMetadata = new OffsetAndMetadata(5, null)
+ consumer.commitSync(Map(tp -> nullMetadata).asJava)
+ assertEquals(nullMetadata, consumer.committed(Set(tp).asJava).get(tp))
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testAsyncCommit(quorum: String, groupProtocol: String): Unit = {
+ val consumer = createConsumer()
+ consumer.assign(List(tp).asJava)
+
+ val callback = new CountConsumerCommitCallback
+ val count = 5
+
+ for (i <- 1 to count)
+ consumer.commitAsync(Map(tp -> new OffsetAndMetadata(i)).asJava, callback)
+
+ TestUtils.pollUntilTrue(consumer, () => callback.successCount >= count || callback.lastError.isDefined,
+ "Failed to observe commit callback before timeout", waitTimeMs = 10000)
+
+ assertEquals(None, callback.lastError)
+ assertEquals(count, callback.successCount)
+ assertEquals(new OffsetAndMetadata(count), consumer.committed(Set(tp).asJava).get(tp))
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testAutoCommitIntercept(quorum: String, groupProtocol: String): Unit = {
+ val topic2 = "topic2"
+ createTopic(topic2, 2, brokerCount)
+
+ // produce records
+ val numRecords = 100
+ val testProducer = createProducer(keySerializer = new StringSerializer, valueSerializer = new StringSerializer)
+ (0 until numRecords).map { i =>
+ testProducer.send(new ProducerRecord(tp.topic(), tp.partition(), s"key $i", s"value $i"))
+ }.foreach(_.get)
+
+ // create consumer with interceptor
+ this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
+ this.consumerConfig.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.apache.kafka.test.MockConsumerInterceptor")
+ val testConsumer = createConsumer(keyDeserializer = new StringDeserializer, valueDeserializer = new StringDeserializer)
+ val rebalanceListener = new ConsumerRebalanceListener {
+ override def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) = {
+ // keep partitions paused in this test so that we can verify the commits based on specific seeks
+ testConsumer.pause(partitions)
+ }
+
+ override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) = {}
+ }
+ changeConsumerSubscriptionAndValidateAssignment(testConsumer, List(topic), Set(tp, tp2), rebalanceListener)
+ testConsumer.seek(tp, 10)
+ testConsumer.seek(tp2, 20)
+
+ // change subscription to trigger rebalance
+ val commitCountBeforeRebalance = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue()
+ changeConsumerSubscriptionAndValidateAssignment(testConsumer,
+ List(topic, topic2),
+ Set(tp, tp2, new TopicPartition(topic2, 0), new TopicPartition(topic2, 1)),
+ rebalanceListener)
+
+ // after rebalancing, we should have reset to the committed positions
+ assertEquals(10, testConsumer.committed(Set(tp).asJava).get(tp).offset)
+ assertEquals(20, testConsumer.committed(Set(tp2).asJava).get(tp2).offset)
+
+ // In both CLASSIC and CONSUMER protocols, interceptors are executed in poll and close.
+ // However, in the CONSUMER protocol, the assignment may be changed outside of a poll, so
+ // we need to poll once to ensure the interceptor is called.
+ if (groupProtocol.toUpperCase == GroupProtocol.CONSUMER.name) {
+ testConsumer.poll(Duration.ZERO);
+ }
+
+ assertTrue(MockConsumerInterceptor.ON_COMMIT_COUNT.intValue() > commitCountBeforeRebalance)
+
+ // verify commits are intercepted on close
+ val commitCountBeforeClose = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue()
+ testConsumer.close()
+ assertTrue(MockConsumerInterceptor.ON_COMMIT_COUNT.intValue() > commitCountBeforeClose)
+ testProducer.close()
+
+ // cleanup
+ MockConsumerInterceptor.resetCounters()
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testCommitSpecifiedOffsets(quorum: String, groupProtocol: String): Unit = {
+ val producer = createProducer()
+ sendRecords(producer, numRecords = 5, tp)
+ sendRecords(producer, numRecords = 7, tp2)
+
+ val consumer = createConsumer()
+ consumer.assign(List(tp, tp2).asJava)
+
+ val pos1 = consumer.position(tp)
+ val pos2 = consumer.position(tp2)
+ consumer.commitSync(Map[TopicPartition, OffsetAndMetadata]((tp, new OffsetAndMetadata(3L))).asJava)
+ assertEquals(3, consumer.committed(Set(tp).asJava).get(tp).offset)
+ assertNull(consumer.committed(Set(tp2).asJava).get(tp2))
+
+ // Positions should not change
+ assertEquals(pos1, consumer.position(tp))
+ assertEquals(pos2, consumer.position(tp2))
+ consumer.commitSync(Map[TopicPartition, OffsetAndMetadata]((tp2, new OffsetAndMetadata(5L))).asJava)
+ assertEquals(3, consumer.committed(Set(tp).asJava).get(tp).offset)
+ assertEquals(5, consumer.committed(Set(tp2).asJava).get(tp2).offset)
+
+ // Using async should pick up the committed changes after commit completes
+ sendAndAwaitAsyncCommit(consumer, Some(Map(tp2 -> new OffsetAndMetadata(7L))))
+ assertEquals(7, consumer.committed(Set(tp2).asJava).get(tp2).offset)
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testAutoCommitOnRebalance(quorum: String, groupProtocol: String): Unit = {
+ val topic2 = "topic2"
+ createTopic(topic2, 2, brokerCount)
+
+ this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
+ val consumer = createConsumer()
+
+ val numRecords = 10000
+ val producer = createProducer()
+ sendRecords(producer, numRecords, tp)
+
+ val rebalanceListener = new ConsumerRebalanceListener {
+ override def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) = {
+ // keep partitions paused in this test so that we can verify the commits based on specific seeks
+ consumer.pause(partitions)
+ }
+
+ override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) = {}
+ }
+
+ consumer.subscribe(List(topic).asJava, rebalanceListener)
+
+ awaitAssignment(consumer, Set(tp, tp2))
+
+ consumer.seek(tp, 300)
+ consumer.seek(tp2, 500)
+
+ // change subscription to trigger rebalance
+ consumer.subscribe(List(topic, topic2).asJava, rebalanceListener)
+
+ val newAssignment = Set(tp, tp2, new TopicPartition(topic2, 0), new TopicPartition(topic2, 1))
+ awaitAssignment(consumer, newAssignment)
+
+ // after rebalancing, we should have reset to the committed positions
+ assertEquals(300, consumer.committed(Set(tp).asJava).get(tp).offset)
+ assertEquals(500, consumer.committed(Set(tp2).asJava).get(tp2).offset)
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testSubscribeAndCommitSync(quorum: String, groupProtocol: String): Unit = {
+ // This test ensure that the member ID is propagated from the group coordinator when the
+ // assignment is received into a subsequent offset commit
+ val consumer = createConsumer()
+ assertEquals(0, consumer.assignment.size)
+ consumer.subscribe(List(topic).asJava)
+ awaitAssignment(consumer, Set(tp, tp2))
+
+ consumer.seek(tp, 0)
+
+ consumer.commitSync()
+ }
+
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testPositionAndCommit(quorum: String, groupProtocol: String): Unit = {
+ val producer = createProducer()
+ var startingTimestamp = System.currentTimeMillis()
+ sendRecords(producer, numRecords = 5, tp, startingTimestamp = startingTimestamp)
+
+ val topicPartition = new TopicPartition(topic, 15)
+ val consumer = createConsumer()
+ assertNull(consumer.committed(Set(topicPartition).asJava).get(topicPartition))
+
+ // position() on a partition that we aren't subscribed to throws an exception
+ assertThrows(classOf[IllegalStateException], () => consumer.position(topicPartition))
+
+ consumer.assign(List(tp).asJava)
+
+ assertEquals(0L, consumer.position(tp), "position() on a partition that we are subscribed to should reset the offset")
+ consumer.commitSync()
+ assertEquals(0L, consumer.committed(Set(tp).asJava).get(tp).offset)
+ consumeAndVerifyRecords(consumer = consumer, numRecords = 5, startingOffset = 0, startingTimestamp = startingTimestamp)
+ assertEquals(5L, consumer.position(tp), "After consuming 5 records, position should be 5")
+ consumer.commitSync()
+ assertEquals(5L, consumer.committed(Set(tp).asJava).get(tp).offset, "Committed offset should be returned")
+
+ startingTimestamp = System.currentTimeMillis()
+ sendRecords(producer, numRecords = 1, tp, startingTimestamp = startingTimestamp)
+
+ // another consumer in the same group should get the same position
+ val otherConsumer = createConsumer()
+ otherConsumer.assign(List(tp).asJava)
+ consumeAndVerifyRecords(consumer = otherConsumer, numRecords = 1, startingOffset = 5, startingTimestamp = startingTimestamp)
+ }
+
+ def changeConsumerSubscriptionAndValidateAssignment[K, V](consumer: Consumer[K, V],
+ topicsToSubscribe: List[String],
+ expectedAssignment: Set[TopicPartition],
+ rebalanceListener: ConsumerRebalanceListener): Unit = {
+ consumer.subscribe(topicsToSubscribe.asJava, rebalanceListener)
+ awaitAssignment(consumer, expectedAssignment)
+ }
+}
+
+object PlaintextConsumerCommitTest {
+
+ def getTestQuorumAndGroupProtocolParametersAll: Stream[Arguments] =
+ BaseConsumerTest.getTestQuorumAndGroupProtocolParametersAll()
+}
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
index 8784e5b43fc..77ba47fa08f 100644
--- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
+++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala
@@ -15,7 +15,7 @@ package kafka.api
import java.time.Duration
import java.util
import java.util.Arrays.asList
-import java.util.{Locale, Optional, Properties}
+import java.util.{Locale, Properties}
import kafka.server.{KafkaBroker, QuotaType}
import kafka.utils.{TestInfoUtils, TestUtils}
import org.apache.kafka.clients.admin.{NewPartitions, NewTopic}
@@ -135,58 +135,6 @@ class PlaintextConsumerTest extends BaseConsumerTest {
testHeadersSerializeDeserialize(extendedSerializer, extendedDeserializer)
}
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testAutoCommitOnClose(quorum: String, groupProtocol: String): Unit = {
- this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
- val consumer = createConsumer()
-
- val numRecords = 10000
- val producer = createProducer()
- sendRecords(producer, numRecords, tp)
-
- consumer.subscribe(List(topic).asJava)
- awaitAssignment(consumer, Set(tp, tp2))
-
- // should auto-commit sought positions before closing
- consumer.seek(tp, 300)
- consumer.seek(tp2, 500)
- consumer.close()
-
- // now we should see the committed positions from another consumer
- val anotherConsumer = createConsumer()
- assertEquals(300, anotherConsumer.committed(Set(tp).asJava).get(tp).offset)
- assertEquals(500, anotherConsumer.committed(Set(tp2).asJava).get(tp2).offset)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testAutoCommitOnCloseAfterWakeup(quorum: String, groupProtocol: String): Unit = {
- this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
- val consumer = createConsumer()
-
- val numRecords = 10000
- val producer = createProducer()
- sendRecords(producer, numRecords, tp)
-
- consumer.subscribe(List(topic).asJava)
- awaitAssignment(consumer, Set(tp, tp2))
-
- // should auto-commit sought positions before closing
- consumer.seek(tp, 300)
- consumer.seek(tp2, 500)
-
- // wakeup the consumer before closing to simulate trying to break a poll
- // loop from another thread
- consumer.wakeup()
- consumer.close()
-
- // now we should see the committed positions from another consumer
- val anotherConsumer = createConsumer()
- assertEquals(300, anotherConsumer.committed(Set(tp).asJava).get(tp).offset)
- assertEquals(500, anotherConsumer.committed(Set(tp2).asJava).get(tp2).offset)
- }
-
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testAutoOffsetReset(quorum: String, groupProtocol: String): Unit = {
@@ -211,48 +159,6 @@ class PlaintextConsumerTest extends BaseConsumerTest {
consumeAndVerifyRecords(consumer = consumer, numRecords = 1, startingOffset = 0, startingTimestamp = startingTimestamp)
}
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testCommitMetadata(quorum: String, groupProtocol: String): Unit = {
- val consumer = createConsumer()
- consumer.assign(List(tp).asJava)
-
- // sync commit
- val syncMetadata = new OffsetAndMetadata(5, Optional.of(15), "foo")
- consumer.commitSync(Map((tp, syncMetadata)).asJava)
- assertEquals(syncMetadata, consumer.committed(Set(tp).asJava).get(tp))
-
- // async commit
- val asyncMetadata = new OffsetAndMetadata(10, "bar")
- sendAndAwaitAsyncCommit(consumer, Some(Map(tp -> asyncMetadata)))
- assertEquals(asyncMetadata, consumer.committed(Set(tp).asJava).get(tp))
-
- // handle null metadata
- val nullMetadata = new OffsetAndMetadata(5, null)
- consumer.commitSync(Map(tp -> nullMetadata).asJava)
- assertEquals(nullMetadata, consumer.committed(Set(tp).asJava).get(tp))
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testAsyncCommit(quorum: String, groupProtocol: String): Unit = {
- val consumer = createConsumer()
- consumer.assign(List(tp).asJava)
-
- val callback = new CountConsumerCommitCallback
- val count = 5
-
- for (i <- 1 to count)
- consumer.commitAsync(Map(tp -> new OffsetAndMetadata(i)).asJava, callback)
-
- TestUtils.pollUntilTrue(consumer, () => callback.successCount >= count || callback.lastError.isDefined,
- "Failed to observe commit callback before timeout", waitTimeMs = 10000)
-
- assertEquals(None, callback.lastError)
- assertEquals(count, callback.successCount)
- assertEquals(new OffsetAndMetadata(count), consumer.committed(Set(tp).asJava).get(tp))
- }
-
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testPartitionsFor(quorum: String, groupProtocol: String): Unit = {
@@ -338,39 +244,6 @@ class PlaintextConsumerTest extends BaseConsumerTest {
producer.close()
}
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testPositionAndCommit(quorum: String, groupProtocol: String): Unit = {
- val producer = createProducer()
- var startingTimestamp = System.currentTimeMillis()
- sendRecords(producer, numRecords = 5, tp, startingTimestamp = startingTimestamp)
-
- val topicPartition = new TopicPartition(topic, 15)
- val consumer = createConsumer()
- assertNull(consumer.committed(Set(topicPartition).asJava).get(topicPartition))
-
- // position() on a partition that we aren't subscribed to throws an exception
- assertThrows(classOf[IllegalStateException], () => consumer.position(topicPartition))
-
- consumer.assign(List(tp).asJava)
-
- assertEquals(0L, consumer.position(tp), "position() on a partition that we are subscribed to should reset the offset")
- consumer.commitSync()
- assertEquals(0L, consumer.committed(Set(tp).asJava).get(tp).offset)
- consumeAndVerifyRecords(consumer = consumer, numRecords = 5, startingOffset = 0, startingTimestamp = startingTimestamp)
- assertEquals(5L, consumer.position(tp), "After consuming 5 records, position should be 5")
- consumer.commitSync()
- assertEquals(5L, consumer.committed(Set(tp).asJava).get(tp).offset, "Committed offset should be returned")
-
- startingTimestamp = System.currentTimeMillis()
- sendRecords(producer, numRecords = 1, tp, startingTimestamp = startingTimestamp)
-
- // another consumer in the same group should get the same position
- val otherConsumer = createConsumer()
- otherConsumer.assign(List(tp).asJava)
- consumeAndVerifyRecords(consumer = otherConsumer, numRecords = 1, startingOffset = 5, startingTimestamp = startingTimestamp)
- }
-
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testPartitionPauseAndResume(quorum: String, groupProtocol: String): Unit = {
@@ -450,65 +323,6 @@ class PlaintextConsumerTest extends BaseConsumerTest {
MockProducerInterceptor.resetCounters()
}
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testAutoCommitIntercept(quorum: String, groupProtocol: String): Unit = {
- val topic2 = "topic2"
- createTopic(topic2, 2, brokerCount)
-
- // produce records
- val numRecords = 100
- val testProducer = createProducer(keySerializer = new StringSerializer, valueSerializer = new StringSerializer)
- (0 until numRecords).map { i =>
- testProducer.send(new ProducerRecord(tp.topic(), tp.partition(), s"key $i", s"value $i"))
- }.foreach(_.get)
-
- // create consumer with interceptor
- this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
- this.consumerConfig.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.apache.kafka.test.MockConsumerInterceptor")
- val testConsumer = createConsumer(keyDeserializer = new StringDeserializer, valueDeserializer = new StringDeserializer)
- val rebalanceListener = new ConsumerRebalanceListener {
- override def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) = {
- // keep partitions paused in this test so that we can verify the commits based on specific seeks
- testConsumer.pause(partitions)
- }
-
- override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) = {}
- }
- changeConsumerSubscriptionAndValidateAssignment(testConsumer, List(topic), Set(tp, tp2), rebalanceListener)
- testConsumer.seek(tp, 10)
- testConsumer.seek(tp2, 20)
-
- // change subscription to trigger rebalance
- val commitCountBeforeRebalance = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue()
- changeConsumerSubscriptionAndValidateAssignment(testConsumer,
- List(topic, topic2),
- Set(tp, tp2, new TopicPartition(topic2, 0), new TopicPartition(topic2, 1)),
- rebalanceListener)
-
- // after rebalancing, we should have reset to the committed positions
- assertEquals(10, testConsumer.committed(Set(tp).asJava).get(tp).offset)
- assertEquals(20, testConsumer.committed(Set(tp2).asJava).get(tp2).offset)
-
- // In both CLASSIC and CONSUMER protocols, interceptors are executed in poll and close.
- // However, in the CONSUMER protocol, the assignment may be changed outside of a poll, so
- // we need to poll once to ensure the interceptor is called.
- if (groupProtocol.toUpperCase == GroupProtocol.CONSUMER.name) {
- testConsumer.poll(Duration.ZERO);
- }
-
- assertTrue(MockConsumerInterceptor.ON_COMMIT_COUNT.intValue() > commitCountBeforeRebalance)
-
- // verify commits are intercepted on close
- val commitCountBeforeClose = MockConsumerInterceptor.ON_COMMIT_COUNT.intValue()
- testConsumer.close()
- assertTrue(MockConsumerInterceptor.ON_COMMIT_COUNT.intValue() > commitCountBeforeClose)
- testProducer.close()
-
- // cleanup
- MockConsumerInterceptor.resetCounters()
- }
-
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testInterceptorsWithWrongKeyValue(quorum: String, groupProtocol: String): Unit = {
@@ -626,74 +440,6 @@ class PlaintextConsumerTest extends BaseConsumerTest {
consumeAndVerifyRecords(consumer = consumer, numRecords = 0, startingOffset = 5, startingTimestamp = startingTimestamp)
}
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testCommitSpecifiedOffsets(quorum: String, groupProtocol: String): Unit = {
- val producer = createProducer()
- sendRecords(producer, numRecords = 5, tp)
- sendRecords(producer, numRecords = 7, tp2)
-
- val consumer = createConsumer()
- consumer.assign(List(tp, tp2).asJava)
-
- val pos1 = consumer.position(tp)
- val pos2 = consumer.position(tp2)
- consumer.commitSync(Map[TopicPartition, OffsetAndMetadata]((tp, new OffsetAndMetadata(3L))).asJava)
- assertEquals(3, consumer.committed(Set(tp).asJava).get(tp).offset)
- assertNull(consumer.committed(Set(tp2).asJava).get(tp2))
-
- // Positions should not change
- assertEquals(pos1, consumer.position(tp))
- assertEquals(pos2, consumer.position(tp2))
- consumer.commitSync(Map[TopicPartition, OffsetAndMetadata]((tp2, new OffsetAndMetadata(5L))).asJava)
- assertEquals(3, consumer.committed(Set(tp).asJava).get(tp).offset)
- assertEquals(5, consumer.committed(Set(tp2).asJava).get(tp2).offset)
-
- // Using async should pick up the committed changes after commit completes
- sendAndAwaitAsyncCommit(consumer, Some(Map(tp2 -> new OffsetAndMetadata(7L))))
- assertEquals(7, consumer.committed(Set(tp2).asJava).get(tp2).offset)
- }
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testAutoCommitOnRebalance(quorum: String, groupProtocol: String): Unit = {
- val topic2 = "topic2"
- createTopic(topic2, 2, brokerCount)
-
- this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
- val consumer = createConsumer()
-
- val numRecords = 10000
- val producer = createProducer()
- sendRecords(producer, numRecords, tp)
-
- val rebalanceListener = new ConsumerRebalanceListener {
- override def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) = {
- // keep partitions paused in this test so that we can verify the commits based on specific seeks
- consumer.pause(partitions)
- }
-
- override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) = {}
- }
-
- consumer.subscribe(List(topic).asJava, rebalanceListener)
-
- awaitAssignment(consumer, Set(tp, tp2))
-
- consumer.seek(tp, 300)
- consumer.seek(tp2, 500)
-
- // change subscription to trigger rebalance
- consumer.subscribe(List(topic, topic2).asJava, rebalanceListener)
-
- val newAssignment = Set(tp, tp2, new TopicPartition(topic2, 0), new TopicPartition(topic2, 1))
- awaitAssignment(consumer, newAssignment)
-
- // after rebalancing, we should have reset to the committed positions
- assertEquals(300, consumer.committed(Set(tp).asJava).get(tp).offset)
- assertEquals(500, consumer.committed(Set(tp2).asJava).get(tp2).offset)
- }
-
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testPerPartitionLeadMetricsCleanUpWithSubscribe(quorum: String, groupProtocol: String): Unit = {
@@ -893,14 +639,6 @@ class PlaintextConsumerTest extends BaseConsumerTest {
brokers.foreach(assertNoMetric(_, "throttle-time", QuotaType.Request, consumerClientId))
}
- def changeConsumerSubscriptionAndValidateAssignment[K, V](consumer: Consumer[K, V],
- topicsToSubscribe: List[String],
- expectedAssignment: Set[TopicPartition],
- rebalanceListener: ConsumerRebalanceListener): Unit = {
- consumer.subscribe(topicsToSubscribe.asJava, rebalanceListener)
- awaitAssignment(consumer, expectedAssignment)
- }
-
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testConsumingWithNullGroupId(quorum: String, groupProtocol: String): Unit = {
@@ -1067,19 +805,4 @@ class PlaintextConsumerTest extends BaseConsumerTest {
consumer2.close()
}
-
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
- @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
- def testSubscribeAndCommitSync(quorum: String, groupProtocol: String): Unit = {
- // This test ensure that the member ID is propagated from the group coordinator when the
- // assignment is received into a subsequent offset commit
- val consumer = createConsumer()
- assertEquals(0, consumer.assignment.size)
- consumer.subscribe(List(topic).asJava)
- awaitAssignment(consumer, Set(tp, tp2))
-
- consumer.seek(tp, 0)
-
- consumer.commitSync()
- }
}