smjn commented on code in PR #18521:
URL: https://github.com/apache/kafka/pull/18521#discussion_r1914781299
##########
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:
@AndrewJSchofield It was a blunder on my part. The SGS topic is set to have
3 partitions in the cluster config and I was only assigning to 1 partition.
Will rectify.
--
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]
