kgo-verifier: add an option to commit offsets right after fetch by bashtanov · Pull Request #27480 · redpanda-data/redpanda

bashtanov · 2025-09-08T16:43:05Z

For kgo-verifier status thread to minimize the number of uncommitted messages reported as valid reads to the status thread.
Based on redpanda-data/kgo-verifier#54.

Backports Required

Release Notes

none

dotnwat · 2025-09-08T17:15:22Z

+		if grw.config.commitAfterFetch {
+			if err := client.CommitUncommittedOffsets(ctx); err != nil {
+				return err
+			}
+		}


should this come after the EachRecord loop below? It seems like the docs say something like

First PollFetches, then process every record, then call CommitUncommittedOffsets.

Processing a record involves validating, which, in turn, bumps the counters. This way we may end up in a situation where counters have been bumped but the offsets cannot be committed, as the consumer group on the cluster may not be available for writing into it any more. It's specifically the case I'm testing, as the cluster is migrating away the consumer group while we are using it.

maybe we should handle the error here. It is always possible that we won't be able to commit. In this case we should either:

give up and fetch the data once again, then trying to commit them (at least once semantics)

move on with what we have processed and commit with the next batch, risking reprocessing more data if group rebalance happens

give up and fetch the data once again, then trying to commit them (at least once semantics)

isn't it what happens when the error is handled in the caller (GroupReadWorker::Wait)?

@bashtanov the way I read @dotnwat comment is this: "are you sure franz-go internally bumped the "consumed up to" offsets after fetch but before EachRecord loop executed?" Otherwise you are just committing offsets from the previous consume round rather than the current one and you haven't solved the problem the PR claims to.

From my glancing at the code looks like the answer is yes.

But now I have a different question. What happens if commit succeeds but we fail to get the response and error out?

On that error, we'll bubble up the error and the caller will retry.

restarting consumer group reader for error

Now the problem with this is that the next round will restart from "higher up" than was previously recorded. This way the number of valid reads reported by the status endpoint might end up being lower than number of records in topic/partition.

Grep the repo for "valid_reads >=". We allow in a couple of places that to happen. With the current PR logic this will be broken.

I would move this to below any logic that is important to be done before for at-least-once logic.

If a crash happens after the commit goes through right after this line but before the EachRecord below, then ValidateRecord will not be called.

thanks Travis. cc @nvartolomei @bashtanov

nvartolomei · 2025-09-09T11:13:08Z

+		if grw.config.commitAfterFetch {
+			if err := client.CommitUncommittedOffsets(ctx); err != nil {
+				return err
+			}
+		}


From my glancing at the code looks like the answer is yes.

But now I have a different question. What happens if commit succeeds but we fail to get the response and error out?

On that error, we'll bubble up the error and the caller will retry.

restarting consumer group reader for error

Now the problem with this is that the next round will restart from "higher up" than was previously recorded. This way the number of valid reads reported by the status endpoint might end up being lower than number of records in topic/partition.

Grep the repo for "valid_reads >=". We allow in a couple of places that to happen. With the current PR logic this will be broken.

nvartolomei · 2025-09-09T11:13:30Z

Didn't mean to approve just yet.

bashtanov · 2025-09-09T15:49:00Z

Thanks everyone for explaining me why proposed logic was incorrect. Please could you have a look at the new version?

bashtanov · 2025-09-11T13:00:58Z

/dt

vbotbuildovich · 2025-09-11T15:13:56Z

CI test results

test results on build#72075

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
AvailabilityTests	test_availability_when_one_node_failed	null	integration	https://buildkite.com/redpanda/redpanda/builds/72075#019938f7-5743-42fc-b46b-b5a16fde9ab9	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS

