[
https://issues.apache.org/jira/browse/SPARK-58774?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Joaquín Antonio De Vicente López updated SPARK-58774:
-----------------------------------------------------
Description:
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.
was:
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.
> 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, 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]
