Skip to content

fix(engine): each INTEGRITY run reads in a group of its own, and a consumer with no committed offset starts at its task's first record, not the topic's oldest - #376

Merged
bmscomp merged 1 commit into
mainfrom
fix/audit-07-integrity-groups
Oct 17, 2026
Merged

fix(engine): each INTEGRITY run reads in a group of its own, and a consumer with no committed offset starts at its task's first record, not the topic's oldest#376
bmscomp merged 1 commit into
mainfrom
fix/audit-07-integrity-groups

Conversation

@bmscomp

@bmscomp bmscomp commented Oct 17, 2026

Copy link
Copy Markdown
Owner

Why

Two defaults made a native consumer read records that weren't its run's. Checked in the code on main (02e0df2):

  • INTEGRITY: one group for every run. TestOrchestrator.buildTasks set an INTEGRITY task's group to "integrity-cg" when the request named none (TestOrchestrator.java:1718), and NativeKafkaBackend.runIntegrityConsumer joins it with -integrity appended (:786). So every default INTEGRITY run read in integrity-cg-integrity. Run A produces for 10 minutes; two minutes in, run B, one minute long, starts on the same default topic. B's consumer reads past A's records, skipping them as another run's, and auto-commits up to the head. When A's consumer starts after its produce phase, it resumes at B's committed offset and never sees A's first minutes. A reports those acked records lost: DATA_LOSS, about 40% lost, while the cluster lost nothing. When the two consume phases overlap, the group splits the partitions between the runs, with the same result.
  • LOAD and ENDURANCE: a new group starts at the oldest record. consumeTask already gives each consumer a group of its own (taskId + "-group", :1933), but both native consumers set auto.offset.reset to earliest (:738, :789), so a new group starts at the oldest record still on the topic. On a reused topic, the Nth default LOAD run reads every earlier run's records before its own, and its consume throughput is its own records over that whole time. It is understated by up to the ratio of all the records it read to its own. That ratio is reached only when the consumer's read rate is what limits it; a consumer that reads several times faster than the producer writes loses less. ENDURANCE has the same consumer (runConsumer), and the run's avgThroughput averages the produce and consume rows (MetricUtils.computeSummary), so the run's average is understated too.

A per-run INTEGRITY group alone would not fix it: a new group with earliest would still read the whole topic. The start-offset change alone would not either: it seeks only partitions with no committed offset, so a shared group would still resume at another run's offset. This PR does both.

Changes

TestOrchestrator

  • The INTEGRITY default is the task id plus -group, as consumeTask's: runId + "-integrity-0-group". The backend still appends -integrity, so a default run's consumer reads in <run id>-integrity-0-group-integrity.
  • The block stays at 4 lines (a 3-line comment and one 120-column line, using spec.hasConsumerGroup(), the same null check), so no CLI ref past it moves.

NativeKafkaBackend

  • runConsumer and runIntegrityConsumer subscribe with a rebalance listener, sinceSubmit(consumer, state.submittedAtMs). On each assigned partition the group has no committed offset for, it:
    • reads the end offset first;
    • then looks up the first record written since the task was submitted, less a second (offsetsForTimes);
    • seeks there, or to the end offset it read first when the lookup finds nothing.
  • Reading the end before the lookup is what keeps a LOAD run from skipping records. The LOAD producer is a separate task writing at the same time, and seekToEnd is applied only at the next poll, so records written between the lookup and that poll would be skipped. Any record written after the end was read is at or past it.
  • The time is WorkerState.submittedAtMs, taken when the orchestrator submits the task, not startTimeMs, which the worker's own virtual thread sets. On a loaded pod with one CPU that thread can start more than a second after the producer's first send. The orchestrator submits the producer just before the consumer, so the consumer's submit time less a second comes before the producer's first record. For INTEGRITY the producer and consumer are one task.
  • A partition with a committed offset resumes from it, and earliest stays to cover a committed offset that is out of range.
  • kafka-clients 3.9.2 has no by_duration offset reset, hence the listener.
  • 6 import lines, so every ref into the file below them moves by +6.

cli/examples/perf-integrity.yaml

  • No longer names consumerGroup: integrity-cg. Any named group shared by every run of the example has the same problem: overlapping runs split its partitions, and a run resumes wherever another run committed.