test results on build#72179

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
AuditLogTestOauth	test_kafka_oauth	{"authz_match": "acl"}	integration	https://buildkite.com/redpanda/redpanda/builds/72179#01993eeb-0c70-473b-b84c-b67ae69a59b0	FLAKY	10/21	upstream reliability is '93.01587301587301'. current run reliability is '47.61904761904761'. drift is 45.39683 and the allowed drift is set to 50. The test should PASS
DatalakeClusterRestoreTest	test_slow_tiered_storage_dupe_records	{"catalog_type": "nessie", "cloud_storage_type": 1}	integration	https://buildkite.com/redpanda/redpanda/builds/72179#01993eeb-0c66-4142-af9e-fd9ed2c089df	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS
SchemaEvolutionE2ETests	test_legal_schema_evolution	{"catalog_type": "rest_hadoop", "cloud_storage_type": 1, "produce_mode": "avro", "query_engine": "trino", "test_case": "drop_column"}	integration	https://buildkite.com/redpanda/redpanda/builds/72179#01993eeb-0c78-414a-a408-96e3908ad7b0	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS

test results on build#72244

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
PartitionBalancerTest	test_recovery_mode_rebalance_finish	null	integration	https://buildkite.com/redpanda/redpanda/builds/72244#01994cbf-40a2-4a5b-84d7-05c20e285a02	FLAKY	15/21	upstream reliability is '92.5233644859813'. current run reliability is '71.42857142857143'. drift is 21.09479 and the allowed drift is set to 50. The test should PASS
TieredStorageTest	test_tiered_storage	{"cloud_storage_type_and_url_style": [2, "virtual_host"], "test_case": {"name": "(TS_Read == True, TS_Timequery == True, SpilloverManifestUploaded == True)"}}	integration	https://buildkite.com/redpanda/redpanda/builds/72244#01994cbf-409a-499b-93c0-d42f4ac886bf	FAIL	0/1

test results on build#72257

test_class	test_method	test_arguments	test_kind	job_url	test_status	passed	reason
CloudStorageInventoryTest	test_load_inventory_report	null	integration	https://buildkite.com/redpanda/redpanda/builds/72257#01994d84-a5c5-444f-b836-62fd9918dd81	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS
DataMigrationsApiTest	test_migrated_topic_data_integrity	{"include_groups": false, "params": {"cancellation": {"dir": "in", "stage": "executing"}, "use_alias": false}, "transfer_leadership": true}	integration	https://buildkite.com/redpanda/redpanda/builds/72257#01994d8b-68f5-43f3-a417-e6600b03991b	FLAKY	20/21	upstream reliability is '100.0'. current run reliability is '95.23809523809523'. drift is 4.7619 and the allowed drift is set to 50. The test should PASS

bashtanov · 2025-09-12T16:47:17Z

I've changed it according to @nvartolomei's suggestion (commit once every N messages, N configurable), as only having N=1 option may kill performance.

nvartolomei · 2025-09-13T07:25:58Z

@@ -0,0 +1,94 @@
+@rem


what's this?

some IDE plugin added it? I'll remove, thanks

nvartolomei · 2025-09-13T07:38:55Z

 	batchMaxBytes       = flag.Int("batch_max_bytes", 1048576, "the maximum batch size to allow per-partition (must be less than Kafka's max.message.bytes, producing)")
 	cgReaders           = flag.Int("consumer_group_readers", 0, "Number of parallel readers in the consumer group")
 	cgName              = flag.String("consumer_group_name", "", "The name of the consumer group. Generated randomly if not set.")
+	cgCommitFrequency   = flag.Int("commit-frequency", 0, "if non-zero, commit offsets after fetching this many records rather than relying on auto-commit")


nit: cgMaxUncommitted and consumer_group_max_uncommitted

-1 — auto

0 — at most once at we are not really interested in it now so be lazy: throw/do not implement it

>0 — what you want

commit-frequency is too vague imho and makes me think about rates rather than "lag" which it actually implements

also the value of 0 if you don't have the implementation context and you don't read the doc deep down here (i.e. someone else on our team is reading test code) could think it means don't commit at all

maxUncommitted seems more descriptive

Also, add a note that this is per topic rather than per partition.

nvartolomei · 2025-09-15T09:58:33Z

            cmd += " --tolerate-data-loss"
        if self._group_name is not None:
            cmd += f" --consumer_group_name {self._group_name}"
+        if self._max_uncommitted != 0:


this should have been if self._max_uncommitted != -1? I.e. pass the flag if default has been altered. This way if someone passes 0 it will crash as not supported. Current behavior is to continue with auto-commit which is confusing imho.

