sjhajharia opened a new pull request, #22858:
URL: https://github.com/apache/kafka/pull/22858
### Rationale
The existing DLQ system tests (`ShareConsumerDLQTest`) and the JUnit
integration suite of the same name both exercise record-copy correctness
(`errors.deadletterqueue.record.copy.enable`) against source records that are
still on local disk. Neither exercises the path where the DLQ producer has to
read a rejected/expired record back from **tiered storage** after it has been
offloaded and deleted locally — the JUnit suite only covers this with a mocked
`RemoteLogManager`, never against a real broker with a real
`RemoteStorageManager` plugin wired in and real segment offload/deletion timing.
That gap matters because the DLQ copy path re-reads the original record via
a normal fetch, and tiered reads go through a completely different code path
(`RemoteLogManager` / `RemoteStorageManager`) than local reads. A regression
that only breaks the remote-read path would be invisible to every other DLQ
test in the suite.
This PR adds two ducktape tests that produce records, force them onto
`LocalTieredStorage` and off local disk (via a very short `local.retention.ms`
and small segment sizes), then reject them through a share consumer and verify
the DLQ topic ends up with the correct values — proving the DLQ fetch pulled
the original record back from remote storage rather than silently producing a
null/wrong value or hanging.
### Changes
- **`build.gradle`**: add `testImplementation
testFixtures(project(':storage'))` to `core`, so `LocalTieredStorage` (in
`storage`'s test fixtures) is available on the broker's test classpath
(`core/build/dependant-testlibs/`) for ducktape runs.
- **`tests/kafkatest/services/kafka/kafka.py`**: add
`earliest_local_offset(topic, partition)`, a thin wrapper around
`kafka-get-offsets.sh --time earliest-local` (KIP-405's
`OffsetSpec.earliestLocal()`), used to detect when segments have actually been
tiered and removed locally.
-
**`tests/kafkatest/tests/client/share_consumer_dlq_tiered_storage_test.py`**
(new): a single-broker `ShareConsumerDLQTieredStorageTest` with
`LocalTieredStorage` wired onto the broker via
`remote.log.storage.manager.class.path` (pointed at the resolved test-fixtures
jar file, not the whole lib directory, to avoid a `ClassCastException` from
double-loading the storage jars through `ChildFirstClassLoader`). Two tests:
1. `test_tiered_topic_dlq_reject` — every record is tiered and evicted
locally, then REJECTed; DLQ values must match the originals, which can only
happen if the DLQ read came from remote storage.
2. `test_mixed_tiered_and_local_dlq_reject` — first batch is tiered and
evicted locally, second batch is produced afterwards and stays local; both are
REJECTed together and the DLQ must contain correct values for both, exercising
the remote-read and local-read paths in the same run.
Both tests run against `@matrix(metadata_quorum=[isolated_kraft,
combined_kraft])` consistent with the rest of the DLQ test suite.
### Test plan
Ran on a fresh 14-node ducker-ak cluster
(`TC_PATHS=tests/kafkatest/tests/client/share_consumer_dlq_tiered_storage_test.py`):
| Test | Quorum | Time |
|---|---|---|---|
| `test_tiered_topic_dlq_reject` | ISOLATED_KRAFT | 44.0s |
| `test_tiered_topic_dlq_reject` | COMBINED_KRAFT | 35.7s |
| `test_mixed_tiered_and_local_dlq_reject` | ISOLATED_KRAFT | 51.6s |
| `test_mixed_tiered_and_local_dlq_reject` | COMBINED_KRAFT | 46.2s |
--
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]
