anshul98ks123 opened a new pull request, #20137:
URL: https://github.com/apache/pinot/pull/20137
## Problem
A realtime consumer drops any message its decoder can't decode right now. A
`null` row or any exception becomes a decode error:
`INVALID_REALTIME_ROWS_DROPPED` goes up and the offset moves past the message,
unless `stopOnDecodeError` stops the replica instead. That is right for a
malformed payload. It is wrong when the decoder depends on a service that is
briefly unavailable. With a schema registry, a timeout on the first lookup of a
new schema id loses the message permanently, although the same message decodes
a few seconds later.
## Root cause
`StreamMessageDecoder` has two outcomes: a row, or drop. The Confluent
decoders rethrow registry errors, and the consumer can't tell a transient
failure from a bad payload.
## Approaches
| Option | Trade-off |
|---|---|
| Block and retry inside each decoder | Ignores `_shouldStop`, end criteria
and catch-up deadlines; every plugin reimplements it |
| `stopOnDecodeError=true` | Stops the replica until validation repairs it;
a truly bad message stalls it for good |
| **Retryable outcome in the SPI; the consumer keeps the offset and
re-fetches with backoff** | **Chosen** |
## Change
- **pinot-spi:**
- new unchecked `RetriableDecodeException`;
- new `StreamDataDecoderResult.isRetriable()`;
- the `StreamMessageDecoder` javadoc states the contract.
- **`RealtimeSegmentDataManager`:** on a retriable result, it ends the batch
without advancing `_currentOffset`, backs off (100 ms doubling to
`decodeRetryMaxBackoffMillis`, default 10 s), and re-fetches from the same
offset.
- `_shouldStop` and end criteria are checked between attempts.
- Catch-up stays bounded by its deadline.
- A time threshold reached while stuck ends the segment at the stuck
message.
- `decodeRetryMaxDurationMillis` bounds the retrying: no limit by default,
and `0` restores the old behaviour.
- New meter: `realtimeDecodeRetries`.
- **`StatelessRealtimeSegmentWriter`:** retries the same way, within the
re-ingestion timeout.
- **Confluent Avro / JSON / Protobuf decoders:** throw
`RetriableDecodeException` for:
- Kafka `RetriableException`;
- registry 408, 429 and 5xx other than 501;
- socket-level errors.
Unknown magic byte, 404, 401/403 and malformed payloads behave as before.
- **Separate commit:** within a batch, `INVALID_REALTIME_ROWS_DROPPED` and
`REALTIME_ROWS_FILTERED` shared one cached meter, so whichever was marked first
absorbed the other's counts.
## Behaviour change
During a registry outage, tables on the Confluent decoders now pause instead
of dropping rows. Ingestion delay and `realtimeDecodeRetries` show the pause.
`decodeRetryMaxDurationMillis=0` restores dropping.
## Testing
- New and extended tests: `RealtimeSegmentDataManagerTest` (48),
`StatelessRealtimeSegmentWriterTest`, `StreamConfigTest`,
`StreamDataDecoderImplTest`, and the Confluent Avro/JSON/Protobuf decoders
against a local HTTP registry.
- On unmodified master, 9 of the 10 new consumer tests fail, with rows
dropped and the offset advanced past the message.
- Spotless, license and checkstyle pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
