AndrewJSchofield commented on code in PR #18521:
URL: https://github.com/apache/kafka/pull/18521#discussion_r1914571923
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -1922,6 +1930,40 @@ private void warmup() throws InterruptedException {
}
}
+ private void maybeVerifyShareGroupStateTopicRecordCount(String persister,
int messageCount) {
+ if (!persister.equals(DEFAULT_STATE_PERSISTER)) {
+ return;
+ }
+ try {
+ TestUtils.waitForCondition(() ->
+
!cluster.brokers().get(0).metadataCache().getAliveBrokerNodes(new
ListenerName("EXTERNAL")).isEmpty(),
+ DEFAULT_MAX_WAIT_MS, 100L, () -> "cache not up yet");
+ Map<String, Object> consumerConfigs = new HashMap<>();
+ consumerConfigs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers());
+ consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG,
UUID.randomUUID().toString());
+ consumerConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
consumerConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ consumerConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest");
+ try (KafkaConsumer<byte[], byte[]> consumer = new
KafkaConsumer<>(consumerConfigs)) {
+
consumer.subscribe(Collections.singleton(Topic.SHARE_GROUP_STATE_TOPIC_NAME));
+ Set<ConsumerRecord<byte[], byte[]>> records = new HashSet<>();
+ TestUtils.waitForCondition(() -> {
+ ConsumerRecords<byte[], byte[]> msgs =
consumer.poll(Duration.ofMillis(5000L));
+ if (msgs.count() > 0) {
+
msgs.records(Topic.SHARE_GROUP_STATE_TOPIC_NAME).forEach(records::add);
+ }
+ return records.size() == messageCount + 2; // +2
because of extra warmup records
Review Comment:
nit: extra space before `==`
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -1922,6 +1930,40 @@ private void warmup() throws InterruptedException {
}
}
+ private void maybeVerifyShareGroupStateTopicRecordCount(String persister,
int messageCount) {
+ if (!persister.equals(DEFAULT_STATE_PERSISTER)) {
+ return;
+ }
+ try {
+ TestUtils.waitForCondition(() ->
+
!cluster.brokers().get(0).metadataCache().getAliveBrokerNodes(new
ListenerName("EXTERNAL")).isEmpty(),
+ DEFAULT_MAX_WAIT_MS, 100L, () -> "cache not up yet");
+ Map<String, Object> consumerConfigs = new HashMap<>();
+ consumerConfigs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers());
+ consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG,
UUID.randomUUID().toString());
+ consumerConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+
consumerConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
+ consumerConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest");
Review Comment:
Then you could also use `seekToBeginning()`
##########
core/src/test/java/kafka/test/api/ShareConsumerTest.java:
##########
@@ -1922,6 +1930,40 @@ private void warmup() throws InterruptedException {
}
}
+ private void maybeVerifyShareGroupStateTopicRecordCount(String persister,
int messageCount) {
+ if (!persister.equals(DEFAULT_STATE_PERSISTER)) {
+ return;
+ }
+ try {
+ TestUtils.waitForCondition(() ->
+
!cluster.brokers().get(0).metadataCache().getAliveBrokerNodes(new
ListenerName("EXTERNAL")).isEmpty(),
+ DEFAULT_MAX_WAIT_MS, 100L, () -> "cache not up yet");
+ Map<String, Object> consumerConfigs = new HashMap<>();
+ consumerConfigs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers());
+ consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG,
UUID.randomUUID().toString());
Review Comment:
I think you could use `KafkaConsumer.assign()` instead so that the test
doesn't have to wait for the consumer to join the consumer group.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
