KAFKA-18513: Validate share state topic records produced in tests. by smjn · Pull Request #18521 · apache/kafka

smjn · 2025-01-14T10:08:24Z

Currently, the tests using DefaultStatePersister do not perform any validations on the __share_group_state topic to verify
the records.
In this PR, we have added a small util method to do the same for the relevant tests.

AndrewJSchofield

Thanks for the PR. Just a few minor comments.

AndrewJSchofield · 2025-01-14T10:16:16Z

+                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());


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 too is causing flake

I think that implies that the test would also be flaky with subscribe instead of assign, just that it runs slightly more slowly so the risk of flakiness is lower.

Perhaps but since the pattern is same as existing tests. This addition should not decrease reliability.

@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.

AndrewJSchofield · 2025-01-14T10:16:31Z

+                        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


nit: extra space before ==

AndrewJSchofield · 2025-01-14T10:17:02Z

+            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");


Then you could also use seekToBeginning()

this is causing flake - I will keep the config

AndrewJSchofield

It seems to me that testMultipleConsumersInGroupSequentialConsumption is flaky with this PR.

smjn · 2025-01-14T17:00:29Z

It seems to me that testMultipleConsumersInGroupSequentialConsumption is flaky with this PR.

It appears that occasionally this error is happening

[2025-01-14 22:23:21,456] ERROR [group-coordinator-event-processor-0]: Failed to run event CoordinatorWriteEvent(name=share-group-heartbeat) due to: Histogram recorded value cannot be negative.. (org.apache.kafka.coordinator.common.runtime.MultiThreadedEventProcessor$EventProcessorThread:150)
java.lang.ArrayIndexOutOfBoundsException: Histogram recorded value cannot be negative.
	at org.HdrHistogram.AbstractHistogram.countsArrayIndex(AbstractHistogram.java:2399) ~[HdrHistogram-2.2.2.jar:?]

It would seem this is resulting in off by one errors - sometimes RPCs are 4 and sometimes 5. I will remove this check from this test for now

AndrewJSchofield · 2025-01-14T22:32:31Z

I also had a failure locally in testConsumerCloseInGroupSequential because the number of records was incorrect. I expect it's unpredictable for the more complex tests.

smjn · 2025-01-15T04:55:37Z

I also had a failure locally in testConsumerCloseInGroupSequential because the number of records was incorrect. I expect it's unpredictable for the more complex tests.

Then maybe exact record count match is too strict. Perhaps checking for record being produced should be the way to go. Another reason for this is the RPC coalescing and batching which might make the number of records actually written different each time and as a result make the tests brittle.
@AndrewJSchofield

AndrewJSchofield

Thanks for the changes. I ran this test repeatedly yesterday and I think it's time to remove the no-op persister from this test now. The new changes add about 6 minutes to the execution time. The no-op persister is not adding any value any longer and it takes about 6 minutes too. Please remove the parameterisation of the persister and just use the default persister in the tests.

AndrewJSchofield · 2025-01-15T09:05:27Z

    }

-    private void maybeVerifyShareGroupStateTopicRecordCount(String persister, int messageCount) {
+    private void maybeVerifyShareGroupStateTopicRecordCount(String persister) {


The method name now seems redundant. I suggest verifyShareGroupStateTopicRecordsProduced.

…params

AndrewJSchofield

Just a couple of additional comments. Thanks for the update.

AndrewJSchofield · 2025-01-15T09:29:56Z

            .setConfigProp("group.share.enable", "true")
            .setConfigProp("group.share.partition.max.record.locks", "10000")
-            .setConfigProp("group.share.persister.class.name", persisterClassName)
+            .setConfigProp("group.share.persister.class.name", DEFAULT_STATE_PERSISTER)


This is the default. Not required.

AndrewJSchofield · 2025-01-15T09:30:58Z

+                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());


I think you can remove the group.id config too because of the use of KafkaConsumer.assign.

AndrewJSchofield

One final comment. When the build next finishes, please can you triage the failed tests and make sure there are open issues for failed tests. ClientUtilsTest#testParseAndValidateAddressesWithReverseLookup is expected to fail until #18549 is merged. Thanks.

AndrewJSchofield · 2025-01-15T13:38:18Z


+    private void verifyShareGroupStateTopicRecordsProduced() {
+        try {
+            TestUtils.waitForCondition(() ->


This condition will already be satisfied because it mirrors the warmup() method which runs before each test. I think you can remove this as redundant code.

smjn · 2025-01-15T13:47:34Z

One final comment. When the build next finishes, please can you triage the failed tests and make sure there are open issues for failed tests. ClientUtilsTest#testParseAndValidateAddressesWithReverseLookup is expected to fail until #18549 is merged. Thanks.

They seem to be completely unrelated to our work.

AndrewJSchofield · 2025-01-15T13:53:38Z

One final comment. When the build next finishes, please can you triage the failed tests and make sure there are open issues for failed tests. ClientUtilsTest#testParseAndValidateAddressesWithReverseLookup is expected to fail until #18549 is merged. Thanks.

They seem to be completely unrelated to our work.

True, but if some of the failed tests are flaky but not marked as flaky, then we ought to do something about it.

smjn · 2025-01-15T15:00:23Z

One final comment. When the build next finishes, please can you triage the failed tests and make sure there are open issues for failed tests. ClientUtilsTest#testParseAndValidateAddressesWithReverseLookup is expected to fail until #18549 is merged. Thanks.

They seem to be completely unrelated to our work.

True, but if some of the failed tests are flaky but not marked as flaky, then we ought to do something about it.

Did multiple runs for each of the tests:

NO ERRORS

UserQuotaTest > testThrottledProducerConsumer(String, String).quorum=kraft.groupProtocol=classic
MetricsDuringTopicCreationDeletionTest > "testMetricsDuringTopicCreateDelete(String).quorum=kraft"
PlaintextAdminIntegrationTest > "testElectPreferredLeaders(String).quorum=kraft"
AbstractCoordinatorTest > testWakeupAfterSyncGroupReceivedExternalCompletion() [ALREADY MARKED FLAKY]

FLAKE FOUND

https://issues.apache.org/jira/browse/KAFKA-18550

 KafkaAdminClientTest > testAdminClientApisAuthenticationFailure()

org.opentest4j.AssertionFailedError: Expected an authentication error, but got java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: createPartitions
	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:155)
	at org.apache.kafka.clients.admin.KafkaAdminClientTest.lambda$callAdminClientApisAndExpectAnAuthenticationError$38(KafkaAdminClientTest.java:1808)
	at org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:53)
	at org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:35)
	at org.junit.jupiter.api.Assertions.assertThrows(Assertions.java:3115)
	at org.apache.kafka.clients.admin.KafkaAdminClientTest.callAdminClientApisAndExpectAnAuthenticationError(KafkaAdminClientTest.java:1808)
	at org.apache.kafka.clients.admin.KafkaAdminClientTest.testAdminClientApisAuthenticationFailure(KafkaAdminClientTest.java:1793)

https://issues.apache.org/jira/browse/KAFKA-18551

EagerConsumerCoordinatorTest  > testOutdatedCoordinatorAssignment()

org.opentest4j.AssertionFailedError: 
Expected :1
Actual   :2
<Click to see difference>


	at org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
	at org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
	at org.junit.jupiter.api.AssertEquals.failNotEqual(AssertEquals.java:197)
	at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:150)
	at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:145)
	at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:531)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.testOutdatedCoordinatorAssignment(ConsumerCoordinatorTest.java:1066)
	at org.apache.kafka.clients.consumer.internals.EagerConsumerCoordinatorTest.some(EagerConsumerCoordinatorTest.java:29)
	at java.base/java.lang.reflect.Method.invoke(Method.java:568)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)

