From 0fe70ccbc483195ee97440d4e1187c98e895f31c Mon Sep 17 00:00:00 2001 From: Yuta Kasai Date: Thu, 8 Oct 2026 18:05:30 +0900 Subject: [PATCH 1/5] NO-ISSUE Set auto.offset.reset to earliest by default --- .../decaton/benchmark/DecatonRunner.java | 2 +- .../decaton/processor/OffsetResetTest.java | 348 ++++++++++++++++++ .../runtime/internal/ConsumerSupplier.java | 13 +- .../internal/ConsumerSupplierTest.java | 43 +++ .../linecorp/decaton/testing/KafkaAdmin.java | 15 + .../linecorp/decaton/testing/TestUtils.java | 2 +- 6 files changed, 420 insertions(+), 3 deletions(-) create mode 100644 processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java create mode 100644 processor/src/test/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplierTest.java diff --git a/benchmark/src/main/java/com/linecorp/decaton/benchmark/DecatonRunner.java b/benchmark/src/main/java/com/linecorp/decaton/benchmark/DecatonRunner.java index cd5090ed..e3cd4671 100644 --- a/benchmark/src/main/java/com/linecorp/decaton/benchmark/DecatonRunner.java +++ b/benchmark/src/main/java/com/linecorp/decaton/benchmark/DecatonRunner.java @@ -78,7 +78,7 @@ public void init(Config config, Recording recording, ResourceTracker resourceTra // at that time the offset might not have initialized (reset) to 0. If we start producing tasks before // it completes fetching the current offset from the broker (which is zero), the offset counter might // increments before it obtains offset information and consequences to reset the offset to a larger - // value than zero with the default "latest" reset policy. + // value than zero with "latest" reset policy. props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); subPartitionRuntime = SubPartitionRuntime.THREAD_POOL; diff --git a/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java b/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java new file mode 100644 index 00000000..1e52056d --- /dev/null +++ b/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java @@ -0,0 +1,348 @@ +/* + * Copyright 2026 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package com.linecorp.decaton.processor; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import com.linecorp.decaton.processor.runtime.PerKeyQuotaConfig; +import com.linecorp.decaton.processor.runtime.ProcessorProperties; +import com.linecorp.decaton.processor.runtime.ProcessorSubscription; +import com.linecorp.decaton.processor.runtime.ProcessorsBuilder; +import com.linecorp.decaton.processor.runtime.Property; +import com.linecorp.decaton.processor.runtime.RetryConfig; +import com.linecorp.decaton.processor.runtime.StaticPropertySupplier; +import com.linecorp.decaton.processor.runtime.SubscriptionBuilder; +import com.linecorp.decaton.testing.KafkaClusterExtension; +import com.linecorp.decaton.testing.TestUtils; + +/** + * Verifies how partitions without committed offsets are consumed, + * comparing Decaton's default auto.offset.reset (earliest) with explicitly configured latest. + */ +public class OffsetResetTest { + @RegisterExtension + public static KafkaClusterExtension rule = new KafkaClusterExtension(); + + // All tasks share the same key so that they are processed in order + private static final String KEY = "key"; + + enum OffsetReset { + DEFAULT(null), + LATEST("latest"); + + private final String value; + + OffsetReset(String value) { + this.value = value; + } + + boolean readsFromBeginning() { + return this == DEFAULT; + } + } + + private final List topics = new ArrayList<>(); + private String topic; + private String groupId; + private Producer producer; + + @BeforeEach + public void setUp() { + topic = createTopic("test-" + UUID.randomUUID()); + groupId = "offset-reset-test-" + UUID.randomUUID(); + producer = TestUtils.producer(rule.bootstrapServers(), new StringSerializer(), new StringSerializer()); + } + + @AfterEach + public void tearDown() { + producer.close(); + rule.admin().deleteTopics(true, topics.toArray(new String[0])); + } + + @ParameterizedTest + @EnumSource(OffsetReset.class) + @Timeout(60) + public void testTaskProducedBeforeSubscriptionStarts(OffsetReset offsetReset) throws Exception { + // scenario: + // * a task is produced before the subscription of a new consumer group starts + // (e.g. producer is deployed before processor) + TopicPartition partition = new TopicPartition(topic, 0); + produce(partition, "produced-before-start"); + + Set processed = ConcurrentHashMap.newKeySet(); + try (ProcessorSubscription subscription = subscription(offsetReset, recording(processed))) { + produceMarkersUntilProcessed(partition, processed); + } + + assertProcessedOnlyIfReadsFromBeginning(offsetReset, processed, + Collections.singleton("produced-before-start")); + } + + @ParameterizedTest + @EnumSource(OffsetReset.class) + @Timeout(60) + public void testTaskProducedToAddedPartition(OffsetReset offsetReset) throws Exception { + // scenario: + // * the consumer group has committed offset for the existing partition + // * a partition is added, and a task is produced to it before the partition gets assigned + TopicPartition existingPartition = new TopicPartition(topic, 0); + Set processedBeforeIncrease = ConcurrentHashMap.newKeySet(); + try (ProcessorSubscription subscription = subscription(offsetReset, recording(processedBeforeIncrease))) { + long endOffset = produceMarkersUntilProcessed(existingPartition, processedBeforeIncrease); + awaitCommitted(existingPartition, endOffset); + } + + rule.admin().increasePartitions(topic, 2); + TopicPartition addedPartition = new TopicPartition(topic, 1); + produce(addedPartition, "produced-to-added-partition"); + + Set processed = ConcurrentHashMap.newKeySet(); + try (ProcessorSubscription subscription = subscription(offsetReset, recording(processed))) { + produceMarkersUntilProcessed(addedPartition, processed); + // If the existing partition were consumed from the beginning, + // tasks before the marker would be processed before it + produceMarkersUntilProcessed(existingPartition, processed); + } + + assertProcessedOnlyIfReadsFromBeginning(offsetReset, processed, + Collections.singleton("produced-to-added-partition")); + assertTrue(Collections.disjoint(processedBeforeIncrease, processed), + "partition with committed offset should be consumed from the committed offset"); + } + + @ParameterizedTest + @EnumSource(OffsetReset.class) + @Timeout(60) + public void testRetryTaskInFlightOnReassignment(OffsetReset offsetReset) throws Exception { + // scenario: + // * a task is retried, and the retry topic partition gets reassigned before the retried task completes + // (e.g. awaiting backoff at the shutdown) + // * the retry topic partition has no committed offset since no retried task has completed on it + String retryTopic = createTopic(topic + RetryConfig.DEFAULT_RETRY_TOPIC_SUFFIX); + TopicPartition originPartition = new TopicPartition(topic, 0); + TopicPartition retryPartition = new TopicPartition(retryTopic, 0); + RetryConfig retryConfig = RetryConfig.withBackoff(Duration.ofMillis(10)); + + Set processedByFirst = ConcurrentHashMap.newKeySet(); + Set inFlightOnFirst = ConcurrentHashMap.newKeySet(); + // The first subscription always uses the default so that it surely fetches the retried task + // regardless of when its position gets determined. + // What this test verifies is how the second subscription consumes the partition without committed offset. + try (ProcessorSubscription first = subscription( + OffsetReset.DEFAULT, + builder -> builder.enableRetry(retryConfig).processorsBuilder( + ProcessorsBuilder.consuming(topic, new StringDeserializer()) + .thenProcess((ctx, task) -> { + if (!"retried-task".equals(task)) { + processedByFirst.add(task); + } else if (ctx.metadata().retryCount() == 0) { + ctx.retry(); + } else { + // Never complete to keep the retried task in-flight + ctx.deferCompletion(); + inFlightOnFirst.add(task); + } + })))) { + produceMarkersUntilProcessed(originPartition, processedByFirst); + long offset = produce(originPartition, "retried-task"); + TestUtils.awaitCondition("retried task should be in-flight", + () -> inFlightOnFirst.contains("retried-task")); + awaitCommitted(originPartition, offset + 1); + } + assertFalse(committedOffsets().containsKey(retryPartition), + "retry topic partition should not have committed offset"); + + Set processedBySecond = ConcurrentHashMap.newKeySet(); + try (ProcessorSubscription second = subscription( + offsetReset, + recording(processedBySecond).andThen(builder -> builder.enableRetry(retryConfig)))) { + produceMarkersUntilProcessed(retryPartition, processedBySecond); + } + + assertProcessedOnlyIfReadsFromBeginning(offsetReset, processedBySecond, + Collections.singleton("retried-task")); + } + + @ParameterizedTest + @EnumSource(OffsetReset.class) + @Timeout(60) + public void testShapedTaskInFlightOnReassignment(OffsetReset offsetReset) throws Exception { + // scenario: + // * tasks of a bursting key are shaped, and the shaping topic partition gets reassigned + // before the shaped tasks complete + // * the shaping topic partition has no committed offset since no shaped task has completed on it + String shapingTopic = createTopic(topic + PerKeyQuotaConfig.DEFAULT_SHAPING_TOPIC_SUFFIX); + TopicPartition originPartition = new TopicPartition(topic, 0); + TopicPartition shapingPartition = new TopicPartition(shapingTopic, 0); + Set shapedTasks = ConcurrentHashMap.newKeySet(); + Consumer enablePerKeyQuota = builder -> builder + .addProperties(StaticPropertySupplier.of( + Property.ofStatic(ProcessorProperties.CONFIG_PER_KEY_QUOTA_PROCESSING_RATE, 1L))) + .enablePerKeyQuota(PerKeyQuotaConfig.shape() + .toBuilder() + .window(Duration.ofMillis(50L)) + .callback((record, metrics) -> { + shapedTasks.add(new String(record.value(), UTF_8)); + return shapingTopic; + }) + .build()); + + Set inFlightOnFirst = ConcurrentHashMap.newKeySet(); + // The first subscription always uses the default so that it surely fetches the shaped tasks + // regardless of when its position gets determined. + // What this test verifies is how the second subscription consumes the partition without committed offset. + try (ProcessorSubscription first = subscription( + OffsetReset.DEFAULT, + builder -> enablePerKeyQuota.accept(builder.processorsBuilder( + ProcessorsBuilder.consuming(topic, new StringDeserializer()) + .thenProcess((ctx, task) -> { + if (shapedTasks.contains(task)) { + // Never complete to keep the shaped task in-flight + ctx.deferCompletion(); + inFlightOnFirst.add(task); + } + }))))) { + long endOffset = 0; + for (int i = 0; shapedTasks.isEmpty(); i++) { + endOffset = produce(originPartition, "task-" + i) + 1; + } + awaitCommitted(originPartition, endOffset); + TestUtils.awaitCondition("shaped tasks should be in-flight", + () -> inFlightOnFirst.containsAll(shapedTasks)); + } + Set shapedByFirst = new HashSet<>(shapedTasks); + assertFalse(committedOffsets().containsKey(shapingPartition), + "shaping topic partition should not have committed offset"); + + Set processedBySecond = ConcurrentHashMap.newKeySet(); + try (ProcessorSubscription second = subscription( + offsetReset, + recording(processedBySecond).andThen(enablePerKeyQuota))) { + produceMarkersUntilProcessed(shapingPartition, processedBySecond); + } + + assertProcessedOnlyIfReadsFromBeginning(offsetReset, processedBySecond, shapedByFirst); + } + + private static void assertProcessedOnlyIfReadsFromBeginning(OffsetReset offsetReset, + Set processed, + Collection tasks) { + if (offsetReset.readsFromBeginning()) { + assertTrue(processed.containsAll(tasks), "tasks should be processed: " + tasks); + } else { + assertTrue(Collections.disjoint(processed, tasks), "tasks should be skipped: " + tasks); + } + } + + private String createTopic(String name) { + rule.admin().createTopic(name, 1, 3); + topics.add(name); + return name; + } + + private ProcessorSubscription subscription(OffsetReset offsetReset, + Consumer builderConfigurer) + throws Exception { + Properties consumerConfig = new Properties(); + consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId); + // Refresh metadata frequently so that newly created topics and added partitions get assigned soon + // even if the subscription sees stale metadata right after they are created + consumerConfig.setProperty(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000"); + if (offsetReset.value != null) { + consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset.value); + } + return TestUtils.subscription("subscription-" + UUID.randomUUID(), + rule.bootstrapServers(), + builderConfigurer, + consumerConfig); + } + + private Consumer recording(Set processed) { + return builder -> builder.processorsBuilder( + ProcessorsBuilder.consuming(topic, new StringDeserializer()) + .thenProcess((ctx, task) -> processed.add(task))); + } + + private long produce(TopicPartition partition, String task) throws Exception { + return producer.send(new ProducerRecord<>(partition.topic(), partition.partition(), KEY, task)) + .get() + .offset(); + } + + /** + * Produce markers until one of them is processed. + * The position of a partition without committed offset is determined after the assignment asynchronously, + * so a marker produced before that could be skipped by latest. + * Since tasks of the same key are processed in order, all tasks before the processed marker + * have been processed (or skipped) when this method returns. + * @return the end offset of the partition + */ + private long produceMarkersUntilProcessed(TopicPartition partition, Set processed) throws Exception { + while (true) { + String marker = "marker-" + UUID.randomUUID(); + long offset = produce(partition, marker); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (System.nanoTime() < deadline) { + if (processed.contains(marker)) { + return offset + 1; + } + Thread.sleep(50L); + } + } + } + + private Map committedOffsets() { + return rule.admin().consumerGroupOffsets(groupId); + } + + private void awaitCommitted(TopicPartition partition, long offset) throws InterruptedException { + TestUtils.awaitCondition("offset " + offset + " should be committed for " + partition, () -> { + OffsetAndMetadata committed = committedOffsets().get(partition); + return committed != null && committed.offset() == offset; + }); + } +} diff --git a/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java b/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java index 65b171c2..62411df2 100644 --- a/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java +++ b/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java @@ -45,7 +45,8 @@ public Consumer get() { return new KafkaConsumer<>(mergedProps(), new ByteArrayDeserializer(), new ByteArrayDeserializer()); } - private Properties mergedProps() { + // visible for testing + Properties mergedProps() { Properties props = new Properties(); for (String key : config.stringPropertyNames()) { props.setProperty(key, config.getProperty(key)); @@ -60,6 +61,16 @@ private Properties mergedProps() { String.valueOf(DEFAULT_MAX_POLL_RECORDS)); } + // auto.offset.reset handling to avoid skipping records silently. + // Decaton commits offsets only for completed records, so a partition where no record has completed yet has no + // committed offset (e.g. newly added partitions, retry/shaping topic partitions which rarely receive records). + // With "latest", records produced to such a partition before the consumer determines its position + // (e.g. before the partition gets assigned or reassigned) are skipped. + // See https://github.com/line/decaton/blob/master/docs/auto-offset-reset.adoc for details. + if (props.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG) == null) { + props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + } + return props; } } diff --git a/processor/src/test/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplierTest.java b/processor/src/test/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplierTest.java new file mode 100644 index 00000000..036a8f99 --- /dev/null +++ b/processor/src/test/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplierTest.java @@ -0,0 +1,43 @@ +/* + * Copyright 2026 LY Corporation + * + * LY Corporation licenses this file to you under the Apache License, + * version 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package com.linecorp.decaton.processor.runtime.internal; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.Properties; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.junit.jupiter.api.Test; + +public class ConsumerSupplierTest { + @Test + public void testAutoOffsetResetDefaultsToEarliest() { + Properties props = new ConsumerSupplier(new Properties()).mergedProps(); + + assertEquals("earliest", props.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)); + } + + @Test + public void testAutoOffsetResetCanBeOverridden() { + Properties config = new Properties(); + config.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); + + Properties props = new ConsumerSupplier(config).mergedProps(); + + assertEquals("latest", props.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)); + } +} diff --git a/testing/src/main/java/com/linecorp/decaton/testing/KafkaAdmin.java b/testing/src/main/java/com/linecorp/decaton/testing/KafkaAdmin.java index 3d579562..7ba7c7e7 100644 --- a/testing/src/main/java/com/linecorp/decaton/testing/KafkaAdmin.java +++ b/testing/src/main/java/com/linecorp/decaton/testing/KafkaAdmin.java @@ -24,6 +24,7 @@ import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.NewPartitions; import org.apache.kafka.clients.admin.NewTopic; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; @@ -65,6 +66,20 @@ public void createTopic(String topicName, int numPartitions, int replicationFact } } + /** + * Increase the number of partitions of the topic + * @param topicName topic to add partitions + * @param totalCount total number of partitions after the increase + */ + public void increasePartitions(String topicName, int totalCount) { + try { + adminClient.createPartitions(Collections.singletonMap(topicName, NewPartitions.increaseTo(totalCount))) + .all().get(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + public void deleteTopics(boolean ignoreFailures, String... topicNames) { try { adminClient.deleteTopics(Arrays.asList(topicNames)).all().get(); diff --git a/testing/src/main/java/com/linecorp/decaton/testing/TestUtils.java b/testing/src/main/java/com/linecorp/decaton/testing/TestUtils.java index 646b7694..7c9d16b7 100644 --- a/testing/src/main/java/com/linecorp/decaton/testing/TestUtils.java +++ b/testing/src/main/java/com/linecorp/decaton/testing/TestUtils.java @@ -185,7 +185,7 @@ public SubscriptionBuilder stateListener(SubscriptionStateListener stateListener props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "test-" + subscriptionId); props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID); - props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + // Don't set auto.offset.reset here so that tests run with Decaton's default if (additionalConsumerConfig != null) { props.putAll(additionalConsumerConfig); From 6d1004830667e465df84e2b7e719bbfeeee0ec9b Mon Sep 17 00:00:00 2001 From: Yuta Kasai Date: Thu, 8 Oct 2026 18:05:30 +0900 Subject: [PATCH 2/5] NO-ISSUE Update docs for 10.0.0 --- docs/auto-offset-reset.adoc | 109 +++++++++++++++++++++++ docs/dynamic-property-configuration.adoc | 2 +- docs/getting-started.adoc | 4 +- docs/index.adoc | 3 +- docs/key-blocking.adoc | 2 +- docs/monitoring.adoc | 2 +- docs/rate-limiting.adoc | 4 +- docs/retry-queueing.adoc | 9 +- docs/runtime.adoc | 2 +- docs/spring-integration.adoc | 2 +- docs/task-batching.adoc | 2 +- docs/task-compaction.adoc | 2 +- docs/task-extractor.adoc | 2 +- docs/tracing.adoc | 2 +- docs/why-decaton.adoc | 2 +- 15 files changed, 134 insertions(+), 15 deletions(-) create mode 100644 docs/auto-offset-reset.adoc diff --git a/docs/auto-offset-reset.adoc b/docs/auto-offset-reset.adoc new file mode 100644 index 00000000..ebc49464 --- /dev/null +++ b/docs/auto-offset-reset.adoc @@ -0,0 +1,109 @@ += Where to Start Consuming (`auto.offset.reset`) +:base_version: 10.0.0 +:modules: processor + +== Introduction + +Since Decaton 10.0.0, Decaton sets `auto.offset.reset` to `earliest` by default. +Kafka's default is `latest`. + +`auto.offset.reset` decides where a processor starts consuming a partition when the partition has no committed offset, or when the offset to resume from no longer exists in the partition (e.g. deleted by retention before being processed). +See https://kafka.apache.org/39/configuration/consumer-configs/#consumerconfigs_auto.offset.reset[Kafka documentation] for details. + +Kafka consumers with `latest` can skip tasks silently in the scenarios below. +Decaton uses `earliest` by default because it is safer for at-least-once delivery. + +== Why `earliest` + +=== Adding partitions + +Producers can start producing tasks to new partitions before processors get them assigned. +With `latest`, tasks produced in between are skipped. + +=== Starting to use a topic + +When you deploy a producer and a processor of a new consumer group (or change `group.id`), tasks produced before the processor starts are skipped with `latest`. + +=== Restarting after committed offsets expired + +If all processors of a consumer group are stopped longer than the offset retention of your Kafka cluster (`offsets.retention.minutes`, 7 days by default), the committed offsets expire. +With `latest`, tasks produced while the processors were stopped are skipped after restarting. + +=== Retry and per-key quota + +With link:./retry-queueing.adoc[retry queueing] or per-key quota, Decaton produces tasks to the retry topic or shaping topics. +These topics receive tasks only when tasks are retried or shaped, and Decaton commits offsets only for processed tasks, so their partitions can stay without committed offsets for a long time. +For example, a partition of the retry topic has no committed offset until a retried task on it is processed for the first time. + +If such a partition is reassigned before the task on it is processed, the task is skipped with `latest`. +This typically happens during a deployment: a task is retried, and the rebalance reassigns the partition while the retried task is waiting for its backoff. + +== Trade-off + +With `earliest`, tasks in all of the scenarios above are processed. +In exchange, processors may process old tasks: + +* A processor of a new consumer group processes all tasks remaining in the topic. +* After committed offsets expired, tasks remaining in the topic are processed again. +* When processors lag behind or are stopped longer than the retention of the topic, they process the remaining old tasks instead of jumping to the end of the partition. + +[#options] +== Options if `earliest` doesn't fit + +If your processor shouldn't process old tasks (e.g. it sends notifications), you can use options such as the following. +Each has its own risk. + +=== Use `latest` + +Set `auto.offset.reset` in `consumerConfig`: + +[source,java] +---- +Properties consumerConfig = new Properties(); +... +consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); + +ProcessorSubscription subscription = + SubscriptionBuilder.newBuilder("my-decaton-processor") + ... + .consumerConfig(consumerConfig) + .buildAndStart(); +---- + +The risk is that tasks are skipped silently in the scenarios above. + +If you use `latest` only temporarily (e.g. for the first deployment), partitions on which no task has been processed in the meantime still have no committed offsets after you remove it, and are consumed from the beginning. + +=== Prepare committed offsets before starting processors + +Keep `earliest`, and create committed offsets for the new consumer group with `kafka-consumer-groups.sh` (the CLI tool of Apache Kafka) before starting processors. +For example: + +[source,sh] +---- +kafka-consumer-groups.sh \ + --bootstrap-server $BOOTSTRAP_SERVERS \ + --group $GROUP_ID \ + --topic $TOPIC \ + --reset-offsets \ + --to-latest \ + --execute +---- + +Specify only the origin topic, so that the retry topic and shaping topics are consumed from the beginning. +See https://kafka.apache.org/39/operations/basic-kafka-operations/#managing-consumer-groups[Kafka documentation] for details of `kafka-consumer-groups.sh`. + +=== Discard old tasks in the processor + +Keep `earliest`, and discard old tasks in your processor based on the timestamp of the task. + +`context.metadata().timestampMillis()` is: + +* the time when the task was produced, if it was produced by DecatonClient (unless you set it explicitly) +* the timestamp of the Kafka record, if it wasn't produced by DecatonClient +* the value set by the extractor, if you use a custom `TaskExtractor` (0 if not set) + +You can also use a timestamp in the headers or the body of the task, if any. + +A retried task keeps the timestamp of its original task, so the threshold should be longer than the total period of retries. +A task that wasn't produced by DecatonClient gets a new timestamp when it's shaped (the time when it was produced to the shaping topic). diff --git a/docs/dynamic-property-configuration.adoc b/docs/dynamic-property-configuration.adoc index 7bd40afd..ba552b51 100644 --- a/docs/dynamic-property-configuration.adoc +++ b/docs/dynamic-property-configuration.adoc @@ -1,5 +1,5 @@ = Dynamic Property Configuration -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: centraldogma,processor :toc: diff --git a/docs/getting-started.adoc b/docs/getting-started.adoc index c1b71407..bbaf0905 100644 --- a/docs/getting-started.adoc +++ b/docs/getting-started.adoc @@ -1,5 +1,5 @@ = Getting Started Decaton -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: common,client,processor,protobuf Let's start from the most basic usage of Decaton client/processor. @@ -163,6 +163,8 @@ public final class ProcessorMain { ---- <1> `subscriptionId` is Decaton's internal-use only ID for metric labeling, logging and so on as of current version. +NOTE: Decaton sets `auto.offset.reset` to `earliest` unless you set it in `consumerConfig`. See link:./auto-offset-reset.adoc[Where to Start Consuming] for the details. + == Test it === Preparing topic diff --git a/docs/index.adoc b/docs/index.adoc index c6dd8152..e2ab3329 100644 --- a/docs/index.adoc +++ b/docs/index.adoc @@ -1,5 +1,5 @@ = Decaton Documents -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: - link:./why-decaton.adoc[Why Decaton] @@ -10,6 +10,7 @@ - link:./task-extractor.adoc[Implement custom task extractor] - link:./dynamic-property-configuration.adoc[Dynamic property configuration for the processor] - link:./monitoring.adoc[Monitoring Decaton] +- link:./auto-offset-reset.adoc[Where to Start Consuming (`auto.offset.reset`)] - Features * link:./retry-queueing.adoc[Retry Queueing] * link:./rate-limiting.adoc[Rate Limiting] diff --git a/docs/key-blocking.adoc b/docs/key-blocking.adoc index 9904d90e..d1aa8da7 100644 --- a/docs/key-blocking.adoc +++ b/docs/key-blocking.adoc @@ -1,5 +1,5 @@ = Key Blocking -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor == Introduction diff --git a/docs/monitoring.adoc b/docs/monitoring.adoc index 96f18259..24aef1f4 100644 --- a/docs/monitoring.adoc +++ b/docs/monitoring.adoc @@ -1,5 +1,5 @@ = Monitoring Decaton -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor This document guides you how to monitor your Decaton processor applications. diff --git a/docs/rate-limiting.adoc b/docs/rate-limiting.adoc index 1f17470b..33860200 100644 --- a/docs/rate-limiting.adoc +++ b/docs/rate-limiting.adoc @@ -1,5 +1,5 @@ = Rate Limiting -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor == Introduction @@ -32,7 +32,7 @@ The value can have three different meanings: |-1 (default) |Unlimited -|Any positive number +|1 to 1,000,000 |Do the best to process tasks with the given value per second. Note that the actual processing rate may not reach the given value if a task takes over a second to process or the value is greater than actual throughput per second. |=== diff --git a/docs/retry-queueing.adoc b/docs/retry-queueing.adoc index 77cf86b0..748f7c18 100644 --- a/docs/retry-queueing.adoc +++ b/docs/retry-queueing.adoc @@ -1,5 +1,5 @@ = Retry Queuing -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor == Introduction @@ -61,6 +61,13 @@ SubscriptionBuilder.newBuilder("testProcessor") If you are using distributed tracing and want to see retries as part of the same trace, make sure that RetryConfig contains a KafkaProducerSupplier that will propagate trace information. +[NOTE] +==== +Retry topic partitions tend to stay without committed offsets for a long time since they receive tasks only when tasks are retried. +If you set `auto.offset.reset` to `latest` explicitly, retry tasks may be skipped silently on rebalance. +See link:./auto-offset-reset.adoc[Where to Start Consuming] for the details. +==== + === Retry with synchronous processing After enabling the retry function, you can use it in your processor: diff --git a/docs/runtime.adoc b/docs/runtime.adoc index 90158d30..76851e7d 100644 --- a/docs/runtime.adoc +++ b/docs/runtime.adoc @@ -1,5 +1,5 @@ = Subpartition Runtime -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor This document guides you what Subpartition Runtime is and how to use it. diff --git a/docs/spring-integration.adoc b/docs/spring-integration.adoc index 64546b5a..07f1ff48 100644 --- a/docs/spring-integration.adoc +++ b/docs/spring-integration.adoc @@ -1,5 +1,5 @@ = Spring Integration -:base_version: 0.0.43 +:base_version: 10.0.0 :modules: TBC diff --git a/docs/task-batching.adoc b/docs/task-batching.adoc index 918002d6..af068ce4 100644 --- a/docs/task-batching.adoc +++ b/docs/task-batching.adoc @@ -1,5 +1,5 @@ = Task Batching -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor == Introduction diff --git a/docs/task-compaction.adoc b/docs/task-compaction.adoc index 0a7a308c..f1f4ab8e 100644 --- a/docs/task-compaction.adoc +++ b/docs/task-compaction.adoc @@ -1,5 +1,5 @@ = Task Compaction -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor == Introduction diff --git a/docs/task-extractor.adoc b/docs/task-extractor.adoc index f94267ea..0ad2a55f 100644 --- a/docs/task-extractor.adoc +++ b/docs/task-extractor.adoc @@ -1,5 +1,5 @@ = TaskExtractor -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: common,protocol,processor [NOTE] diff --git a/docs/tracing.adoc b/docs/tracing.adoc index adf60c08..bba55f90 100644 --- a/docs/tracing.adoc +++ b/docs/tracing.adoc @@ -1,5 +1,5 @@ = Tracing -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: brave,processor Decaton can integrate with distributed tracing frameworks so that you can associate the processing of a message diff --git a/docs/why-decaton.adoc b/docs/why-decaton.adoc index acf84574..079844e2 100644 --- a/docs/why-decaton.adoc +++ b/docs/why-decaton.adoc @@ -1,5 +1,5 @@ = Why Decaton -:base_version: 9.0.0 +:base_version: 10.0.0 :modules: processor This document explains why we have decided to create a new consumer framework. From 63eac14cce740034487f3839270bdbc928baa9b7 Mon Sep 17 00:00:00 2001 From: Yuta Kasai Date: Thu, 8 Oct 2026 19:00:59 +0900 Subject: [PATCH 3/5] NO-ISSUE avoid flaky test --- README.md | 1 + .../decaton/processor/OffsetResetTest.java | 24 +++++++++++-------- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 0c77a6dc..164ec2b7 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,7 @@ Below are some examples. See [Index](./docs/index.adoc) for the full list. * Retry Queuing - Retry a failed task with back-off without blocking other tasks flow * Dynamic Rate Limiting - Apply and update processing rate quota dynamically * Task Compaction - Crush preceding tasks which its processing results will be overwritten by following task +* Per-Key Quota - Detect bursting keys and isolate their tasks into shaping topics so that they don't dominate processing of other keys # Performance diff --git a/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java b/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java index 1e52056d..82d8b69f 100644 --- a/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java +++ b/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java @@ -56,6 +56,7 @@ import com.linecorp.decaton.processor.runtime.RetryConfig; import com.linecorp.decaton.processor.runtime.StaticPropertySupplier; import com.linecorp.decaton.processor.runtime.SubscriptionBuilder; +import com.linecorp.decaton.processor.runtime.internal.RateLimiter; import com.linecorp.decaton.testing.KafkaClusterExtension; import com.linecorp.decaton.testing.TestUtils; @@ -143,8 +144,8 @@ public void testTaskProducedToAddedPartition(OffsetReset offsetReset) throws Exc Set processed = ConcurrentHashMap.newKeySet(); try (ProcessorSubscription subscription = subscription(offsetReset, recording(processed))) { produceMarkersUntilProcessed(addedPartition, processed); - // If the existing partition were consumed from the beginning, - // tasks before the marker would be processed before it + // If the existing partition were consumed from the beginning, its old tasks would be processed + // before this marker produceMarkersUntilProcessed(existingPartition, processed); } @@ -160,7 +161,7 @@ public void testTaskProducedToAddedPartition(OffsetReset offsetReset) throws Exc public void testRetryTaskInFlightOnReassignment(OffsetReset offsetReset) throws Exception { // scenario: // * a task is retried, and the retry topic partition gets reassigned before the retried task completes - // (e.g. awaiting backoff at the shutdown) + // (e.g. awaiting backoff during shutdown) // * the retry topic partition has no committed offset since no retried task has completed on it String retryTopic = createTopic(topic + RetryConfig.DEFAULT_RETRY_TOPIC_SUFFIX); TopicPartition originPartition = new TopicPartition(topic, 0); @@ -170,8 +171,7 @@ public void testRetryTaskInFlightOnReassignment(OffsetReset offsetReset) throws Set processedByFirst = ConcurrentHashMap.newKeySet(); Set inFlightOnFirst = ConcurrentHashMap.newKeySet(); // The first subscription always uses the default so that it surely fetches the retried task - // regardless of when its position gets determined. - // What this test verifies is how the second subscription consumes the partition without committed offset. + // regardless of when its position is determined. Only the second one uses the offsetReset under test. try (ProcessorSubscription first = subscription( OffsetReset.DEFAULT, builder -> builder.enableRetry(retryConfig).processorsBuilder( @@ -229,12 +229,16 @@ public void testShapedTaskInFlightOnReassignment(OffsetReset offsetReset) throws shapedTasks.add(new String(record.value(), UTF_8)); return shapingTopic; }) - .build()); + .build()) + // The shaping topic is processed at the per-key quota rate (1 task/sec) unless overridden, + // which can't catch up with produceMarkersUntilProcessed producing a marker every second + .overrideShapingRate(shapingTopic, StaticPropertySupplier.of( + Property.ofStatic(PerKeyQuotaConfig.shapingRateProperty(shapingTopic), RateLimiter.UNLIMITED)) + ); Set inFlightOnFirst = ConcurrentHashMap.newKeySet(); // The first subscription always uses the default so that it surely fetches the shaped tasks - // regardless of when its position gets determined. - // What this test verifies is how the second subscription consumes the partition without committed offset. + // regardless of when its position is determined. Only the second one uses the offsetReset under test. try (ProcessorSubscription first = subscription( OffsetReset.DEFAULT, builder -> enablePerKeyQuota.accept(builder.processorsBuilder( @@ -315,8 +319,8 @@ private long produce(TopicPartition partition, String task) throws Exception { /** * Produce markers until one of them is processed. - * The position of a partition without committed offset is determined after the assignment asynchronously, - * so a marker produced before that could be skipped by latest. + * The position of a partition without committed offset is determined asynchronously after the assignment, + * so a marker produced before that could be skipped with latest. * Since tasks of the same key are processed in order, all tasks before the processed marker * have been processed (or skipped) when this method returns. * @return the end offset of the partition From 97305e06a5044694f3f46449e904b342384adc87 Mon Sep 17 00:00:00 2001 From: Yuta Kasai Date: Mon, 19 Oct 2026 14:39:15 +0900 Subject: [PATCH 4/5] NO-ISSUE reference, rephrase --- docs/auto-offset-reset.adoc | 13 ++++++++----- .../linecorp/decaton/processor/OffsetResetTest.java | 9 ++++----- .../runtime/internal/ConsumerSupplier.java | 2 +- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/docs/auto-offset-reset.adoc b/docs/auto-offset-reset.adoc index ebc49464..47dc362b 100644 --- a/docs/auto-offset-reset.adoc +++ b/docs/auto-offset-reset.adoc @@ -5,7 +5,7 @@ == Introduction Since Decaton 10.0.0, Decaton sets `auto.offset.reset` to `earliest` by default. -Kafka's default is `latest`. +Kafka's default is `latest` as of Apache Kafka 4.3. `auto.offset.reset` decides where a processor starts consuming a partition when the partition has no committed offset, or when the offset to resume from no longer exists in the partition (e.g. deleted by retention before being processed). See https://kafka.apache.org/39/configuration/consumer-configs/#consumerconfigs_auto.offset.reset[Kafka documentation] for details. @@ -20,13 +20,15 @@ Decaton uses `earliest` by default because it is safer for at-least-once deliver Producers can start producing tasks to new partitions before processors get them assigned. With `latest`, tasks produced in between are skipped. +To understand this deeply, you can read https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406619800/KIP-1282+Prevent+data+loss+during+partition+expansion+for+dynamically+added+partitions[KIP-1282: Prevent data loss during partition expansion for dynamically added partitions] + === Starting to use a topic When you deploy a producer and a processor of a new consumer group (or change `group.id`), tasks produced before the processor starts are skipped with `latest`. === Restarting after committed offsets expired -If all processors of a consumer group are stopped longer than the offset retention of your Kafka cluster (`offsets.retention.minutes`, 7 days by default), the committed offsets expire. +If all processors of a consumer group are stopped longer than the offset retention of your Kafka cluster (`offsets.retention.minutes`, 7 days by default as of Apache Kafka 4.3), the committed offsets expire. With `latest`, tasks produced while the processors were stopped are skipped after restarting. === Retry and per-key quota @@ -35,8 +37,9 @@ With link:./retry-queueing.adoc[retry queueing] or per-key quota, Decaton produc These topics receive tasks only when tasks are retried or shaped, and Decaton commits offsets only for processed tasks, so their partitions can stay without committed offsets for a long time. For example, a partition of the retry topic has no committed offset until a retried task on it is processed for the first time. -If such a partition is reassigned before the task on it is processed, the task is skipped with `latest`. -This typically happens during a deployment: a task is retried, and the rebalance reassigns the partition while the retried task is waiting for its backoff. +With `latest`, a retried or shaped task on such a partition is skipped if a rebalance happens before the task is processed. +Tasks are often retried during a rebalance because processing can become unstable temporarily (e.g. during a deployment). +A retried task also waits for its backoff before being processed, so this can lead to subtle delivery loss. == Trade-off @@ -53,7 +56,7 @@ In exchange, processors may process old tasks: If your processor shouldn't process old tasks (e.g. it sends notifications), you can use options such as the following. Each has its own risk. -=== Use `latest` +=== Use `latest` or other options Set `auto.offset.reset` in `consumerConfig`: diff --git a/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java b/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java index 82d8b69f..217825bd 100644 --- a/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java +++ b/processor/src/it/java/com/linecorp/decaton/processor/OffsetResetTest.java @@ -158,9 +158,9 @@ public void testTaskProducedToAddedPartition(OffsetReset offsetReset) throws Exc @ParameterizedTest @EnumSource(OffsetReset.class) @Timeout(60) - public void testRetryTaskInFlightOnReassignment(OffsetReset offsetReset) throws Exception { + public void testRetryTaskInFlightOnRebalance(OffsetReset offsetReset) throws Exception { // scenario: - // * a task is retried, and the retry topic partition gets reassigned before the retried task completes + // * a task is retried, and a rebalance happens before the retried task completes // (e.g. awaiting backoff during shutdown) // * the retry topic partition has no committed offset since no retried task has completed on it String retryTopic = createTopic(topic + RetryConfig.DEFAULT_RETRY_TOPIC_SUFFIX); @@ -210,10 +210,9 @@ public void testRetryTaskInFlightOnReassignment(OffsetReset offsetReset) throws @ParameterizedTest @EnumSource(OffsetReset.class) @Timeout(60) - public void testShapedTaskInFlightOnReassignment(OffsetReset offsetReset) throws Exception { + public void testShapedTaskInFlightOnRebalance(OffsetReset offsetReset) throws Exception { // scenario: - // * tasks of a bursting key are shaped, and the shaping topic partition gets reassigned - // before the shaped tasks complete + // * tasks of a bursting key are shaped, and a rebalance happens before the shaped tasks complete // * the shaping topic partition has no committed offset since no shaped task has completed on it String shapingTopic = createTopic(topic + PerKeyQuotaConfig.DEFAULT_SHAPING_TOPIC_SUFFIX); TopicPartition originPartition = new TopicPartition(topic, 0); diff --git a/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java b/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java index 62411df2..2fb7f62a 100644 --- a/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java +++ b/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ConsumerSupplier.java @@ -65,7 +65,7 @@ Properties mergedProps() { // Decaton commits offsets only for completed records, so a partition where no record has completed yet has no // committed offset (e.g. newly added partitions, retry/shaping topic partitions which rarely receive records). // With "latest", records produced to such a partition before the consumer determines its position - // (e.g. before the partition gets assigned or reassigned) are skipped. + // (e.g. before the partition gets assigned, initially or by a rebalance) are skipped. // See https://github.com/line/decaton/blob/master/docs/auto-offset-reset.adoc for details. if (props.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG) == null) { props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); From 692ab7232b101931c42dbca9c92f532b6bd01bda Mon Sep 17 00:00:00 2001 From: Yuta Kasai Date: Mon, 19 Oct 2026 14:50:04 +0900 Subject: [PATCH 5/5] NO-ISSUE refine sentence --- docs/auto-offset-reset.adoc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/auto-offset-reset.adoc b/docs/auto-offset-reset.adoc index 47dc362b..ce361bc3 100644 --- a/docs/auto-offset-reset.adoc +++ b/docs/auto-offset-reset.adoc @@ -20,7 +20,7 @@ Decaton uses `earliest` by default because it is safer for at-least-once deliver Producers can start producing tasks to new partitions before processors get them assigned. With `latest`, tasks produced in between are skipped. -To understand this deeply, you can read https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406619800/KIP-1282+Prevent+data+loss+during+partition+expansion+for+dynamically+added+partitions[KIP-1282: Prevent data loss during partition expansion for dynamically added partitions] +To understand this in detail, you can read https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406619800/KIP-1282+Prevent+data+loss+during+partition+expansion+for+dynamically+added+partitions[KIP-1282: Prevent data loss during partition expansion for dynamically added partitions]. === Starting to use a topic