[CORE-2841] Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp) #19975

oleiman · 2024-06-24T21:09:00Z

This PR wires up the ability to configure the start offset of a transform at deploy time. This can be either a unix timestamp (ms since epoch) or an offset delta (+oo from start offset of -oo from end).

Includes rpk experience.

Backports Required

Release Notes

Improvements

Adds the ability to start transform processing from an arbitrary offset on the input topic.

src/v/transform/transform_processor.cc

oleiman · 2024-06-27T04:28:13Z

/ci-repeat 1

src/v/model/transform.cc

rockwotj · 2024-07-01T14:32:05Z

src/v/model/transform.h

+    serde::variant<
+      latest_offset,
+      model::timestamp,
+      model::transform_from_start,
+      model::transform_from_end>
+      position;


We need a feature gate to prevent using the new position options on a partially rolled out cluster.

Ah, right, because it's a new envelope.

this is done. feels odd to have a whole feature flag for such a small change, but maybe we just don't do things like this very often.

Maybe, or we just overlook it and customers are well behaved, but in this case if someone does this old brokers won't be able to read transform metadata - which would be real bad.

src/v/model/transform.cc

Thin proxy for `kafka::partition_proxy::start_offset` Signed-off-by: Oren Leiman <[email protected]>

model::offset_delta resolves arithmetic operator overloads that perform automatic conversions between kafka and model offsets. This commit introduces an offset_delta that is used specifically for applying a numeric delta to an existing kafka::offset without adjusting its type. Useful for transform start offset calculations. Signed-off-by: Oren Leiman <[email protected]>

oleiman · 2024-07-01T18:58:57Z

force push rebase dev to avoid conflicts on feature flag

oleiman · 2024-07-01T19:57:30Z

force push feature flag for the new position types, minor CR changes

oleiman · 2024-07-01T19:58:19Z

empty diff for commit signoff

rockwotj · 2024-07-01T20:47:28Z

src/v/transform/api.cc

