[
https://issues.apache.org/jira/browse/SPARK-58774?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58774:
-----------------------------------
Labels: kafka pull-request-available structured-streaming usability (was:
kafka structured-streaming usability)
> 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
> Priority: Minor
> Labels: kafka, pull-request-available, structured-streaming,
> usability
>
> 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".
> In steady state this rarely matters: a single {{"earliest"}} or {{"latest"}}
> covers the whole subscription, which is why {{subscribePattern}} works well
> today. The gap becomes acute during incident recovery or a targeted replay,
> when
> one topic has to be rewound while the others stay where they are — precisely
> the
> moment when hand-editing a multi-thousand-character string is most
> error-prone.
> 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 enumerated set no longer
> matches the assigned one and the query fails to start with
> {{KAFKA_START_OFFSET_DOES_NOT_MATCH_ASSIGNED}} ("Partitions specified for
> Kafka
> start offsets don't match what are 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.* 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, no edit is needed when the partition count changes: batch queries
> resolve
> on every run, and streaming queries on start. Partitions appearing later
> within
> a running streaming query keep the existing behaviour of starting at earliest.
> The same syntax should apply to {{endingOffsets}} for batch queries, for
> symmetry.
> As today, 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, exactly as
> with the enumerated form.
> 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.
> Likewise, a topic-level entry naming a topic with no assigned partition should
> be rejected rather than expanding to nothing, so that a typo in a topic name
> cannot silently cause the query to read less than intended.
> 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]