kates mcp

  • The merged-spec caveat (mcp_caveats.go) gives the new INTEGRITY default, and keeps integrity-cg-integrity for runs created before this change, which every such run shared. Its closing sentence about runs stored before requestedSpec is unchanged: those runs did use that group.
  • get_run's consumerGroup description says the same, and testdata/mcp/tools/get_run.json is regenerated.
  • Refs: the TestOrchestrator.java:1619-1623,1714-1718 anchor "integrity-cg") becomes "-integrity-0-group"), with the range unchanged. The four refs into NativeKafkaBackend.java are re-pointed past the new imports (:786 → :792 in mcp_caveats.go; mcp_tools_runs.go :2248 and :2312; mcp_tools_scenario.go :1285), keeping their anchors.

Book

  • 03, Consumer Groups, Offsets and Lag: a native LOAD, ENDURANCE or INTEGRITY consumer starts each partition its group has no committed offset for at the first record written since its task was submitted, less a second. It used to say earliest, the oldest record on the topic.
  • 05: the INTEGRITY consumerGroup row's default is the run's own, <run id>-integrity-0-group-integrity. After the LOAD "Interpreting Results" table, one sentence on earlier runs (see Decisions).
  • 08: the integrity-under-chaos diagram, its alt text and the transactional walkthrough say the consumer reads from the run's first record, not from the start of the topic, and link to book 03.
  • 11: consumerGroup defaults, <task id>-group or <task id>-group-integrity, and where a new group starts on the native backend.
  • 13: the scenario-file field table's INTEGRITY default.

Tests

  • New NativeKafkaBackendConsumerStartTest, plain JUnit. It drives sinceSubmit on a MockConsumer subclass, because the 3.9.2 MockConsumer throws UnsupportedOperationException from offsetsForTimes. The override answers from a map and records the timestamps it was asked for.
    • aPartitionWithNoCommittedOffsetStartsAtTheFirstRecordSinceSubmit: the lookup finds offset 5, the end is 8. The position is 5, and the lookup asked for the submit time less 1,000 ms.
    • aPartitionWithNothingSinceStartsAtTheEndReadBeforeTheLookup: the lookup finds nothing and, while it runs, raises the end from 3 to 9, standing for the producer's writes. The position is 3, so no record is skipped.
    • aCommittedPartitionResumesFromIt: offset 7 committed. The position is 7, and that partition is not looked up.
  • TestOrchestratorTest: defaultConsumerGroupIsIntegrityCg is renamed defaultConsumerGroupIsTheTasksOwn. It and integrityConsumerGroupKeepsItsDefaultWhenUnset expect run-1-integrity-0-group.
  • TestMCPGoldenToolsList with the regenerated get_run.json. TestMCPCaveatRefsExist and TestMCPCaveatRefAnchors pass after the re-pointing, with no test code changed.

