Set auto.offset.reset=earliest by default not to lost messages

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
 

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;

docs/auto-offset-reset.adoc

@@ -0,0 +1,112 @@
+= 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` 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.
+
+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.
+
+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
+
+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 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
+
+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.
+
+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
+
+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` or other options
+
+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).

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:
 

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

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]

docs/key-blocking.adoc

@@ -1,5 +1,5 @@
 = Key Blocking
-:base_version: 9.0.0
+:base_version: 10.0.0
 :modules: processor
 
 == Introduction

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.

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.
 |===

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:
 

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.

docs/spring-integration.adoc

@@ -1,5 +1,5 @@
 = Spring Integration
-:base_version: 0.0.43
+:base_version: 10.0.0
 :modules:
 
 TBC

docs/task-batching.adoc

@@ -1,5 +1,5 @@
 = Task Batching
-:base_version: 9.0.0
+:base_version: 10.0.0
 :modules: processor
 
 == Introduction

docs/task-compaction.adoc

@@ -1,5 +1,5 @@
 = Task Compaction
-:base_version: 9.0.0
+:base_version: 10.0.0
 :modules: processor
 
 == Introduction

docs/task-extractor.adoc

@@ -1,5 +1,5 @@
 = TaskExtractor
-:base_version: 9.0.0
+:base_version: 10.0.0
 :modules: common,protocol,processor
 
 [NOTE]

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

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.