+      || !_feature_table->local().is_active(
+        features::feature::transforms_specify_offset)) {


This blocks all deploys during the upgrade - we should only block usage of new position offsets because old versions will crash trying to read them.

Right, there might be a disconnect in my understanding of how the variant serialization works. Won't we invariably write out a transform_offset_options which is versioned on the new variant layout?

like if we allow a deploy (any deploy) to go through on the new version before the upgrade is complete, then we need to downgrade, won't metadata deser trigger this code?

redpanda/src/v/serde/rw/variant.h

Lines 146 to 156 in fd1434d

if (size != std::variant_size_v<UnderlyingType>) [[unlikely]] {

throw serde_exception(fmt_with_ctx(

ssx::sformat,

"reading type {} of size {}: {} bytes left - unexpected variant "

"size: {}, current variant size: {}, likely backwards compat issues.",

type_str<Type>(),

sizeof(Type),

in.bytes_left(),

size,

std::variant_size_v<UnderlyingType>));

}

done, i think. needs a test of some kind, but posting now for visibility

rockwotj · 2024-07-01T20:49:21Z

src/v/model/transform.h

+    serde::variant<
+      latest_offset,
+      model::timestamp,
+      model::transform_from_start,
+      model::transform_from_end>
+      position;


Maybe, or we just overlook it and customers are well behaved, but in this case if someone does this old brokers won't be able to read transform metadata - which would be real bad.

Thin wrappers around a kafka::offset_delta meant for use in a sum type to indicate how the delta should be applied - i.e. added to the start offset of a topic or subtracted from the latest offset of a topic. Enclosed delta should always be greater than or equal to zero. Signed-off-by: Oren Leiman <[email protected]>

oleiman · 2024-07-02T02:21:48Z

force push contents:

serde shenanigans
adjust feature gate to guard new offset alternatives only

As a consequence of this, to maintain compatibility during partial upgrades, introduce a legacy_transform_offset_options. This gives us the ability to feature gate use of the new offset alternatives and avoid writing non-forwards-compatible transform metadata in case a partial upgrade needs to be rolled back. This commit also adds transform_metadata::serde_write, which conditionally writes the legacy version of the offset options struct iff the position variant holds one of the legacy alternatives. This code can be removed after v24.3 ships. Signed-off-by: Oren Leiman <[email protected]>

Signed-off-by: Oren Leiman <[email protected]>

{ "format": enum[timestamp, from_start, from_end], "value": int64 } Signed-off-by: Oren Leiman <[email protected]>

Specifically, we want to prevent the use of the new offset options alternatives before an upgrade has been finalized. Signed-off-by: Oren Leiman <[email protected]>

@t

--from-offset to start from this offset * @t: start from UNIX timestamp (ms from epoch) * +oo: start offset + oo * -oo: latest ofset - oo Signed-off-by: Oren Leiman <[email protected]>

Signed-off-by: Oren Leiman <[email protected]>

- Consume records that were produced before the deploy - Specify offsets that run off the end of the input topic - Ill-formed offsets Signed-off-by: Oren Leiman <[email protected]>

oleiman · 2024-07-02T02:28:10Z

force push make legacy_transform_offset_options struct write-only.

rockwotj

One small edge case to redeem otherwise LGTM

rockwotj · 2024-07-02T02:50:16Z

src/v/model/transform.cc

@@ -102,6 +170,26 @@ std::ostream& operator<<(std::ostream& os, const transform_metadata& meta) {
    return os;
 }

+void transform_metadata::serde_write(iobuf& out) const {
+    serde::write(out, name);
+    write(out, input_topic);


No serde namespace?

surprisingly (to me anyway) apparently not. we're shooting for this overload:

redpanda/src/v/model/metadata.h

Lines 363 to 367 in fd1434d

friend void write(iobuf& out, topic_namespace t) {

using serde::write;

write(out, std::move(t.ns));

write(out, std::move(t.tp));

}

perhaps making that a free function in namespace serde would be a good cleanup though

Ah ADL before we had the tag invoke pattern

Just leave it it's fine for now

oh yeah wasn't going to change it today

rockwotj · 2024-07-02T02:51:42Z

src/v/transform/transform_processor.cc

+                  },
+                  [this, &latest](model::transform_from_end off) {
+                      vlog(_logger.debug, "starting at offset: {}", off);
+                      auto actual_offset = latest - off.delta;


sigh. goood catch. thought i made a note somewhere to address that 😕

rockwotj · 2024-07-02T02:52:01Z

src/v/transform/transform_processor.cc

+                  },
+                  [this](model::transform_from_start off) {
+                      vlog(_logger.debug, "starting at offset: {}", off);
+                      auto actual_offset = _source->start_offset() + off.delta;


oleiman self-assigned this Jun 24, 2024

github-actions bot added area/rpk area/redpanda area/wasm WASM Data Transforms labels Jun 24, 2024

rockwood-openai reviewed Jun 24, 2024

View reviewed changes

src/v/transform/transform_processor.cc Outdated Show resolved Hide resolved

This comment was marked as resolved.

Sign in to view

oleiman changed the title ~~Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp)~~ [CORE-2841] Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp) Jun 24, 2024

oleiman force-pushed the xform/core-2841/specify-start-offset branch 5 times, most recently from b177085 to 78e09c0 Compare June 25, 2024 17:26

oleiman marked this pull request as ready for review June 25, 2024 17:28

oleiman requested review from a team, twmb, r-vasquez, gene-redpanda and Deflaimun as code owners June 25, 2024 17:28

oleiman requested review from a team and michael-redpanda and removed request for a team June 25, 2024 17:29

oleiman force-pushed the xform/core-2841/specify-start-offset branch from 78e09c0 to 612cd16 Compare June 25, 2024 18:01

oleiman force-pushed the xform/core-2841/specify-start-offset branch from 612cd16 to 2d34443 Compare June 27, 2024 19:13

oleiman added doc-needed labels Jun 27, 2024

oleiman requested review from rockwood-openai and pgellert and removed request for rockwood-openai June 27, 2024 19:15

oleiman requested a review from rockwotj July 1, 2024 14:26

rockwotj reviewed Jul 1, 2024

View reviewed changes

kbatuigas previously approved these changes Jul 1, 2024

View reviewed changes

oleiman added 2 commits July 1, 2024 11:58

xform/io: Add transform::source::start_offset

6f11b97

Thin proxy for `kafka::partition_proxy::start_offset` Signed-off-by: Oren Leiman <[email protected]>

oleiman force-pushed the xform/core-2841/specify-start-offset branch from d0dd6cd to 4177c74 Compare July 1, 2024 18:58

oleiman dismissed stale reviews from kbatuigas and pgellert via 28b7d63 July 1, 2024 19:53

