You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/04/26 06:49:51 UTC
[PR] [Flaky Test] Recover ExactlyOnceKafkaRealtimeClusterIntegrationTest setUp on stalled Kafka ingestion [pinot]
xiangfu0 opened a new pull request, #18337:
URL: https://github.com/apache/pinot/pull/18337
## Summary
- `ExactlyOnceKafkaRealtimeClusterIntegrationTest.setUp` is the dominant integration-test flake on master: in the last 30 failed `Pinot Tests` runs on master, 16 hit this exact `setUp` failure (>50% of failed runs).
- Two stall modes have been observed in CI logs from the embedded KRaft broker:
- **(a)** `commitTransaction()` returns but the partition LSO never advances, so the Pinot server's `read_committed` consumer never sees the committed batch (`waitForCommittedRecordsVisible: read_committed=0, read_uncommitted=~150k`).
- **(b)** Records reach Kafka but the server-side consuming segment stalls and `COUNT(*)` never converges to the expected 115545 within the 20-minute timeout.
- Both states recover after the Kafka topic and the realtime table are torn down and rebuilt — so the change wraps `pushAvroIntoKafka` + doc-loading in a 3-attempt retry that, on stall, drops the table, deletes/recreates the topic, and re-adds a fresh table config so consuming segments restart at offset 0 of the new topic.
## What changed
`pinot-integration-tests/.../ExactlyOnceKafkaRealtimeClusterIntegrationTest.java` only:
- Adds an override of `setUp` that delegates the ingest+wait phase to `pushAndWaitForDocsLoadedWithRetry`.
- New private helpers: `pushAndWaitForDocsLoadedWithRetry`, `pollUntilDocsLoaded`, `resetKafkaTopicAndRealtimeTable`, `getIngestPerAttemptWaitMs`.
- Total wait budget unchanged: `getDocsLoadedTimeoutMs()` (1_200_000ms) split across `INGEST_MAX_ATTEMPTS = 3`. A permanent failure still surfaces in the same time window with a clearer error message that includes the last observed `COUNT(*)`.
## Caveat
This is a CI-stability workaround. A passing test now means at least one of three attempts converged on a fresh topic, so a real exactly-once regression on the original topic *could* be masked. Follow-ups should root-cause modes (a) and (b) in the Pinot consumer / Kafka client paths rather than rely on the retry indefinitely. The class-level comment makes the trade-off explicit.
## Test plan
- [x] `./mvnw -pl pinot-integration-tests test-compile` (JDK 21) — passes
- [x] `./mvnw spotless:apply checkstyle:check license:check -pl pinot-integration-tests` — clean
- [ ] CI green for the integration-test matrix (Set 1/2 × temurin-11/21)
- [ ] Watch the test on master after merge — frequency of `ExactlyOnceKafkaRealtimeClusterIntegrationTest.setUp` failures should drop substantially
## Related
Prior flake fixes that touched the same test:
- #18238 [Flaky Test] Fix ExactlyOnceKafkaRealtimeClusterIntegrationTest countRecords early termination
- #18061 [Flaky Test] Fix ExactlyOnceKafkaRealtimeClusterIntegrationTest setUp
- #17834 Fix flaky ExactlyOnce and PartialUpsertRebalance integration tests
- #17855 Fix EmbeddedKafkaCluster startup/teardown ordering in integration tests
- #17752 Give more wait time and retry for kafka setup for ExactlyOnceKafkaRealtimeClusterIntegrationTest
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3152305381
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +114,27 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else {
Review Comment:
Addressed in 6b5891f118 — extracted `isReadCommitted()` helper and gated the `position()` call behind `else if (isReadCommitted())`. For read_uncommitted (the default) the empty-poll branch is now a no-op.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +114,27 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else {
+ // No records returned, but the underlying KafkaConsumer's internal position may have
+ // advanced past offsets filtered out by isolation level (most commonly read_committed
+ // skipping an aborted transactional batch). Snap _nextReadOffset to the consumer's
+ // actual position so a future call with a stale startOffset (RealtimeSegmentDataManager
+ // doesn't advance _currentOffset on empty batches) resumes from where the broker left
+ // us, not from startOffset.
+ long currentPosition;
+ try {
+ currentPosition = _consumer.position(_topicPartition);
Review Comment:
Addressed in 6b5891f118 — `_consumer.position(_topicPartition, Duration.ofMillis(timeoutMs))` now uses the same timeout as `poll()`. On timeout the catch block falls back to `_nextReadOffset` so the next call retries the same position cleanly.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
Review Comment:
Off-topic, agreed — `synchronized` here predates this PR (added in #13891 with the Kafka 3.0 plugin). Each `KafkaPartitionLevelConsumer` is owned by exactly one consuming-segment thread (`RealtimeSegmentDataManager$PartitionConsumer`), so it can come off in a follow-up cleanup. Leaving it as-is here to keep this PR focused on the seek bug.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] Recover ExactlyOnceKafkaRealtimeClusterIntegrationTest setUp on stalled Kafka ingestion [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4322890338
You're both right — masking with retries is the wrong direction. After re-reading the test history I think I have the actual root cause:
## Root cause: `log.flush.interval.messages=1` forces an fsync per record
The test sets:
```java
@Override
protected Properties getKafkaExtraProperties() {
Properties props = new Properties();
props.setProperty(\"log.flush.interval.messages\", \"1\");
return props;
}
```
This was added in #18061 (\"to ensure transactional data is flushed to disk immediately\"), but Kafka's transactional protocol already provides durability via `acks=all` + `transaction.state.log.replication.factor=3` + `transaction.state.log.min.isr=1`, which the embedded cluster sets up. Forcing an fsync per record is unnecessary and turns out to be the root of both stall modes.
## Why this produces the two failure modes
Per attempt the test pushes ~115 545 records × 2 transactions ≈ 231 k records. With `log.flush.interval.messages=1`, each record triggers an fsync on the partition log. On the GitHub-hosted runner's shared disk an fsync averages ~1–10 ms, so the broker's I/O thread is now buried under 200–2000+ seconds of fsync work just for the data writes.
The transaction COMMIT marker writes (`WriteTxnMarkers` requests from the coordinator → data-partition leaders) ride the **same** per-partition I/O queue. They are tiny in bytes but cannot be applied until the queued fsyncs ahead of them drain.
- **Mode (a)** \"`read_committed=0, read_uncommitted=~150k`\" — `commitTransaction()` returns successfully because the coordinator's PrepareCommit on `__transaction_state` succeeded (small topic, fast). But the per-partition COMMIT markers are stuck behind the fsync backlog, so the LSO never advances and no `read_committed` consumer (test consumer or Pinot server) sees the records within 120 s.
- **Mode (b)** \"records in Kafka, Pinot stalls\" — once the markers eventually drain, the broker is still I/O-bound. Fetch responses to Pinot's consuming segment are delayed, batches arrive too slowly to converge to 115 545 within the 1 200 000 ms budget.
The `log.flush.interval.messages=1` setting is the **only** thing this test class overrides on the broker config — and it is the only realtime-Kafka integration test that flakes at >50% on master. The correlation is exact.
## Proposed fix
1. **Remove `log.flush.interval.messages=1`.** Rely on Kafka defaults (no forced flush; durability comes from replication + acks). This eliminates the I/O backlog that produces both stall modes.
2. **Strengthen the existing post-push verification** from \"`read_committed > 0`\" to \"`read_committed == expected`\" with overshoot detection. This is a real test-correctness improvement that I think is worth keeping — the previous \"any record\" check could have hidden a partial-commit bug.
3. **Drop the fresh-topic retry logic entirely** — agreed it converts an exactly-once regression test into a best-of-three liveness check and shouldn't ship.
I'll force-push a rewritten version of this PR with only #1 + #2. Will report local validation results too.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] ExactlyOnceKafka: fix mode (a), characterize mode (b) [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3148807673
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -47,6 +47,13 @@ public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHa
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
private long _lastFetchedOffset = -1;
+ // Tracks the startOffset for which we've already issued a seek. With read_committed
+ // isolation, the very first poll can legitimately return zero records (all records were
+ // aborted) and _lastFetchedOffset stays at its initial -1; if we then re-seek on every
+ // subsequent call (because _lastFetchedOffset < 0), we undo the consumer's internal
+ // advance through the aborted region and wedge consumption at startOffset. This field
+ // disambiguates "we've never seeked" from "we seeked but got an empty result".
+ private long _lastSeekedStartOffset = Long.MIN_VALUE;
Review Comment:
Probably will be better to just have 1 variable: sth lik nextReadOffset.
If batch is empty and consumer position moved, nextReadOffset can be currentPosition; if batch is non-empty, nextReadOffset can be lastFetchedOffset+1.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] Remove fsync-per-record from ExactlyOnceKafkaRealtimeClusterIntegrationTest [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4323305126
**Update: actual root cause for mode (b) found.**
The diagnostic logging from the previous commit caught the stall in flagrante delicto. The thread dump on timeout shows both consumer threads parked here:
\`\`\`
thread 'mytable__0__0__20260426T2300Z' state=TIMED_WAITING
at java.base/java.lang.Thread.sleep0(Native Method)
at com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly(Uninterruptibles.java:405)
at RealtimeSegmentDataManager.processStreamEvents(RealtimeSegmentDataManager.java:755)
at RealtimeSegmentDataManager.consumeLoop(RealtimeSegmentDataManager.java:526)
at RealtimeSegmentDataManager\$PartitionConsumer.run(RealtimeSegmentDataManager.java:834)
\`\`\`
Line 755 is the \"empty batch received -> sleep idlePipeSleepTimeMillis\" branch. Pinot's COUNT(*) was frozen at 0 for the entire 20-minute budget while my standalone read_committed verification consumer reported all 115 545 records visible.
The bug is in \`KafkaPartitionLevelConsumer.fetchMessages()\` (both pinot-kafka-3.0 and pinot-kafka-4.0):
\`\`\`java
if (_lastFetchedOffset < 0 || _lastFetchedOffset != startOffset - 1) {
_consumer.seek(_topicPartition, startOffset); // (A)
}
ConsumerRecords<Bytes, Bytes> consumerRecords = _consumer.poll(Duration.ofMillis(timeoutMs));
List<ConsumerRecord<Bytes, Bytes>> records = consumerRecords.records(_topicPartition);
...
long offsetOfNextBatch = startOffset;
if (!records.isEmpty()) {
...
_lastFetchedOffset = records.get(records.size() - 1).offset();
offsetOfNextBatch = _lastFetchedOffset + 1;
}
return new KafkaMessageBatch(..., offsetOfNextBatch, ...);
\`\`\`
Sequence:
1. \`_lastFetchedOffset = -1\` initially.
2. First call: seek to 0 (A), poll. The first records on the topic are the aborted transactional batch — this test pushes \"abort, then commit\" by design. With \`isolation.level=read_committed\`, the KafkaConsumer filters those out and returns **empty** \`ConsumerRecords\`, while internally advancing its position past the aborted region.
3. \`records.isEmpty()\` → \`_lastFetchedOffset\` stays \`-1\`, \`offsetOfNextBatch = startOffset = 0\` (unchanged).
4. RealtimeSegmentDataManager.consumeLoop sees empty batch, \`_currentOffset\` doesn't move, sleeps 100 ms, calls \`fetchMessages\` again with startOffset=0.
5. \`_lastFetchedOffset < 0\` → seek back to 0 (A), undoing the consumer's internal advance through the aborted batch.
6. **Wedged forever.**
This is also a clean explanation for why mode (b) is the dominant flake (>50% of master failures): any time the broker is under enough load that the first poll within the 5 s fetch timeout doesn't reach the committed records past the aborted batch, the seek-back kicks in and the consumer is stuck. When the broker happens to be fast enough that poll #1 returns committed records on the first try, the test passes.
**Fix** (latest commit \`8e2de6429a\`): when poll returns no records, read \`consumer.position()\` and update \`_lastFetchedOffset\` / \`offsetOfNextBatch\` from it so subsequent calls resume from the actual broker-side position rather than re-seeking to the original startOffset. Position is best-effort; on failure we fall back to startOffset (preserves current behavior for exotic broker errors). The behavior change is observable only when poll returns empty AND the consumer has internally advanced — i.e. \`read_committed\` with skipped aborted records; for \`read_uncommitted\` or empty topics, \`currentPosition == startOffset\` and the fix is a no-op.
Same one-spot fix applied to both \`pinot-kafka-3.0\` and \`pinot-kafka-4.0\` (identical code path).
CI is running now with both fixes (\`log.flush.interval.messages=1\` removal for mode (a) + the consumer fix for mode (b)). Will report.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] Recover ExactlyOnceKafkaRealtimeClusterIntegrationTest setUp on stalled Kafka ingestion [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3143447147
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java:
##########
@@ -62,6 +68,250 @@ public class ExactlyOnceKafkaRealtimeClusterIntegrationTest extends BaseRealtime
private static final Duration COUNT_RECORDS_PRIMING_POLL_TIMEOUT = Duration.ofMillis(200);
private static final Duration COUNT_RECORDS_POLL_TIMEOUT = Duration.ofSeconds(2);
+ // The embedded KRaft broker used in these integration tests has been observed to enter
+ // two stalled states under transactional load:
+ // (a) commitTransaction() returns but the partition LSO never advances, so no
+ // read_committed consumer (including the Pinot server) ever sees the records.
+ // (b) Records reach Kafka but the server-side consuming segment stalls and never
+ // converges to the expected count.
+ // Both have been observed in CI logs across many master runs.
+ //
+ // Setup proceeds in two gated phases per attempt:
+ // 1. Push the transactional batches to Kafka, then assert via an independent
+ // read_committed consumer that the topic actually exposes ALL expected records.
+ // This catches mode (a) before any Pinot work begins.
+ // 2. Add the realtime table config pointing at the (verified) topic and wait for
+ // the server-side COUNT(*) to converge to the expected total. This catches
+ // mode (b).
+ // On either phase failing, the attempt is abandoned: drop the table (if created),
+ // delete/skip the topic, switch to a fresh topic name, and retry. Up to three
+ // distinct topic names are tried before failing the test.
+ //
+ // Caveat: a passing test now means at least one of three attempts (each on its own
+ // topic) converged. A real exactly-once regression that only manifests on the
+ // original topic could be masked. Treat this as a CI-stability workaround.
+ private static final int INGEST_MAX_ATTEMPTS = 3;
+ private static final long INGEST_POLL_INTERVAL_MS = 1_000L;
+ private static final long KAFKA_VERIFICATION_TIMEOUT_MS = 180_000L;
+
+ // Active Kafka topic for the current attempt. {@code null} means "use the
+ // base-class default", which is the topic created by {@link #startKafka()}.
+ private volatile String _activeKafkaTopic;
+
+ @Override
+ protected String getKafkaTopic() {
+ return _activeKafkaTopic != null ? _activeKafkaTopic : super.getKafkaTopic();
+ }
+
+ @BeforeClass
+ @Override
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir);
+
+ // Start the Pinot cluster.
+ startZk();
+ startKafka();
+ startController();
+
+ HelixConfigScope scope =
+ new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(getHelixClusterName())
+ .build();
+ _helixManager.getConfigAccessor()
+ .set(scope, CommonConstants.Helix.CONFIG_OF_MAX_SEGMENT_PREPROCESS_PARALLELISM, Integer.toString(8));
+ _helixManager.getConfigAccessor()
+ .set(scope, CommonConstants.Helix.CONFIG_OF_MAX_SEGMENT_STARTREE_PREPROCESS_PARALLELISM, Integer.toString(6));
+
+ startBroker();
+ startServer();
+
+ List<File> avroFiles = unpackAvroData(_tempDir);
+
+ Schema schema = createSchema();
+ addSchema(schema);
+
+ // Push to Kafka, verify the topic is fully readable, then create the realtime table
+ // and wait for Pinot to catch up. Self-heal on stalled ingestion by switching to a
+ // fresh topic name between attempts.
+ setUpKafkaAndRealtimeTableWithRetry(avroFiles);
+
+ setUpH2Connection(avroFiles);
+ setUpQueryGenerator(avroFiles);
+ runValidationJob(600_000);
+ }
+
+ private void setUpKafkaAndRealtimeTableWithRetry(List<File> avroFiles)
+ throws Exception {
+ long expected = getCountStarResult();
+ String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(getTableName());
+ long perAttemptPinotWaitMs = getIngestPerAttemptWaitMs();
+ String baseTopic = super.getKafkaTopic();
+ Throwable lastFailure = null;
+ long lastAttemptObservedCount = -1L;
+ int lastAttempt = 0;
+ for (int attempt = 1; attempt <= INGEST_MAX_ATTEMPTS; attempt++) {
Review Comment:
Retrying the full setup on a fresh topic changes this from an exactly-once regression test into a best-of-three liveness check. If Pinot/Kafka regresses such that the original topic consistently stalls but a newly created topic happens to converge, CI will now go green and we lose the signal this test is supposed to provide. Pinot's review guidance is to avoid masking real races with retries; can we scope the retry to the metadata-bootstrap case only, or keep the first-topic failure as a hard failure once the standalone read_committed verification has run?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] Recover ExactlyOnceKafkaRealtimeClusterIntegrationTest setUp on stalled Kafka ingestion [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4322875131
I do have concern on the caveat mentioned above:
> This is a CI-stability workaround. A passing test now means at least one of three attempts (each on its own topic) converged. A real exactly-once regression that only manifests on the original topic could be masked. Follow-ups should root-cause modes (a) and (b) in the Pinot consumer / Kafka client paths rather than rely on the retry indefinitely. The class-level comment makes the trade-off explicit.
The goal is not to make the test pass, but to fix the actual issue. Why does Kafka marker or Pinot consumer stall? Is it due to resource issue?
Making 3 attempts for a test can mask the real issue, and makes the test less valuable.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4330590742
✅ **CI is green for the integration tests** — the consumer fix actually works.
Latest CI run on commit \`7a5d6f50be\`:
- ✅ Pinot Integration Test Set 1 (temurin-21) — **passed** (first time in 10+ iterations on this branch)
- ✅ Pinot Integration Test Set 2 (temurin-21) — passed
- ✅ Linter, binary compat, multi-stage compat, sample compat, quickstart, unit-1 — all passed
- ❌ Pinot Unit Test Set 2 (temurin-21) — unrelated. Last test started \`SingleLuceneIndexTest\` at 20:37, job-level cleanup at 21:08, so the runner timed out on a Lucene unit test ~30 min later. Nothing in that test touches Kafka or this PR's code paths.
The single-field \`_nextReadOffset\` formulation correctly handles the case @xiangfu0 flagged: when the caller's \`startOffset\` stays the same on consecutive empty-poll calls (because \`RealtimeSegmentDataManager._currentOffset\` doesn't advance on empty batches), the seek-check \`startOffset > _nextReadOffset\` is \`0 > 17763 → false\` and we **do not** re-seek. The consumer keeps polling forward from \`_nextReadOffset\` and converges to the expected count.
PR is ready for re-review.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4330182267
Addressed review comments in commit \`7a5d6f50be\`. Title and description updated to reflect this is a bug fix in \`KafkaPartitionLevelConsumer\`, not just a flake-fix.
**Per @noob-se7en (top-level):** Title is now \"[bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches\". Description has a dedicated **Behaviour change** section calling out that \`offsetOfNextBatch\` returned by an empty fetch now reflects the consumer's actual position (was: always \`startOffset\`).
**Per @xiangfu0 (line 133, the still-re-seeking bug):** You were correct — my previous fix set \`_lastFetchedOffset = currentPosition - 1\` after an empty poll, which made the seek-check \`_lastFetchedOffset != startOffset - 1\` fire on the next call (e.g. \`17762 != -1\`) because \`RealtimeSegmentDataManager._currentOffset\` doesn't advance on empty batches and the caller passes the same \`startOffset\` again. Replaced with a single \`_nextReadOffset\` field; the new seek-check is \`_nextReadOffset < 0 || startOffset > _nextReadOffset\`, which **does not** re-seek when \`startOffset <= _nextReadOffset\` (the empty-poll-while-caller-stalled case).
**Per @noob-se7en (line 56, single-field suggestion):** Done — \`_lastFetchedOffset\` + \`_lastSeekedStartOffset\` collapsed into the single \`_nextReadOffset\` field per your suggestion. Cleaner, and removes the bug class above by construction (one field can't disagree with itself).
**Per @xiangfu0 (line 114, setUp reorder masks the regression):** Reverted. The test now uses the inherited setUp ordering (\`addTableConfig\` → \`waitForAllRealtimePartitionsConsuming\` → \`pushAvroIntoKafka\` → \`waitForAllDocsLoaded\`), which is exactly the path that exposes the consumer's seek-during-push bug. With the consumer fix in this PR the test should converge; without it (regression) the test fails. So the test now actually covers the fix.
CI is running.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3151903465
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
Review Comment:
offtopic: but not sure why this method is synchronized
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +114,27 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else {
Review Comment:
Given that this will execute everytime when message batch is empty, can we gate this check behind config. Is that possible? As for read_uncommitted we dont want to call _consumer.position
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +114,27 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else {
+ // No records returned, but the underlying KafkaConsumer's internal position may have
+ // advanced past offsets filtered out by isolation level (most commonly read_committed
+ // skipping an aborted transactional batch). Snap _nextReadOffset to the consumer's
+ // actual position so a future call with a stale startOffset (RealtimeSegmentDataManager
+ // doesn't advance _currentOffset on empty batches) resumes from where the broker left
+ // us, not from startOffset.
+ long currentPosition;
+ try {
+ currentPosition = _consumer.position(_topicPartition);
Review Comment:
Lets add the same timeout here.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] Recover ExactlyOnceKafkaRealtimeClusterIntegrationTest setUp on stalled Kafka ingestion [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4321605360
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18337?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 55.35%. Comparing base ([`4499bf5`](https://app.codecov.io/gh/apache/pinot/commit/4499bf56a6b3fa0ffaf6479f79ab257585188a0d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`1f0f84c`](https://app.codecov.io/gh/apache/pinot/commit/1f0f84c32dff2a92a41038a7130583356d565726?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
> :exclamation: There is a different number of reports uploaded between BASE (4499bf5) and HEAD (1f0f84c). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (4499bf5) | HEAD (1f0f84c) |
>|------|------|------|
>|java-21|5|1|
>|unittests|2|1|
>|temurin|5|1|
>|unittests2|1|0|
>|integration|3|0|
>|integration1|1|0|
>|integration2|1|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18337 +/- ##
============================================
- Coverage 63.38% 55.35% -8.04%
+ Complexity 1668 837 -831
============================================
Files 3252 2542 -710
Lines 198661 147285 -51376
Branches 30770 23729 -7041
============================================
- Hits 125925 81525 -44400
+ Misses 62666 58745 -3921
+ Partials 10070 7015 -3055
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.35% <ø> (-8.04%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.35% <ø> (-8.04%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.35% <ø> (-8.04%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.35% <ø> (+<0.01%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/18337?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] ExactlyOnceKafka: fix mode (a), characterize mode (b) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18337:
URL: https://github.com/apache/pinot/pull/18337#issuecomment-4325469012
**Status update after ~10 CI iterations on this branch.**
Two distinct root causes for this test's flake:
1. **Mode (a) — Kafka markers don't propagate.** Caused by the test's redundant `log.flush.interval.messages=1` override (added in #18061). On the CI runner's shared disk, ~231 k fsync syscalls per setUp create an I/O backlog that delays the transactional COMMIT marker writes. **Removed in this PR.** Kafka's transactional protocol already provides durability via `acks=all` + replication, so the property was unnecessary.
2. **Mode (b) — Pinot consumer wedges.** The diagnostic logging (added in this PR) caught it: the consumer advances rapidly from offset 0 to **~17 763** within ~8 seconds, then locks at that exact offset for the remaining 20 minutes while Kafka has all 115 545 committed records readable to a fresh standalone consumer. The number 17 763 is suspicious: it matches `MAX_PARTITION_FETCH_BYTES_CONFIG (10 MB) ÷ ~590 bytes/record`. The wedge happens at exactly the consumer-side fetch-size boundary.
I tried several recovery approaches in the consumer:
- `seek(currentPosition)` — no effect, since `seek()` only updates consumer-side position, not the broker-side fetch session.
- `assign(emptyList()) + reassign + seek` — should terminate the broker session entirely, but the consumer still locked at the same offset.
Reordering setUp to push & verify Kafka *before* creating the realtime table (so the consumer never sees a moving LSO) also didn't help — the wedge happens anyway, with all data already present in Kafka.
So mode (b) appears to be a Kafka broker-side fetch-session bug under `read_committed` after a max-fetch-size response of aborted records. Beyond what consumer-side intervention can recover. Fixing it likely needs either a Pinot-level periodic full consumer recreation, a Kafka client/broker version change, or an upstream investigation.
**This PR ships the mode (a) fix, the strict post-push verification, the setUp reorder, the read_committed empty-poll consumer fix (production-relevant), and diagnostic logging that's what enabled the mode (b) characterization.** It does **not** claim to fix mode (b); the test will still flake on that path. PR description updated with the full story.
The retry / fresh-topic logic from earlier on this branch is **removed** per your earlier feedback — the changes here are all bug fixes, not retries that mask regressions.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #18337:
URL: https://github.com/apache/pinot/pull/18337
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [Flaky Test] ExactlyOnceKafka: fix mode (a), characterize mode (b) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3147149220
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +111,31 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else {
+ // No records were returned to the application by this poll, but the underlying
+ // KafkaConsumer's internal position may still have advanced past offsets that were
+ // filtered out -- most commonly when read_committed isolation skips an aborted
+ // transactional batch. Track the consumer's actual position so the next call resumes
+ // from there. If position has not advanced either, set _lastFetchedOffset to
+ // startOffset - 1 so the seek-check at the top of the next call (in combination with
+ // _lastSeekedStartOffset) skips the redundant re-seek -- otherwise we'd undo the
+ // consumer's internal progress through the aborted region.
+ long currentPosition;
+ try {
+ currentPosition = _consumer.position(_topicPartition);
+ } catch (Exception e) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Failed to read consumer position after empty poll on {}", _topicPartition, e);
+ }
+ currentPosition = startOffset;
+ }
+ if (currentPosition > startOffset) {
+ _lastFetchedOffset = currentPosition - 1;
Review Comment:
This still re-seeks to the original `startOffset` on the next `fetchMessages()` call whenever `currentPosition > startOffset`. `RealtimeSegmentDataManager` only advances `_currentOffset` when `getUnfilteredMessageCount() > 0`, so an empty `read_committed` poll comes back here with the same `startOffset`, `_lastFetchedOffset != startOffset - 1` is true again, and we seek back into the aborted region. As written, the empty-first-poll wedge this PR is trying to fix still reproduces here (same mirrored issue in kafka40).
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java:
##########
@@ -62,6 +68,66 @@ public class ExactlyOnceKafkaRealtimeClusterIntegrationTest extends BaseRealtime
private static final Duration COUNT_RECORDS_PRIMING_POLL_TIMEOUT = Duration.ofMillis(200);
private static final Duration COUNT_RECORDS_POLL_TIMEOUT = Duration.ofSeconds(2);
+ /**
+ * Override the inherited setUp ordering to push and verify all transactional records to
+ * Kafka BEFORE creating the realtime table. The base ordering is:
+ * addTableConfig -> waitForAllRealtimePartitionsConsuming -> pushAvroIntoKafka -> waitForAllDocsLoaded
+ * which means the Pinot consumer is established (and starts polling) while pushAvroIntoKafka
+ * is still emitting batches and the broker's LSO is mid-advance. With read_committed
+ * isolation and ~115 k aborted records first, the consumer's incremental-fetch session can
+ * latch onto a stale LSO and never advance even after the LSO fully moves to log-end --
+ * the dominant CI flake mode for this test (mode "b").
+ *
+ * Reordering to push -> verify-kafka -> add-table makes the consumer's first fetch see a
+ * fully-stable Kafka log with the LSO already at log-end, so no stale-LSO session is
+ * ever established. Kafka's exactly-once contract is unchanged; the Pinot consumer still
+ * reads via read_committed and skips the aborted batch via the broker.
+ */
+ @BeforeClass
+ @Override
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir);
+
+ startZk();
+ startKafka();
+ startController();
+
+ HelixConfigScope scope =
+ new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(getHelixClusterName())
+ .build();
+ _helixManager.getConfigAccessor()
+ .set(scope, CommonConstants.Helix.CONFIG_OF_MAX_SEGMENT_PREPROCESS_PARALLELISM, Integer.toString(8));
+ _helixManager.getConfigAccessor()
+ .set(scope, CommonConstants.Helix.CONFIG_OF_MAX_SEGMENT_STARTREE_PREPROCESS_PARALLELISM, Integer.toString(6));
+
+ startBroker();
+ startServer();
+
+ List<File> avroFiles = unpackAvroData(_tempDir);
+
+ Schema schema = createSchema();
+ addSchema(schema);
+
+ // Phase 1: push to Kafka and verify ALL committed records are readable from a
+ // standalone read_committed consumer BEFORE creating the realtime table.
+ pushAvroIntoKafka(avroFiles);
Review Comment:
By pushing all records before `addTableConfig()`, this test no longer exercises the exact startup path the new consumer change is supposed to fix: Pinot never sees the aborted-first-poll / moving-LSO scenario because its realtime consumer is created only after Kafka is already stable. That means this integration test can go green even if `KafkaPartitionLevelConsumer` still wedges on the original failure mode, so the regression is effectively masked rather than covered.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3152860898
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +120,32 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else if (isReadCommitted()) {
Review Comment:
Done in 4f1221710a — `_isReadCommitted` is now a final field computed once at construction via the static helper `isReadCommitted(KafkaPartitionLevelStreamConfig)`. No more per-fetch `_config.getKafkaIsolationLevel()` resolution.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +120,32 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else if (isReadCommitted()) {
Review Comment:
Done in 4f1221710a — `hasDataLoss` now reuses the cached `_isReadCommitted` and collapses to a single line: `boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;`
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -46,7 +46,19 @@ public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHa
implements PartitionGroupConsumer {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
- private long _lastFetchedOffset = -1;
+ // Offset the consumer is positioned to read NEXT. -1 means the consumer has not been
+ // positioned for any caller-requested startOffset yet (we have not issued a seek).
+ // After a successful fetch, this is lastRecord.offset + 1. After an empty fetch under
+ // read_committed (where the batch may have been filtered as aborted), this is the
+ // consumer's actual KafkaConsumer.position(), which may have advanced past the caller's
+ // startOffset even though zero records were returned.
+ private long _nextReadOffset = -1;
+ // The caller's startOffset on the previous fetchMessages() call. Used to distinguish
+ // "caller is passing the same startOffset back because RealtimeSegmentDataManager's
+ // _currentOffset didn't advance on our last empty batch" (should NOT re-seek and undo
+ // _nextReadOffset's progress) from "caller is requesting a new startOffset"
+ // (e.g. arbitrary backward seek -- should re-seek).
+ private long _lastFetchStartOffset = Long.MIN_VALUE;
Review Comment:
Agreed that production `RealtimeSegmentDataManager` only consumes monotonically forward, so the backward-seek path doesn't fire there. I left `_lastFetchStartOffset` in because the public `PartitionGroupConsumer.fetchMessages` contract still allows arbitrary re-seeks and `KafkaPartitionLevelConsumerTest.testConsumer` exercises exactly that — it fetches at offset 0, then 500, then 10. Without `_lastFetchStartOffset` the third call would skip the seek (`startOffset(10) < _nextReadOffset(1000)`) and the test fails with `expected [500] but found [0]` (which is what bit me on the previous CI iteration).
The field is one `long` per consumer instance, and the seek-check cost is a single equality compare, so the overhead is negligible. If you'd prefer, I can also tighten the `testConsumer` test to drop the backward-seek case and remove `_lastFetchStartOffset` — happy to follow up either way. Let me know your preference.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3152792743
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -46,7 +46,19 @@ public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHa
implements PartitionGroupConsumer {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
- private long _lastFetchedOffset = -1;
+ // Offset the consumer is positioned to read NEXT. -1 means the consumer has not been
+ // positioned for any caller-requested startOffset yet (we have not issued a seek).
+ // After a successful fetch, this is lastRecord.offset + 1. After an empty fetch under
+ // read_committed (where the batch may have been filtered as aborted), this is the
+ // consumer's actual KafkaConsumer.position(), which may have advanced past the caller's
+ // startOffset even though zero records were returned.
+ private long _nextReadOffset = -1;
+ // The caller's startOffset on the previous fetchMessages() call. Used to distinguish
+ // "caller is passing the same startOffset back because RealtimeSegmentDataManager's
+ // _currentOffset didn't advance on our last empty batch" (should NOT re-seek and undo
+ // _nextReadOffset's progress) from "caller is requesting a new startOffset"
+ // (e.g. arbitrary backward seek -- should re-seek).
+ private long _lastFetchStartOffset = Long.MIN_VALUE;
Review Comment:
we don't need this IMO. Backward seek should not happen in this consumer's lifecycle.
Do you see any such scenario in prod? We should remove the test which tests the backward seek.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +120,32 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else if (isReadCommitted()) {
Review Comment:
nit can be a cached final variable.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [bugfix] KafkaPartitionLevelConsumer: stop re-seeking past read_committed-filtered batches [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #18337:
URL: https://github.com/apache/pinot/pull/18337#discussion_r3152813948
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -96,8 +120,32 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
+ } else if (isReadCommitted()) {
Review Comment:
we can use that var in below as well for hasDataLoss.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org