Local checks, all on this branch:

  • git diff --name-only origin/main...HEAD: one commit on origin/main. It changes kates/** (NativeKafkaBackend, TestOrchestrator, 2 tests), cli/** (mcp_caveats, mcp_tools_runs, mcp_tools_scenario, testdata get_run.json, examples/perf-integrity.yaml) and docs/book/** (03, 05, 08, 11, 13). That triggers ci.yml (code), lint.yml (java, go, yaml), docs.yml, book.yml, integration.yml and ci-docker.yml. charts, Makefile, .github and scripts are untouched.
  • cd kates && ./mvnw -o test -Dquarkus.grpc.codegen.skip=true -Dquarkus.http.test-port=0: 1,469 tests, 0 failures, 0 errors, 0 skipped.
  • JAVA_HOME=$HOME/.sdkman/candidates/java/25.0.2-graalce ./mvnw -o spotless:check -Dquarkus.grpc.codegen.skip=true: clean.
  • cd cli && go test ./...: all packages ok, including cli/cmd, whose caveat-ref anchor tests read this branch's Java.
  • cd cli && go test -p 1 -race -count=1 ./..., CI's form, with the race detector: ok.
  • go vet ./..., golangci-lint run --new-from-rev=origin/main ./... (0 issues), govulncheck ./... (no vulnerabilities).
  • scripts/check-cli-style.sh, scripts/check-cli-compat.sh.
  • scripts/check-book-style.sh (38 pages within the structure baseline), scripts/gen-version-matrix.sh --check, scripts/gen-chart-table.sh --check, scripts/check-book-contrast.py (170 colour pairs meet WCAG AA).
  • scripts/check-mermaid.sh --quiet: 156 diagrams, 0 invalid.
  • A scratch-copy book render (scripts/prepare-book-render.sh, then quarto render docs/book --to html): 39 chapters prepared, no errors or warnings, worktree left clean. scripts/check-book-fragments.py docs/book/_book: all 7,459 fragment links in 38 pages land, including the new 03-cluster.md#consumer-groups-offsets-and-lag.
  • yamllint -c .yamllint.yaml .github config charts: 0 errors. CI's yaml scope doesn't include cli/examples/perf-integrity.yaml, which triggers the filter; a direct yamllint of it also passes.
  • Not run locally: lychee (not installed). By hand, the only new link target is 03-cluster.md#consumer-groups-offsets-and-lag, which matches the heading "Consumer Groups, Offsets and Lag", and the glossary anchors it uses already exist.
  • Not needed: helm lint/template, check-versions.sh, make check-help, actionlint and shellcheck, since no charts, versions, Makefile, .github or scripts file changed.

Only in CI:

  • the Java *IT tests (./mvnw verify in ci.yml Build & Test) and the JaCoCo report;
  • Trivy on the Maven dependencies (ci.yml dependency-scan);
  • the native image build (native.yml);
  • the Docker image builds, jvm and native for amd64 and arm64 (ci-docker.yml);
  • the Kind cluster smoke tests (integration.yml);
  • lychee's Markdown link check (docs.yml);
  • the book PDF render, on push only (book.yml);
  • CodeQL (codeql.yml);
  • the github-advanced-security AI scan, which is red on every PR, on GitHub's side.

Decisions

  • Default INTEGRITY group: the task id plus -group, with the backend's -integrity suffix kept for every group, giving <run id>-integrity-0-group-integrity. That keeps one rule for named and default groups. The alternative, appending -integrity only to a named group, gives a shorter default but two rules.
  • perf-integrity.yaml: consumerGroup is dropped, not renamed, because any name its runs share keeps the problem.
  • Runs from before this change: one sentence in book 05 and this note; no new MCP caveat and no new IT.
    • On a reused topic, a LOAD or ENDURANCE run created before this change read every earlier run's records first, so its consume throughput, and the run's avgThroughput, are understated by up to the ratio of all the records it read to its own.
    • After this change those numbers step up on reused topics. assess_run's noise band, and saved baselines that mix older and newer runs, will show the step.
    • ENDURANCE numbers change too, through the shared runConsumer.
  • kafka-clients 4.4.0: Dependabot's chore(deps): bump org.apache.kafka:kafka-clients from 3.9.2 to 4.4.0 in /kates #362 touches the same consumer API: the rebalance listener, offsetsForTimes and MockConsumer. Whichever lands second should rerun NativeKafkaBackendConsumerStartTest.

Deliberately not done

  • No IT. Only an IT would exercise the real committed() and offsetsForTimes() on a new group; in the unit test MockConsumer stands in for them. A CI-only IT in EngineLifecycleIT (fill a unique topic with another run's records, then run INTEGRITY and LOAD on it and assert PASS and a consume count equal to the run's own) is left for a follow-up.
  • Trogdor's CONSUME task (TrogdorBackend) still uses Trogdor's own offset reset. Not fixed here.
  • Known limits, unchanged on purpose:
    • A topic that stamps LogAppendTime, with a broker clock more than a second behind the Kates API, would start after the run's first records, and INTEGRITY would report them lost. Kates-created topics take the broker default, CreateTime, and nothing in charts/ or config/ sets message.timestamp.type.
    • Overlapping runs that share a named group still split its partitions, and a named group with an old committed offset still reads its backlog first. The docs already say to give each test its own group.
    • Every default INTEGRITY run now leaves its own group in __consumer_offsets until offsets retention expires, as LOAD runs already do, so kates cluster groups lists more groups.
  • kates/src/test/resources/spec-applicability.json:11 keeps integrity-cg: it is a group a request names in an applicability case, not the default.
  • No release note here; the next chore(release) PR collects it ("LOAD numbers step up on reused topics").

Landing

Lands after #375 (plan PR 6, compound runs and resilience faults through the guard and the lease). Both edit:

The hunks are disjoint, and git merge-tree of this branch with #375's head merges with no conflict today. #375 doesn't touch NativeKafkaBackend.java or TestOrchestrator.java, so this PR's refs should hold.

Once #375 merges, rebase onto main. Then:

  1. Re-point any caveat ref that moved, from a diff line map (git diff -U0 against the new main on the files the refs cite), not by eye.
  2. Rerun cd cli && go test ./cmd -run TestMCPCaveat.

The next plan PR in wave 2 (G) edits the same mcp_caveats.go ref block (:115-118) and will need the same treatment after this one.

@bmscomp
fix(engine): each INTEGRITY run reads in a group of its own, and a co…
5d25289 
…nsumer with no committed offset starts at its task's first record, not the topic's oldest

Every INTEGRITY run that named no consumerGroup joined one group,
integrity-cg-integrity. Run A produces for 10 minutes; two minutes in,
run B, one minute long, starts on the same default topic. B's consumer
reads past A's records, skipping them as another run's, and auto-commits
up to the head. When A's consumer starts after its produce phase, it
resumes at B's offset and never sees A's first minutes, so A reports
those acked records lost: DATA_LOSS, about 40% lost, while the cluster
lost nothing. Overlapping consume phases split the partitions between
the two runs, with the same result.

A LOAD or ENDURANCE consumer had a group of its own, but a new group
started at earliest, the oldest record on the topic. On a reused topic
the Nth default LOAD run read the earlier runs' records before its own,
so its consume throughput, and the run's average, were understated by
up to the ratio of all the records it read to its own.

The INTEGRITY default is now the task id plus -group, as for LOAD, and
the backend still appends -integrity: <run id>-integrity-0-group-integrity.
The native backend subscribes with a rebalance listener. On each
assigned partition the group has no committed offset for, it reads the
end offset, then looks up the first record written since the task was
submitted, less a second, and seeks there, or to that end offset when
there is none. The end is read before the lookup so that a record the
producer writes in between is not skipped. The time is taken when the
orchestrator submits the task, not when its worker thread starts, which
on a loaded pod can be after the producer's first send. A partition with
a committed offset resumes from it, and earliest still covers one that
is out of range. kafka-clients 3.9 has no by_duration reset, hence the
listener.

cli/examples/perf-integrity.yaml no longer names integrity-cg, since
any named group shared by every run of the example has the same
problem. The merged-spec caveat and get_run's consumerGroup description
in kates mcp give the new default and keep integrity-cg-integrity for
runs created before. Books 03, 05, 08, 11 and 13 describe the default and
where a new group starts, and book 05 says that earlier LOAD and
ENDURANCE runs on a reused topic understated their consume throughput.
Caveat refs into NativeKafkaBackend are re-pointed past the new imports,
and the TestOrchestrator anchor follows the new default.
This was referenced Oct 17, 2026
Merged
Merged
Merged
Merged
Merged
Merged
bmscomp added a commit that referenced this pull request Oct 17, 2026
@bmscomp
test(kates): the tests that stub a mock a scheduled job also calls ru…
feb9acb 
…n with the scheduler off, so the job cannot take the stub (#385)

AdvisorResourceTest.analyzeReturns404ForMissingRun failed on CI (PR #376)
with "WrongTypeOfReturnValue: Optional cannot be returned by
findByStatus()". Mockito ties when(mock.call()).thenReturn(...) to the
last call made on that mock, from any thread. The timeout reaper calls
findByStatus on the mocked TestRunRepository every 60 s; when that call
lands between the test's findById and its thenReturn, the stub goes to
findByStatus. doReturn().when() has the same window.

Each of the five classes stubs a mock that a job calls:
- AdvisorResourceTest, ReportResourceTest: TestRunRepository, called by
  TestTimeoutReaper.reapStuckTests (findByStatus).
- ScheduleResourceTest: ScheduledTestRunRepository, called by
  TestScheduler.evaluateSchedules (findAllEnabled); TrogdorClient, called
  by the run reconciler for the runs the test starts.
- TestResourceTest: TrogdorClient, called by
  TestOrchestrator.reconcileActiveRuns for the runs the test starts.
- TrendServiceTest: ClusterHealthService, called by
  KafkaReachabilityCache.refresh (isReachable).

None of them needs a job to fire: they read runs over GET, which
refreshes the run itself. SchedulerOffTestProfile sets
quarkus.scheduler.enabled=false and the five classes use it. Every other
test keeps the scheduler as before.
@bmscomp
bmscomp merged commit df09be5 into main Oct 17, 2026
65 of 67 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@bmscomp