Make Decaton can consume any topic with deserializer #241

ocadaruma · 2024-08-15T02:20:10Z

Motivation

As of Decaton 8.0.1, the most common way to use Decaton was to use DecatonClient as the producer and DecatonProcessor as the consumer, with implementing value Serializer/Deserializer respectively.
After TaskMetadata as header #238, DecatonClient no longer wraps tasks with DecatonTaskRequest protobuf, which means records produced by DecatonClient and ones produced by non-DecatonClient producers (serialized by Kafka's serializer) have identical serialized bytes.
- Then why not making Deserializer support consuming non-DecatonClient topics, without forcing users to implement TaskExtractor (which is bit bothersome than just a deserializer) ?

Summary of changes

Update documentations and examples
ProcessorsBuilder#consuming(String topic, Deserializer<T> deserializer) now deserializes messages bytes directly even when task-metadata header is missing. This is a breaking change
- Previous behavior when task-metadata header is missing was to unwrap as DecatonTaskRequest pb first, then deserialize serialized task in it. This was for <= 8.0.0 to 9.0.0 migration where old DecatonClient might produce records during the upgrade.
  - Since "records are produced by old client / or records are produced by non-Decaton client" is not distinguishable from Decaton, we add new decaton.parse.as.legacy.format.when.header.missing property to control the behavior for graceful migration.
- With regards to this, when task-metadata header is missing, we need to fill reasonable default of task metadata.
  - The only field must be filled might be TaskMetadata#timestampMillis.
    - e.g. to calculate delivery latency correctly
  - To fill timestampMillis, we add ConsumedRecord#recordTimestampMillis, which can be retrieved from record itself

Expected upgrade procedure from <= 8.0.1 to 9.0.0

Enable both decaton.retry.task.as.legacy.format and decaton.parse.as.legacy.format.when.header.missing
Upgrade all processor instances
Upgrade all decaton client
Disable decaton.retry.task.as.legacy.format
Disable decaton.parse.as.legacy.format.when.header.missing

I'll describe the detail in 9.0.0 release note

ocadaruma · 2024-08-15T02:43:53Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorProperties.java

