Joaquín Antonio De Vicente López created SPARK-58774:
--------------------------------------------------------
Summary: Support topic-level earliest/latest values in the Kafka
source startingOffsets/endingOffsets JSON
Key: SPARK-58774
URL: https://issues.apache.org/jira/browse/SPARK-58774
Project: Spark
Issue Type: Improvement
Components: Structured Streaming
Affects Versions: 4.2.0
Reporter: Joaquín Antonio De Vicente López
h2. Problem
The Kafka source accepts {{startingOffsets}} in three forms:
* {{"earliest"}} — applies to every subscribed topic-partition
* {{"latest"}} — applies to every subscribed topic-partition
* a JSON object mapping every topic-partition to an explicit offset, where
{{-2}} means earliest and {{-1}} means latest
There is no way to express "earliest for topic A, latest for topic B". As soon
as two topics need different starting positions, the only option is the fully
enumerated JSON, and every partition of every topic must be listed explicitly.
On a production cluster with 24 partitions per topic and 5 subscribed
topics, that is 120 entries whose only content is the sentinel {{{}-2{}}}:
{code:none}
startingOffsets={"topicA":{"0":-2,"1":-2,"2":-2,...,"23":-2},
"topicB":{"0":-2,"1":-2,"2":-2,...,"23":-2},
"topicC":{"0":-1,"1":-1,"2":-1,...,"23":-1}}
{code}
This has several practical drawbacks:
# *Verbosity.* The value is a single-line JSON string of several thousand
characters. In {{.properties}} or YAML configuration this is a frequent
source of accidental line breaks, which silently truncate the value and
surface as a confusing \{{IllegalArgumentException: Expected e.g.
{"topicA":\{"0":23,"1":-1},"topicB":\{"0":-2}}}}.
# *Opacity.* {{-1}} and {{-2}} are magic numbers. A reviewer looking at 120
occurrences of {{-2}} cannot tell at a glance that the intent was simply
"earliest", and a single mistyped {{-1}} in the middle of the block silently
changes the semantics for one partition.
# *Brittleness under repartitioning.* The JSON hard-codes the partition count.
If a topic is expanded from 24 to 32 partitions, the configuration is now
incomplete and the new partitions fall back to the default rather than to the
position the user intended. Nothing in the configuration signals this.
# *No composability.* The all-or-nothing nature means that wanting a different
position for a single topic forces explicit enumeration of all the others.
h2. Proposal
Allow the value associated with a topic key to be the string {{"earliest"}} or
{{{}"latest"{}}}, in addition to the existing partition-to-offset object:
{code:none}
startingOffsets={"topicA":"earliest","topicB":"earliest","topicC":"latest"}
{code}
Mixed forms would be accepted, so per-partition control remains available where
it is actually needed:
{code:none}
startingOffsets={"topicA":"earliest",
"topicB":{"0":1234,"1":-2,...,"23":-2},
"topicC":"latest"}
{code}
Semantics: a topic-level {{"earliest"}} is equivalent to writing {{-2}} for
every partition of that topic that is assigned to the query at the time offsets
are resolved; {{"latest"}} is equivalent to {{{}-1{}}}. Because the expansion
happens against the discovered partition set rather than against a hard-coded
list, it is correct by construction when the partition count changes.
The same syntax should apply to {{endingOffsets}} for batch queries, for
symmetry.
h2. Backwards compatibility
The change 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. Existing configurations are unaffected, and no
currently valid document changes meaning.
An unrecognised string value should be rejected at parse time with a message
naming the accepted values, rather than being silently ignored.
I am happy to work on a patch if the approach seems reasonable.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