@AndrewJSchofield

AndrewJSchofield

lgtm. Will merge once the build is complete.

…pache#18521) Reviewers: Andrew Schofield <aschofield@confluent.io>

KAFKA-18513: Validate share state topic record count in tests.

bcfc9fc

github-actions Bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) small Small PRs labels Jan 14, 2025

AndrewJSchofield added KIP-932 Queues for Kafka ci-approved and removed triage PRs from the community labels Jan 14, 2025

AndrewJSchofield requested changes Jan 14, 2025

View reviewed changes

incorporated comment, updated all tests.

d1d88c5

smjn requested a review from AndrewJSchofield January 14, 2025 11:39

smjn added 2 commits January 14, 2025 18:19

Merge remote-tracking branch 'apache-kafka/trunk' into KAFKA-18513

bf5f99c

use assign and seektobeginning over subscribe and offset config

76d8672

AndrewJSchofield requested changes Jan 14, 2025

View reviewed changes

fix verify condition for multi cons, multi group concurrent consumption.

729793b

smjn requested a review from AndrewJSchofield January 14, 2025 16:48

remove SGS verification from multi cons sequential test.

af187a3

relax condition for exact record match.

6413ada

AndrewJSchofield requested changes Jan 15, 2025

View reviewed changes

remove no_op persister, change sgs record verifier name, remove test …

cc31864

…params

github-actions Bot removed the small Small PRs label Jan 15, 2025

smjn requested a review from AndrewJSchofield January 15, 2025 09:22

smjn changed the title ~~KAFKA-18513: Validate share state topic record count in tests.~~ KAFKA-18513: Validate share state topic records produced in tests. Jan 15, 2025

AndrewJSchofield requested changes Jan 15, 2025

View reviewed changes

inc review comments

4d7b6fc

smjn requested a review from AndrewJSchofield January 15, 2025 09:40

AndrewJSchofield requested changes Jan 15, 2025

View reviewed changes

remove cache wait code in SGS topic record verification

cd5443e

smjn requested a review from AndrewJSchofield January 15, 2025 15:04

AndrewJSchofield approved these changes Jan 15, 2025

View reviewed changes

AndrewJSchofield merged commit e3a56f3 into apache:trunk Jan 15, 2025

pranavt84 pushed a commit to pranavt84/kafka that referenced this pull request Jan 27, 2025

KAFKA-18513: Validate share state topic records produced in tests. (a…

c62b9c3

…pache#18521) Reviewers: Andrew Schofield <aschofield@confluent.io>

airlock-confluentinc Bot pushed a commit to confluentinc/kafka that referenced this pull request Jan 27, 2025

KAFKA-18513: Validate share state topic records produced in tests. (a…

fb96dff

…pache#18521) Reviewers: Andrew Schofield <aschofield@confluent.io>

manoj-mathivanan pushed a commit to manoj-mathivanan/kafka that referenced this pull request Feb 19, 2025

KAFKA-18513: Validate share state topic records produced in tests. (a…

446440d

…pache#18521) Reviewers: Andrew Schofield <aschofield@confluent.io>

Conversation

smjn commented Jan 14, 2025

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

smjn Jan 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

smjn commented Jan 14, 2025

Uh oh!

AndrewJSchofield commented Jan 14, 2025

Uh oh!

smjn commented Jan 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

smjn commented Jan 15, 2025

Uh oh!

AndrewJSchofield commented Jan 15, 2025

Uh oh!

smjn commented Jan 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

NO ERRORS

FLAKE FOUND

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

smjn Jan 14, 2025 •

edited

Loading

smjn commented Jan 15, 2025 •

edited

Loading

smjn commented Jan 15, 2025 •

edited

Loading