chenyulin0719 commented on code in PR #18513:
URL: https://github.com/apache/kafka/pull/18513#discussion_r1928840593
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4004,6 +3922,82 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
ConfigSource.DYNAMIC_TOPIC_CONFIG, false, false,
Collections.emptyList(), null, null),
topicConfigs.get(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG))
}
+
+ class BackgroundConsumerSet(testGroupId: String, testClientId: String,
defaultConsumerConfig: Properties) {
+ private val consumerSet:
scala.collection.mutable.Set[Consumer[Array[Byte], Array[Byte]]] =
scala.collection.mutable.Set.empty
+ private val consumerThreads: scala.collection.mutable.Set[Thread] =
scala.collection.mutable.Set.empty
+ private var startLatch: CountDownLatch = new CountDownLatch(0)
+ private var stopLatch: CountDownLatch = new CountDownLatch(0)
+ private var consumerThreadRunning = new AtomicBoolean(false)
+
+ defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
testGroupId)
+ defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG,
testClientId)
+
+ def addConsumer(groupInstanceId: String, topic: String, configOverrides:
Properties = new Properties()): Unit = {
+ val newConsumerConfig =
defaultConsumerConfig.clone().asInstanceOf[Properties]
+ if (groupInstanceId != "") {
+ // static member
+ newConsumerConfig.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG,
groupInstanceId)
+ }
+ newConsumerConfig.putAll(configOverrides)
+
+ val consumer = createConsumer(configOverrides = newConsumerConfig)
+ val consumerThread = createConsumerThread(consumer, topic)
+ consumerSet.add(consumer)
Review Comment:
The consumerSet is a safeguard to ensure that all the consumers are closed
in `BackgroundConsumerSet#close()`.
I think it still useful if the thread is interrupted before
`consumer.close()` called in thread.
@dajac Does it make sense?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
