Kino1994 opened a new pull request, #58043:
URL: https://github.com/apache/spark/pull/58043
### What changes were proposed in this pull request?
This PR lets a topic key in the `startingOffsets` / `endingOffsets` JSON map
to the string `"earliest"` or `"latest"`, in addition to the existing
partition-to-offset object. Both forms can be mixed in the same document:
```
startingOffsets={"topicA":"earliest","topicB":{"0":23,"1":-1},"topicC":"latest"}
```
A topic-level `"earliest"` is equivalent to writing `-2` for every partition
of that topic assigned to the query, and `"latest"` to `-1`. Because the
expansion happens against the discovered partition set rather than a hard-coded
list, no partition count is baked into the option.
Implementation:
- `JsonUtils.specificOffsets` (new) walks the JSON AST and splits
topic-level strings from partition maps. `JsonUtils.partitionOffsets` is left
untouched and stays strict, so checkpointed offsets cannot accidentally accept
the new syntax.
- `SpecificOffsetRangeLimit` gains `topicOffsets: Map[String, Long]` plus a
`resolve(assignedPartitions)` method. The parameter is by-name, so a fully
enumerated limit never pays for discovering the partitions.
- `KafkaOffsetReader.fetchSpecificOffsets` now takes the limit instead of a
bare map; both the admin- and consumer-based readers expand against their
assigned partitions. `fetchPartitionOffsets` reuses the partition set it has
already fetched.
- Batch queries reject a topic bound to `latest` (start) or `earliest`
(end), mirroring the existing per-partition checks.
- A topic-level entry naming a topic with no assigned partition fails with a
new error condition, `KAFKA_TOPIC_OFFSET_DOES_NOT_MATCH_ASSIGNED`, rather than
silently expanding to nothing, so a typo in a topic name cannot cause the query
to read less than intended.
- `checkOffsetLimitValidity` no longer requires matching topic-partition key
sets between start and end when topic-level entries are present, since the two
sides are legitimately not comparable before resolution. Matching them up is
left to the offset readers, which see the assigned partitions.
JIRA: https://issues.apache.org/jira/browse/SPARK-58774
### Why are the changes needed?
Today the source accepts `"earliest"`, `"latest"`, or a JSON object
enumerating every topic-partition. There is no way to express "earliest for
topic A, latest for topic B" without the fully enumerated form.
In steady state this rarely matters, since a single `"earliest"` or
`"latest"` covers the whole subscription. It becomes acute during incident
recovery or a targeted replay, when one topic has to be rewound while the
others stay where they are. With 24 partitions across 5 subscribed topics, that
is a single-line string of several thousand characters containing 120 entries
whose only content is the sentinel `-2`:
- **Verbosity.** In `.properties` or YAML configuration such a value is a
frequent source of accidental line breaks, which truncate it and surface as a
confusing `IllegalArgumentException`.
- **Opacity.** `-1` and `-2` are magic numbers; a single mistyped sentinel
silently changes the semantics for one partition.
- **Brittleness under repartitioning.** The JSON hard-codes the partition
count. Expanding a topic from 24 to 32 partitions makes the enumerated set stop
matching the assigned one, and the query fails to start with
`KAFKA_START_OFFSET_DOES_NOT_MATCH_ASSIGNED`. Failing loudly is the right
behaviour, but the only remedy is to hand-edit the multi-thousand-character
string after every repartitioning.
- **No composability.** Wanting a different position for a single topic
forces explicit enumeration of all the others.
### Does this PR introduce _any_ user-facing change?
Yes, and it is purely additive.
The topic-level value is disambiguated by JSON type: an object continues to
be parsed as a partition-to-offset map, while a string is the new shorthand.
Every currently valid configuration keeps its meaning, and no previously
accepted document changes behaviour.
Previously:
```
startingOffsets={"topicA":"earliest"}
-> IllegalArgumentException: Expected e.g.
{"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}
```
Now the same option is accepted and applies `earliest` to every partition of
`topicA`.
Two new failure modes, both for input that was previously rejected outright:
- An unrecognised string value (for example `{"topicA":"first"}`) is
rejected at parse time with a message naming the accepted forms.
- A topic-level entry for a topic with no assigned partition raises
`KAFKA_TOPIC_OFFSET_DOES_NOT_MATCH_ASSIGNED`.
As with the enumerated form, the JSON has to account for every topic
subscribed at the moment the offsets are resolved, so with `subscribePattern`
the matched topics must be known then. Topics matched later are discovered as
new partitions and start at earliest, exactly as today. The documentation for
both options has been updated accordingly.
### How was this patch tested?
New tests:
- `KafkaOffsetRangeLimitSuite` (new file) covers `resolve`: expansion
against the assigned partitions, the short-circuit that avoids fetching
partitions when nothing needs expanding, and the unmatched-topic error.
- `JsonUtilsSuite` covers parsing of the topic-level form, the mixed form,
case insensitivity, and rejection of malformed input.
- `KafkaRelationSuite` covers batch end to end: topic-level offsets, the
mixed form, `subscribePattern`, repartitioning between runs, and the two ways
the offsets can fail to line up with the topics a pattern matches.
- `KafkaMicroBatchSourceSuite` covers streaming end to end: a query started
from topic-level offsets with per-topic strategies, and a topic created after
the initial offsets were resolved being picked up as a new partition rather
than tripping the strict topic check.
- `KafkaRelationSuite` "bad batch query options" gains the topic-level
`latest`/`earliest` rejections.
The batch and streaming tests were verified on both V1 and V2, and the
streaming ones on both the admin- and consumer-based offset readers.
The full `sql-kafka-0-10` suite passes: 37 suites, 602 tests, 0 failures.
`scalastyle` is clean for both `Compile` and `Test`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
--
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]