Maybe even make it None by default (not customized) and pass the flag only if not None. -1 as default makes sense in go flag parsing as it simplifies the flag parsing slightly.

nvartolomei · 2025-09-15T09:59:16Z


 		fetches.EachRecord(func(r *kgo.Record) {
+			// Will cancel the context if we have read everything
+			cgOffsets.AddRecord(ctx, r)


Is this reordering of code relevant for the change?

nvartolomei · 2025-09-15T10:01:16Z

-		fetches := client.PollFetches(ctx)
+		fetchBatchSize := 0
+		if grw.config.maxUncommitted > 0 {
+			fetchBatchSize = grw.config.maxUncommitted


you can pass it as-is to PollRecords. It already implements the semantics you need for negative values.

The only problem is value of 0 which it also interprets as -1 so this is where I would add the util.Die check. This is where validation is important and not in the main.

nvartolomei · 2025-09-15T10:02:48Z

 	batchMaxBytes       = flag.Int("batch_max_bytes", 1048576, "the maximum batch size to allow per-partition (must be less than Kafka's max.message.bytes, producing)")
 	cgReaders           = flag.Int("consumer_group_readers", 0, "Number of parallel readers in the consumer group")
 	cgName              = flag.String("consumer_group_name", "", "The name of the consumer group. Generated randomly if not set.")
+	cgMaxUncommitted    = flag.Int("max-uncommitted", -1, "Negative means rely on auto-commit. For positive, commit offsets after fetching this many records. Restricts the number of records shown as reads in statistics but not yet committed.")


nit: Mentioning "consumer group" at least in help message should be helpful for the reader to understand what "commit" refers to.

Commit messages after each N records to control how many messages may be uncommitted but processed in each partition.

bashtanov requested review from bharathv, joe-redpanda, mmaslankaprv and nvartolomei September 8, 2025 16:43

dotnwat reviewed Sep 8, 2025

View reviewed changes

nvartolomei previously approved these changes Sep 9, 2025

View reviewed changes

nvartolomei self-requested a review September 9, 2025 11:13

bashtanov dismissed nvartolomei’s stale review via bb72535 September 9, 2025 15:47

bashtanov force-pushed the kgo-verifier-commit-after-fetch branch from 6b1cbcd to bb72535 Compare September 9, 2025 15:47

bashtanov requested review from dotnwat and twmb September 9, 2025 15:49

bashtanov force-pushed the kgo-verifier-commit-after-fetch branch from bb72535 to e9cb50f Compare September 12, 2025 16:45

nvartolomei reviewed Sep 13, 2025

View reviewed changes

bashtanov force-pushed the kgo-verifier-commit-after-fetch branch 2 times, most recently from e06b7ec to f45cad3 Compare September 15, 2025 09:21

nvartolomei reviewed Sep 15, 2025

View reviewed changes

kgo-verifier: add an option to commit offsets regularly

e07ba08

Commit messages after each N records to control how many messages may be uncommitted but processed in each partition.

bashtanov force-pushed the kgo-verifier-commit-after-fetch branch from f45cad3 to e07ba08 Compare September 15, 2025 12:54

bashtanov requested a review from nvartolomei September 15, 2025 12:55

nvartolomei approved these changes Sep 15, 2025

View reviewed changes

bashtanov enabled auto-merge September 15, 2025 14:29

bashtanov merged commit a1a7842 into redpanda-data:dev Sep 15, 2025
16 checks passed

Uh oh!

Conversation

bashtanov commented Sep 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Backports Required

Release Notes

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!

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!

Choose a reason for hiding this comment

Uh oh!

nvartolomei commented Sep 9, 2025

Uh oh!

bashtanov commented Sep 9, 2025

Uh oh!

bashtanov commented Sep 11, 2025

Uh oh!

vbotbuildovich commented Sep 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI test results

Uh oh!

bashtanov commented Sep 12, 2025

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!

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!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

bashtanov commented Sep 8, 2025 •

edited

Loading

vbotbuildovich commented Sep 11, 2025 •

edited

Loading