-    public static final PropertyDefinition<Boolean> CONFIG_TASK_METADATA_AS_HEADER =
-            PropertyDefinition.define("decaton.task.metadata.as.header", Boolean.class, true,
+    public static final PropertyDefinition<Boolean> CONFIG_RETRY_TASK_AS_LEGACY_FORMAT =
+            PropertyDefinition.define("decaton.retry.task.as.legacy.format", Boolean.class, false,


I renamed the property and flipped the meaning, because

For decaton.parse.as.legacy.format.when.header.missing, I couldn't figure out good "enabled-by-default" naming.

Then, for decaton.task.metadata.as.header, during migration, "Disable one and enable one" might be too confusing

Probably task.in.legacy.format sounds more correct?

kawamuray

returning once

kawamuray · 2024-08-15T04:08:42Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorProperties.java

-    public static final PropertyDefinition<Boolean> CONFIG_TASK_METADATA_AS_HEADER =
-            PropertyDefinition.define("decaton.task.metadata.as.header", Boolean.class, true,
+    public static final PropertyDefinition<Boolean> CONFIG_RETRY_TASK_AS_LEGACY_FORMAT =
+            PropertyDefinition.define("decaton.retry.task.as.legacy.format", Boolean.class, false,


Probably task.in.legacy.format sounds more correct?

kawamuray · 2024-08-15T04:09:54Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorProperties.java

+     * Reloadable: yes
+     */
+    public static final PropertyDefinition<Boolean> CONFIG_PARSE_AS_LEGACY_FORMAT_WHEN_HEADER_MISSING =
+            PropertyDefinition.define("decaton.parse.as.legacy.format.when.header.missing", Boolean.class, false,


How about decaton.legacy.parse.fallback.enabled? (just to make it bit shorter...)

kawamuray · 2024-08-15T07:13:32Z

...ssor/src/main/java/com/linecorp/decaton/processor/runtime/internal/DefaultTaskExtractor.java

@@ -31,8 +31,13 @@

 @RequiredArgsConstructor
 public class DefaultTaskExtractor<T> implements TaskExtractor<T> {
+    private static final ThreadLocal<Boolean> parseAsLegacyWhenHeaderMissing = ThreadLocal.withInitial(() -> false);


h-m this might be a bit too hacky, given that we would have to maintain this compatibility mode for certain period of time isn't it?

Can we do it like;

delay DefaultTaskExctractor instantiation to the time calling ProcessorsBuilder#build

so that we can instantiate it in SubscriptionBuilder#build

at that point we already have access to properties, so we can simply pass a boolean flag to the DefaultTaskExtractor constructor?

Ah, that's way better. Let me fix

kawamuray · 2024-08-15T09:55:33Z

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorsBuilder.java

@@ -116,15 +127,24 @@ public ProcessorsBuilder<T> thenProcess(DecatonProcessor<T> processor) {
        return thenProcess(new DecatonProcessorSupplierImpl<>(() -> processor, ProcessorScope.PROVIDED));
    }

-    Processors<T> build(DecatonProcessorSupplier<byte[]> retryProcessorSupplier) {
-        return new Processors<>(suppliers, retryProcessorSupplier, taskExtractor, retryTaskExtractor);
+    Processors<T> build(DecatonProcessorSupplier<byte[]> retryProcessorSupplier, ProcessorProperties properties) {


I thought that just taking an instance of taskExtractor as an argument of this method was ok, better to do it with "constructor" thing?

To instantiate TaskExtractor on build's caller (i.e. SubscriptionBuilder), ProcessorsBuilder needs to store Deserializer or TaskExtractor passed from user through consuming and SubscriptionBuilder need to instantiate appropriate resulting TaskExtractor accordingly, which is fairly complicated.

Maybe just deferring the instantiation by constructor is straightforward.

Ah, but storing constructor might be unnecessary actually.
Let me fix

kawamuray · 2024-08-15T09:58:15Z

...ssor/src/main/java/com/linecorp/decaton/processor/runtime/internal/DefaultTaskExtractor.java

+            //
+            // From Decaton perspective, there is no way to distinguish between these two cases,
+            // so we need to rely on a configuration to determine how to deserialize the task.
+            if (legacyFallbackEnabledProperty.value()) {


h-m should this be dynamically reloadable? i just sense that it might be better fixed for the whole lifetime of this subscription, but idk

IMO we should make property reloadable unless it's difficult or there is big downside, because it can reduce the number of rolling-restart which is a certain burden and enables quick rollback.

Since upgrade-procedure might be tough already, making reloadable might be better.

The downside I can imagine is the blast radius when property change caused a problem (because property change will happen on all instances at the same time), but quick-rollback by dynamic-reloading might be more preferred than rolling restart the application I guess.

kawamuray

LGTM

Make Decaton can consume any topic with deserializer

fd46ca1

ocadaruma added the breaking change Breaking change for a public API label Aug 15, 2024

ocadaruma requested a review from kawamuray August 15, 2024 02:39

ocadaruma commented Aug 15, 2024

View reviewed changes

fix

9b4e926

kawamuray requested changes Aug 15, 2024

View reviewed changes

ocadaruma added 2 commits August 15, 2024 16:44

get rid of workaround

cbff3c2

nits

b18ac27

ocadaruma requested a review from kawamuray August 15, 2024 07:52

kawamuray requested changes Aug 15, 2024

View reviewed changes

ocadaruma requested a review from kawamuray August 15, 2024 10:39

refactor

31b1b48

kawamuray approved these changes Aug 16, 2024

View reviewed changes

kawamuray merged commit d5dae11 into line:master Aug 16, 2024
5 checks passed

ocadaruma deleted the task-metadata-follow-up branch August 16, 2024 08:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Make Decaton can consume any topic with deserializer #241

Make Decaton can consume any topic with deserializer #241

ocadaruma commented Aug 15, 2024 •

edited

Loading

ocadaruma Aug 15, 2024

kawamuray Aug 15, 2024

kawamuray left a comment

kawamuray Aug 15, 2024

kawamuray Aug 15, 2024

kawamuray Aug 15, 2024

ocadaruma Aug 15, 2024

kawamuray Aug 15, 2024

ocadaruma Aug 15, 2024

ocadaruma Aug 15, 2024

kawamuray Aug 15, 2024

ocadaruma Aug 15, 2024

kawamuray left a comment

Make Decaton can consume any topic with deserializer #241

Make Decaton can consume any topic with deserializer #241

Conversation

ocadaruma commented Aug 15, 2024 • edited Loading

Motivation

Summary of changes

Expected upgrade procedure from <= 8.0.1 to 9.0.0

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kawamuray 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

Choose a reason for hiding this comment

kawamuray left a comment

Choose a reason for hiding this comment

ocadaruma commented Aug 15, 2024 •

edited

Loading