oleiman force-pushed the xform/core-2841/specify-start-offset branch from 4177c74 to 28b7d63 Compare July 1, 2024 19:53

oleiman force-pushed the xform/core-2841/specify-start-offset branch from 28b7d63 to 3b34b67 Compare July 1, 2024 19:57

oleiman requested review from rockwotj and pgellert July 1, 2024 20:12

rockwotj reviewed Jul 1, 2024

View reviewed changes

oleiman force-pushed the xform/core-2841/specify-start-offset branch from 3b34b67 to f743216 Compare July 2, 2024 02:10

oleiman requested a review from rockwotj July 2, 2024 02:12

oleiman added 7 commits July 1, 2024 19:22

x/api: Respect user-specified start offset at deploy time

99ae4bd

Signed-off-by: Oren Leiman <[email protected]>

admin/x: Add 'offset' to deploy request

e1bf3a9

{ "format": enum[timestamp, from_start, from_end], "value": int64 } Signed-off-by: Oren Leiman <[email protected]>

features: introduce feature spec for new transform offset options

7591408

Specifically, we want to prevent the use of the new offset options alternatives before an upgrade has been finalized. Signed-off-by: Oren Leiman <[email protected]>

rpk/x/deploy: Add flag to read from an arbitrary offset

e1fef16

--from-offset to start from this offset * @t: start from UNIX timestamp (ms from epoch) * +oo: start offset + oo * -oo: latest ofset - oo Signed-off-by: Oren Leiman <[email protected]>

dt/rpk: Add offset parameter to deploy_wasm

210e8bc

Signed-off-by: Oren Leiman <[email protected]>

dt/xform: Add tests for 'from-offset' option

f6f1e38

- Consume records that were produced before the deploy - Specify offsets that run off the end of the input topic - Ill-formed offsets Signed-off-by: Oren Leiman <[email protected]>

oleiman force-pushed the xform/core-2841/specify-start-offset branch from f743216 to f6f1e38 Compare July 2, 2024 02:27

rockwotj approved these changes Jul 2, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[CORE-2841] Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp) #19975

[CORE-2841] Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp) #19975

oleiman commented Jun 24, 2024 •

edited

Loading

This comment was marked as resolved.

oleiman commented Jun 27, 2024

rockwotj Jul 1, 2024

oleiman Jul 1, 2024

oleiman Jul 1, 2024

rockwotj Jul 1, 2024

oleiman commented Jul 1, 2024

oleiman commented Jul 1, 2024

oleiman commented Jul 1, 2024

rockwotj Jul 1, 2024

oleiman Jul 1, 2024

oleiman Jul 1, 2024

oleiman Jul 2, 2024

rockwotj Jul 1, 2024

oleiman commented Jul 2, 2024

oleiman commented Jul 2, 2024

rockwotj left a comment

rockwotj Jul 2, 2024

oleiman Jul 2, 2024

rockwotj Jul 2, 2024

rockwotj Jul 2, 2024

oleiman Jul 2, 2024

rockwotj Jul 2, 2024

oleiman Jul 2, 2024

rockwotj Jul 2, 2024

		\|\| !_feature_table->local().is_active(
		features::feature::transforms_specify_offset)) {

	if (size != std::variant_size_v<UnderlyingType>) [[unlikely]] {
	throw serde_exception(fmt_with_ctx(
	ssx::sformat,
	"reading type {} of size {}: {} bytes left - unexpected variant "
	"size: {}, current variant size: {}, likely backwards compat issues.",
	type_str<Type>(),
	sizeof(Type),
	in.bytes_left(),
	size,
	std::variant_size_v<UnderlyingType>));
	}

	friend void write(iobuf& out, topic_namespace t) {
	using serde::write;
	write(out, std::move(t.ns));
	write(out, std::move(t.tp));
	}

[CORE-2841] Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp) #19975

Are you sure you want to change the base?

[CORE-2841] Transforms: Start consuming from an arbitrary offset (numeric from start/end or timestamp) #19975

Conversation

oleiman commented Jun 24, 2024 • edited Loading

Backports Required

Release Notes

Improvements

This comment was marked as resolved.

oleiman commented Jun 27, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oleiman commented Jul 1, 2024

oleiman commented Jul 1, 2024

oleiman commented Jul 1, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oleiman commented Jul 2, 2024

oleiman commented Jul 2, 2024

rockwotj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oleiman commented Jun 24, 2024 •

edited

Loading