Status

Current state: Withdraw

Discussion thread: https://lists.apache.org/thread/bp4zk31zr1sdxjsspg7b7bqddmm9t4gn

Vote Thread: https://lists.apache.org/thread/dgs3t9xmmldof5mmhwp21rpl3mfvpfw1

JIRA: KAFKA-20539 - Getting issue details... STATUS

Motivation

When a Kafka topic is expanded with new partitions, consumers configured with auto.offset.reset=latest will silently miss every record produced to those new partitions. This data loss occurs during the metadata blindness window, the gap between when the partition is created on the broker and when the consumer discovers it during its next periodic metadata refresh. No exception is thrown, no warning is emitted, and no log entry is generated.

Partition expansion is a routine operational procedure, not an exceptional event. Topics are typically created with a conservative partition count and scaled up as throughput grows; nearly every long-lived, high-volume topic will undergo expansion at some point in its lifecycle. Under the current behavior, each such expansion is a silent data-loss opportunity for auto.offset.reset=latest consumers — making data loss an expected outcome of a common operational workflow, not an edge case.

The root cause is that auto.offset.reset is applied uniformly to every partition without a committed offset, regardless of when that partition was created relative to the consumer group. From the user's perspective, however, two cases are fundamentally different:

• Partitions that existed before the consumer group was created. The choice of auto.offset.reset reflects an informed decision about how to handle the historical backlog that predates the group. Selecting latest is an explicit acknowledgment that pre-group history will not be consumed.
• Partitions added to the topic after the consumer group was created. Every record on such a partition was produced during the group's active lifetime. Skipping these records with latest is not an informed choice — it is silent data loss in a window where the user expects continuous, gap-free consumption.

Today there is no way to express different reset behavior for these two cases. None of latest, earliest, none, or by_duration allows scoping the policy to one category, and application-level workarounds such as custom ConsumerRebalanceListener.onPartitionsAssigned() logic lack the temporal metadata needed to distinguish them reliably.

This KIP introduces a complementary configuration, auto.offset.reset.new.partitions, that decouples the reset policy for partitions added after group creation from the policy for partitions that predate it. The new configuration accepts the same value space as auto.offset.reset (earliest, latest, by_duration:<duration>), enabling users to express their actual intent. For example, combining auto.offset.reset=latest with auto.offset.reset.new.partitions=earliest ensures that pre-group history is skipped while no record on a newly expanded partition is silently lost.

Classification is deterministic and server-side: the broker compares the group's creation timestamp (recorded once and propagated via ConsumerGroupHeartbeatResponse) against the partition's creation time. Partitions created after the group are classified as newly expanded. This eliminates fragile client-side heuristics relying on metadata refresh intervals.

Why Existing Alternatives are Insufficient

Custom ConsumerRebalanceListener.onPartitionsAssigned() logic

Clients currently have no way to determine when a partition was created, partition creation time is entirely absent from MetadataResponse, a protocol gap this KIP closes. Without that timestamp, user code must resort to heuristics such as treating any previously unseen partition ID as newly expanded. These heuristics break under ordinary operational conditions, consumer restarts, group migrations, partition reassignments, and topic re-subscriptions, surfacing pre-existing partitions as "new" and triggering exactly the kind of massive historical reprocessing we want to avoid. Preventing silent data loss on partition expansion is correctness-critical behavior and belongs in the core client, not reimplemented by every application.

by_duration

The existing by_duration policy may appear to address partition-expansion data loss, but has three limitations that make it unsuitable:

1. The seek target is computed client-side as now() - duration, introducing clock skew across consumers and forcing operators to choose large durations at the cost of unnecessary reprocessing.
2. The target timestamp is recomputed on each retry, so failed ListOffsetsRequest retries can shift the target forward, causing records produced between attempts to be missed.
3. It applies uniformly to all partitions missing a committed offset, failing to distinguish genuinely new partitions from pre-existing ones newly assigned to the group, which forces unnecessary replay

auto.offset.reset.new.partitions combined with the group-creation-time classifier addresses all three. Classification is derived from two server-recorded timestamps partition.creationTime (set once at partition creation) and group.creationTime (set once at group creation), eliminating any dependency on the consumer's clock. Because both timestamps are fixed at creation and never updated, classification is deterministic and stable across retries. And because classification is per-partition, only genuinely new partitions are affected; pre-existing partitions newly assigned to the group continue to follow the base auto.offset.reset policy.

Even users who choose auto.offset.reset=by_duration as their base policy benefit from this decoupling: setting auto.offset.reset.new.partitions to an explicit value (e.g., earliest) confines the by_duration seek-back semantics to pre-existing partitions, where they are appropriate, while newly expanded partitions follow the dedicated new-partition policy.

Public Interfaces

New Consumer Configuration

auto.offset.reset.new.partitions

Specifies the offset reset policy to apply to newly expanded partitions (i.e., partitions whose creation on the broker postdates the consumer group's creation timestamp). When a partition has a committed offset that is out of the available range (e.g., due to log truncation), the consumer applies the base auto.offset.reset policy regardless of the partition's creation timestamp. This config only affects partitions with no committed offset at all.

• If a partition has a committed offset, this configuration has no effect. The consumer resumes from the committed offset as usual.
• If a partition has no committed offset, the consumer applies the reset policy selected by the group coordinator. The group coordinator classifies partitions server-side and reports the result through a per-topic side channel: ConsumerGroupHeartbeatResponse.Assignment.TopicPartitions[].NewPartitions is a tagged []int32 containing the subset of partition indices in Partitions that are classified as newly expanded.
  • Partition index ∈ NewPartitions → apply auto.offset.reset.new.partitions.
  • Partition index ∉ NewPartitions → apply the base auto.offset.reset.
• When connected to a broker that does not support the new heartbeat version, NewPartitions is absent (and therefore defaults to an empty list). As a result, every partition is treated as pre-existing and the base auto.offset.reset is applied uniformly. The new behavior becomes effective automatically once the brokers are upgraded.

When this config is unset (default), the consumer applies the base auto.offset.reset uniformly to all partitions without committed offsets.

Interaction matrix

The full interaction between the base policy and the new-partition policy:

1. Default safety: real-time consumers that should never miss data on newly created partitions, but don't want to replay history on first deployment.
2. Real-time consumer that wants bounded recovery on new partitions (e.g., last hour) without replaying years of history.
3. Backfill on first deployment but treat post-deployment partition expansions as live-only.
4. Time-windowed consumer that wants a different lookback for newly created partitions than for pre-existing ones

Kafka Streams

auto.offset.reset in Kafka Streams applies only to source topics and repartition topics. Changelog topics and global state topics always use none for offset reset and are restored independently by StoreChangelogReader and GlobalStreamThread, respectively. As a result, auto.offset.reset.new.partitions does not apply to those topics.

At the StreamsConfig level, users can configure auto.offset.reset.new.partitions to apply a reset policy for newly added partitions across all source topics in the application. It has the same value space, default value, and semantics as the consumer configuration of the same name.

For per-source-topic overrides, org.apache.kafka.streams.AutoOffsetReset gains a new method, withNewPartitionsResetPolicy(...), which attaches a new-partitions reset policy to an existing AutoOffsetReset instance. The existing precedence rules remain unchanged: if a source topic specifies a Consumed.withOffsetResetPolicy(...) override, that policy (including its attached new-partitions policy, if present) takes precedence. Otherwise, the topic uses the pair of global settings configured through StreamsConfig.

public class AutoOffsetReset {
    /**
     * Returns a new {@code AutoOffsetReset} with the given policy attached as
     * the reset policy for newly expanded partitions — partitions whose creation
     * timestamp on the broker postdates the streams group's creation timestamp.
     * The base policy of this {@code AutoOffsetReset} continues to apply to all
     * other source-topic partitions without a committed offset.
     */
    public AutoOffsetReset withNewPartitionsResetPolicy(AutoOffsetReset newPartitionsPolicy) { ... }
}

Example:

Consumed.with(Serdes.String(), Serdes.String())
    .withOffsetResetPolicy(AutoOffsetReset.latest()
        .withNewPartitionsResetPolicy(AutoOffsetReset.earliest()));

Share groups

share.auto.offset.reset.new.partitions

A group-level dynamic configuration that parallels the existing share.auto.offset.reset. Like its base counterpart, it is configured on a share group and is consumed by the broker when the share-partition coordinator initializes a partition's Share Partition Start Offset (SPSO).

Its semantics mirror those of auto.offset.reset.new.partitions for consumer groups and Kafka Streams:

• If the SPSO has already been persisted for a (group, topic, partition), this configuration has no effect.

• If the SPSO has not yet been persisted, the share-partition coordinator classifies the partition server-side using the same algorithm:

  partition.creationTime > group.creationTime

  The classification result is propagated to the partition leader through the share-state persister path (see Schema Changes).

  • Classified as newly expanded → initialize the SPSO using share.auto.offset.reset.new.partitions.
  • Classified as pre-existing → initialize the SPSO using the base share.auto.offset.reset.
• If share.auto.offset.reset.new.partitions is not configured (the default), the coordinator ignores the classification result and initializes the SPSO using the base share.auto.offset.reset for all partitions. This differs from the current behavior.

The interaction matrix has the same shape as the consumer configuration described above, except that "seek to LEO" becomes "initialize the SPSO to the LEO on first fetch", and "seek to offset 0" becomes "initialize the SPSO to the log start offset". Because share consumers do not support the none policy, the corresponding row does not apply.

Schema Changes

Summary

RPC Schemas

ConsumerGroupHeartbeatResponse v2 — Adds a tagged NewPartitions field used to identify newly created partitions.

{
  "apiKey": 68,
  "type": "response",
  "name": "ConsumerGroupHeartbeatResponse",
  "validVersions": "0-2",
  "flexibleVersions": "0+",
  "fields": [
    // ... existing fields unchanged ...
    { "name": "Assignment", "type": "Assignment", "versions": "0+",
      "nullableVersions": "0+", "default": "null",
      "fields": [
        { "name": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+",
          "fields": [
            { "name": "TopicId",       "type": "uuid",    "versions": "0+" },
            { "name": "Partitions",    "type": "[]int32", "versions": "0+" },
            { "name": "NewPartitions", "type": "[]int32",
              "versions": "2+", "taggedVersions": "2+", "tag": 0,
              "default": "[]", "ignorable": true,
              "about": "Subset of partition indices in Partitions that the group coordinator classified as newly expanded (partition.creationTime > group.creationTime). For each partition index listed here that has no committed offset, the consumer applies auto.offset.reset.new.partitions; for any other partition without a committed offset, it applies the base auto.offset.reset. Absent (or empty) on older brokers; consumers then apply auto.offset.reset uniformly." }
          ]
        }
      ]
    }
  ]
}

StreamsGroupHeartbeatResponse v2 — Adds the same per-topic NewPartitions field (tagged []TopicPartition) to each assigned topic in the Streams group heartbeat response.

{
  "apiKey": 88,
  "type": "response",
  "name": "StreamsGroupHeartbeatResponse",
  // Version 1 adds TopologyDescriptionRequired (KIP-1331).
  // Version 2 adds NewPartitions (KIP-1327).
  "validVersions": "0-2",
  "flexibleVersions": "0+",
  "fields": [
    // ... existing fields unchanged ...
    { "name": "NewPartitions", "type": "[]TopicPartition",
      "versions": "2+", "taggedVersions": "2+", "tag": 0,
      "default": "[]", "ignorable": true,
      "about": "Source-topic partitions that the streams group coordinator classified as newly expanded (partition.creationTime > group.creationTime). For each (topic, partition) listed here that has no committed offset, the Streams consumer applies auto.offset.reset.new.partitions; for any other source-topic partition without a committed offset, it applies the base auto.offset.reset. Absent on older brokers; in that case the base auto.offset.reset applies uniformly." }
  ]
}

Note: The StreamsGroupHeartbeat RPC version (v2) assumes that KIP-1331's version bump (v0 → v1) has been merged first. If the merge order changes, the version numbers will be adjusted accordingly during implementation.

InitializeShareGroupStateRequest — Adds a tagged IsNewPartition (bool) field (tag 0) to each PartitionData. Carries the share-group coordinator's partition classification to the share-state persister when a share-partition is first initialized.

{
  "apiKey": 83,
  "type": "request",
  "name": "InitializeShareGroupStateRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    // ... existing fields unchanged ...
    { "name": "Topics", "type": "[]InitializeStateData", "versions": "0+", "fields": [
      // ... existing fields unchanged ...
      { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
        // ... existing fields unchanged (Partition, StateEpoch, StartOffset) ...
        { "name": "IsNewPartition", "type": "bool",
          "versions": "0+", "taggedVersions": "0+", "tag": 0,
          "default": "false", "ignorable": true,
          "about": "True if the share-group coordinator classified this partition as newly expanded (partition.creationTime > group.creationTime) at the moment the share-partition state was initialized. The share-coordinator persists this value alongside the snapshot so the partition leader can apply share.auto.offset.reset.new.partitions when computing the SPSO. False (and absent) on older coordinators." }
      ]}
    ]}
  ]
}

ReadShareGroupStateResponse — Adds a tagged IsNewPartition (bool) field (tag 0) to each PartitionResult. Returns the persisted partition classification to the partition leader when SharePartition.maybeInitialize() reads the share state.

{
  "apiKey": 84,
  "type": "response",
  "name": "ReadShareGroupStateResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Results", "type": "[]ReadStateResult", "versions": "0+", "fields": [
      // ... existing fields unchanged ...
      { "name": "Partitions", "type": "[]PartitionResult", "versions": "0+", "fields": [
        // ... existing fields unchanged (Partition, ErrorCode, ErrorMessage, StateEpoch, StartOffset, StateBatches) ...
        { "name": "IsNewPartition", "type": "bool",
          "versions": "0+", "taggedVersions": "0+", "tag": 0,
          "default": "false", "ignorable": true,
          "about": "True if this share-partition was classified as newly expanded when it was initialized. The partition leader reads this field at SharePartition init time and applies share.auto.offset.reset.new.partitions instead of share.auto.offset.reset when computing the initial SPSO. False (and absent) for share-partitions whose snapshot predates KIP-1327." }
      ]}
    ]}
  ]
}

Record Schemas

PartitionRecord — Adds tagged field CreationTimeMs at tag 3. The controller persists this value once when the partition is initially created; it is never updated thereafter. Because the field is tagged, no schema version bump is required: existing brokers silently ignore the unknown tag, while upgraded brokers populate it for all newly created partitions going forward.

{
  "apiKey": 3,
  "type": "metadata",
  "name": "PartitionRecord",
  "validVersions": "0-2",
  "flexibleVersions": "0+",
  "fields": [
    // ... existing fields unchanged ...
    { "name": "CreationTimeMs", "type": "int64", "versions": "0+",
      "taggedVersions": "0+", "tag": 3, "default": "-1",
      "about": "The time in milliseconds when this partition was first created. -1 if unknown." }
  ]
}

ConsumerGroupMetadataValue — Adds tagged field CreationTimeMs at tag 0, recorded once when the share group is first created.

{
  "apiKey": 3,
  "type": "coordinator-value",
  "name": "ConsumerGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "type": "int32", "versions": "0+",
      "about": "The group epoch." },
    { "name": "MetadataHash", "type": "int64", "versions": "0+",
      "default": "0", "taggedVersions": "0+", "tag": 0,
      "about": "The hash of all topics in the group." },
    { "name": "CreationTimeMs", "type": "int64", "versions": "0+",
      "default": "-1", "taggedVersions": "0+", "tag": 1,
      "about": "The time in milliseconds when this consumer group was first created. -1 if unknown." }
  ]
}

StreamsGroupMetadataValue — Adds tagged field CreationTimeMs at tag 1, recorded once when the consumer group is first created.

{
  "apiKey": 17,
  "type": "coordinator-value",
  "name": "StreamsGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "type": "int32", "versions": "0+",
      "about": "The group epoch." },
    // ... existing fields unchanged (including any tagged fields at tags 0 through 3) ...
    { "name": "CreationTimeMs", "type": "int64", "versions": "0+",
      "default": "-1", "taggedVersions": "0+", "tag": 4,
      "about": "The time in milliseconds when this streams group was first created. -1 if unknown." }
  ]
}

ShareGroupMetadataValue — Adds tagged field CreationTimeMs at tag 1, recorded once when the consumer group is first created.

{
  "apiKey": 11,
  "type": "coordinator-value",
  "name": "ShareGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "type": "int32", "versions": "0+",
      "about": "The group epoch." },
    { "name": "MetadataHash", "type": "int64", "versions": "0+",
      "about": "The hash of all topics in the group." },
    { "name": "CreationTimeMs", "type": "int64", "versions": "0+",
      "default": "-1", "taggedVersions": "0+", "tag": 0,
      "about": "The time in milliseconds when this share group was first created. -1 if unknown." }
  ]
}

ShareSnapshotValue — Adds a tagged IsNewPartition (bool) field (tag 1). Persists the partition classification in the initial snapshot so that it survives broker and share-state persister restarts. The persisted value is returned in every subsequent ReadShareGroupStateResponse.

{
  "apiKey": 0,
  "type": "coordinator-value",
  "name": "ShareSnapshotValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    // ... existing fields unchanged (SnapshotEpoch, StateEpoch, LeaderEpoch, StartOffset, DeliveryCompleteCount [tag 0], CreateTimestamp, WriteTimestamp, StateBatches) ...
    { "name": "IsNewPartition", "type": "bool", "versions": "0+",
      "taggedVersions": "0+", "tag": 1, "default": "false",
      "about": "True if this share-partition was classified as newly expanded when the snapshot was first written. Written once at initialization and preserved across subsequent snapshots." }
  ]
}

Feature Gating (MetadataVersion)

PartitionRecord is persisted in the KRaft metadata log. Every broker in the cluster reads from the same log and must agree on which tagged fields the controller is allowed to emit. To ensure this, this KIP introduces a new MetadataVersion:

• Name: IBP_4_4_IV1
• Description: Add PartitionRecord.CreationTimeMs tagged field (KIP-1327).
• Accessor: MetadataVersion.isPartitionCreationTimeSupported() (returns true for IBP_4_4_IV1 and later).

Before the cluster's active MetadataVersion reaches IBP_4_4_IV1, the controller does not write CreationTimeMs, even when running a broker binary that supports the feature. Once the cluster is upgraded to IBP_4_4_IV1, every newly created PartitionRecord includes the field.

Feature Gating (GroupVersion / StreamsVersion / ShareVersion)

Wire compatibility is provided by the new RPC versions together with the tagged CreationTimeMs and IsNewPartition fields. Older clients ignore unknown response fields, while newer clients automatically fall back when communicating with coordinators that do not support the updated protocol versions.

Coordinator behavior, however, must also be gated. Whether a coordinator records CreationTimeMs when a group is created and whether it computes and emits NewPartitions or IsNewPartition must be consistent across all coordinator replicas. Otherwise, during a rolling upgrade, leadership could move between upgraded and not-yet-upgraded brokers, causing CreationTimeMs to be recorded inconsistently or classification results to appear in some responses but not others.

To ensure consistent behavior, this KIP introduces a new feature level for each of the existing feature tracks, following the same feature-gating pattern used by GV_1, StreamsVersion.SV_1, and ShareVersion.SV_1.

GroupVersion

GroupVersion gates the consumer-group coordinator only.

• Name: GV_2
• Description: Enable partition-expansion classification for consumer groups (KIP-1327).
• Accessor: GroupVersion.isNewPartitionsClassificationSupported() (returns true for GV_2 and later).
• Dependency: metadata.version = IBP_4_4_IV1, so partition.creationTime is available in the metadata image.

Before the cluster's active group.version reaches GV_2, the consumer-group coordinator neither records ConsumerGroupMetadataValue.CreationTimeMs when creating a consumer group nor computes or emits NewPartitions in ConsumerGroupHeartbeatResponse, even when running on a broker binary that supports the feature.

Once the cluster reaches GV_2, both behaviors become active for consumer groups created after the upgrade.

StreamsVersion

StreamsVersion gates the streams-group coordinator only.

• Name: SV_2
• Description: Enable partition-expansion classification for Streams groups (KIP-1327).
• Accessor: StreamsVersion.isNewPartitionsClassificationSupported() (returns true for SV_2 and later).
• Dependency: metadata.version = IBP_4_4_IV1.

SV_2 has the same semantics as GV_2, but applies to the streams-group coordinator. It gates recording StreamsGroupMetadataValue.CreationTimeMs when creating a Streams group and emitting NewPartitions in StreamsGroupHeartbeatResponse.

ShareVersion

ShareVersion gates the share-group coordinator only. Since ShareVersion.SV_2 is already reserved by KIP-1191 for the share-group DLQ feature, this KIP introduces SV_3.

• Name: SV_3
• Description: Enable partition-expansion classification for share groups (KIP-1327).
• Accessor: ShareVersion.isNewPartitionsClassificationSupported() (returns true for SV_3 and later).
• Dependency: metadata.version = IBP_4_4_IV1.

ShareVersion.SV_1 is already a transitive prerequisite because share groups themselves must be enabled before any per-share-group behavior is meaningful, so no additional dependency is required.

Before the cluster's active share.version reaches SV_3, the share-group coordinator neither records ShareGroupMetadataValue.CreationTimeMs when creating a share group nor includes IsNewPartition in InitializeShareGroupStateRequest. Consequently, the persisted IsNewPartition value defaults to false, and SharePartition always applies the base share.auto.offset.reset, regardless of whether share.auto.offset.reset.new.partitions is configured.

Once the cluster reaches SV_3, these behaviors become active for share groups created after the upgrade and for share-partitions initialized after the upgrade.

Proposed Changes

Classification Algorithm

Classification is performed server-side by the group coordinator: the consumer-group coordinator for consumer groups, and the streams-group coordinator for Streams groups. In both cases, the coordinator computes a NewPartitions subset for each assignment—of Partitions for consumer groups and of each task's partition list for Streams groups—and includes it in the assignment. The client simply checks whether a partition index is present in NewPartitions and applies the corresponding reset policy.

Group coordinator

For each topic included in a heartbeat assignment, the group coordinator computes NewPartitions from the following inputs:

• partition.creationTime — read from the metadata image (PartitionRecord.CreationTimeMs), which is already replicated to both the consumer-group coordinator and the streams-group coordinator.
• group.creationTime — read from the in-memory group state, populated from ConsumerGroupMetadataValue.CreationTimeMs or StreamsGroupMetadataValue.CreationTimeMs.

A partition is included in NewPartitions if: partition.creationTime > group.creationTime

If either timestamp is -1 (the upgrade fallback value), the comparison uses conservative defaults. An unknown group.creationTime is treated as -∞, so any partition with a known creation time is classified as new. An unknown partition.creationTime is treated as pre-existing the consumer group. The following table summarizes all possible combinations:

Consumer

If auto.offset.reset.new.partitions is not configured, the consumer ignores NewPartitions and applies the base auto.offset.reset to every partition without a committed offset. This is the default behavior.

If auto.offset.reset.new.partitions is configured, then for each partition without a committed offset:

• If the partition index is present in NewPartitions, apply auto.offset.reset.new.partitions.
• Otherwise, apply the base auto.offset.reset.

Server-Side Timestamp Recording

Both CreationTimeMs fields are write-once on the server side. Once recorded, they are never overwritten — neither by subsequent metadata-log entries (for partitions) nor by group lifecycle events (for groups). This immutability is what makes classification deterministic and stable across retries.

PartitionRecord.CreationTimeMs

The controller writes a PartitionRecord exactly once per partition, at the moment of initial creation, whether via topic creation or CreatePartitionsRequest, and records CreationTimeMs from its current wall-clock time at that point.

All subsequent partition state changes (leader elections, ISR updates, replica reassignments, ELR / LastKnownElr updates, log-directory changes, etc.) are written as PartitionChangeRecord, a distinct metadata record type that does not carry CreationTimeMs. The original timestamp from the initial PartitionRecord is therefore never overwritten, and remains the authoritative value for the lifetime of the partition.

Partitions that existed before the broker upgrade have no CreationTimeMs in their original PartitionRecord (the tagged field is simply absent) and read as -1 (unknown) on the wire. Such partitions are always classified as pre-existing — the safer default, as it aligns with the user's intent when selecting the base auto.offset.reset.

ConsumerGroupMetadataValue.CreationTimeMs and StreamsGroupMetadataValue.CreationTimeMs

The group coordinator sets CreationTimeMs to its current wall-clock time when it writes the first metadata value for a group—that is, when a member sends the first ConsumerGroupHeartbeatRequest (consumer groups) or StreamsGroupHeartbeatRequest (Streams groups) and the group does not yet exist.

Subsequent writes (epoch bumps, member joins and leaves, rebalances, metadata-hash changes) preserve the original CreationTimeMs. If a group is deleted (via DeleteGroupsRequest or expiration) and later re-created, a new CreationTimeMs is recorded. As a result, all partitions assigned to the re-created group are classified relative to the new group lifetime. Existing partitions are therefore classified as pre-existing, which is the intended behavior because a re-created group represents a clean slate from the user's perspective.

Groups that existed before the upgrade have no CreationTimeMs field and therefore read as -1 (unknown). During classification, an unknown group.creationTime is conservatively treated as -∞, so any partition with a known partition.creationTime is classified as new, while partitions whose partition.creationTime is also unknown are classified as pre-existing. This behavior is identical for both consumer groups and Streams groups.

Share Groups

Share groups differ from consumer and Streams groups in two ways that make the consumer-side classification path inapplicable:

1. Reset is broker-side, not client-side. share.auto.offset.reset is a group-level dynamic configuration, and the initial Share Partition Start Offset (SPSO) is computed by SharePartition.startOffsetDuringInitialization on the partition leader the first time the share-partition is fetched—not by the share consumer. There is no client-side auto.offset.reset logic on which to layer a new policy.
2. The partition leader is not the share-group coordinator. Classification requires group.creationTime, which is maintained by the share-group coordinator. The partition leader, where the SPSO is initialized, is typically a different broker and has no direct access to the group's state.

The existing share-state persister path naturally solves both problems:

• The share-group coordinator performs classification once, when it first determines that a share-partition must be initialized via InitializeShareGroupStateRequest.
• The persister carries the resulting IsNewPartition flag alongside the existing initialization payload, persists it in the snapshot, and returns it in every subsequent ReadShareGroupStateResponse.
• During SPSO initialization, the partition leader reads IsNewPartition and selects either share.auto.offset.reset.new.partitions or the base share.auto.offset.reset accordingly.

Share-group Coordinator: Classification

When the share-group coordinator determines that a share-partition requires initialization (see GroupMetadataManager.subscribedTopicsChangeMap and maybeCreateInitializeShareGroupStateRequest), it also computes IsNewPartition for each partition included in the resulting InitializeShareGroupStateRequest.

The inputs are identical to those used by the consumer-group coordinator:

• partition.creationTime — read from the metadata image (PartitionRecord.CreationTimeMs).
• group.creationTime — read from ShareGroupMetadataValue.CreationTimeMs.

The same conservative classification rules described in Group coordinator apply. An unknown group.creationTime is treated as -∞, so partitions with a known creation time are classified as new. An unknown partition.creationTime is treated as pre-existing.

IsNewPartition is computed once, when the share-partition is initialized, and is then persisted. It is not recomputed later. This is intentional because SPSO itself is a one-time decision: once SharePartition initializes the SPSO, it never recomputes it. Recomputing the classification later would therefore have no effect and would introduce unnecessary divergence if group.creationTime were ever to change (it does not; see Server-Side Timestamp Recording).

ShareGroupMetadataValue.CreationTimeMs

The share-group coordinator records CreationTimeMs using its current wall-clock time when it writes the first ShareGroupMetadataValue for a group—that is, when it processes the first ShareGroupHeartbeatRequest for a group that does not yet exist.

Subsequent writes (epoch bumps and metadata-hash changes) preserve the original timestamp, following the same semantics as ConsumerGroupMetadataValue.CreationTimeMs.

As with consumer groups, deleting and re-creating a share group records a new CreationTimeMs. Consequently, partitions that already exist when the group is re-created are classified as pre-existing, matching the intended clean-slate behavior.

Share groups created before the upgrade have no CreationTimeMs field and therefore read as -1 (unknown). During classification, an unknown group.creationTime is conservatively treated as -∞, so any partition with a known partition.creationTime is classified as new and automatically benefits from partition-expansion protection.

Persister Path: IsNewPartition Propagation

The IsNewPartition flag travels through the existing share-state persister path:

1. InitializeShareGroupStateRequest (coordinator → persister). The share-group coordinator includes IsNewPartition for each PartitionData in the request.
2. ShareSnapshotValue (persister → metadata log). The share coordinator persists IsNewPartition in the initial snapshot for the share-partition. The field is immutable and is not carried in subsequent ShareUpdateValue records.
3. ReadShareGroupStateResponse (persister → partition leader). When SharePartition.maybeInitialize() loads the share state, the persisted IsNewPartition value is returned on every read. Since SPSO initialization occurs at most once per share-partition lifetime, only the first read is relevant in practice.

All three fields are tagged with ignorable: true, ensuring safe behavior during rolling upgrades:

• share.version < SV_3. The share-group coordinator does not emit IsNewPartition, so SharePartition applies the base share.auto.offset.reset for all share-partitions, matching pre-KIP-1327 behavior.
• Older share-coordinator persister. The persister ignores the unknown tagged field on InitializeShareGroupStateRequest, causing IsNewPartition to be stored as its default value (false). SharePartition therefore applies the base share.auto.offset.reset.
• Older partition leader. The leader ignores the unknown tagged field on ReadShareGroupStateResponse and continues applying the base share.auto.offset.reset. No incompatibility is introduced during rollout.

Partition Leader: SPSO Initialization

SharePartition.startOffsetDuringInitialization (currently core/src/main/java/kafka/server/share/SharePartition.java:3151) takes one additional input from ReadShareGroupStateResponse: IsNewPartition.

The initialization logic becomes:

1. If partitionDataStartOffset != UNINITIALIZED_START_OFFSET, use the persisted SPSO. (Unchanged.)
2. Otherwise, select the reset policy:
   • If IsNewPartition == true and share.auto.offset.reset.new.partitions is configured for the group, use share.auto.offset.reset.new.partitions.
   • Otherwise, use the base share.auto.offset.reset.
3. Resolve the selected policy to an offset through metadataProvider.offsetFor*Timestamp(...), exactly as today.

The existing ShareGroupConfigProvider.autoOffsetReset(groupId) is extended with an overload autoOffsetReset(groupId, isNewPartition) to resolve the appropriate configuration.

Compatibility, Deprecation, and Migration Plan

We discuss this improvement separately for different consumer types:

Modern consumer (KIP-848)

This is a new, opt-in feature. auto.offset.reset.new.partitions defaults to null, so existing deployments are unaffected — the consumer continues to apply the base auto.offset.reset uniformly to all partitions without committed offsets. No behavior changes for any current user until the config is explicitly set.

When the config is enabled on a cluster that does not yet support partition-expansion classification, NewPartitions is absent from the heartbeat response and treated as empty. The consumer therefore falls back to auto.offset.reset for all partitions without committed offsets. No exception is thrown, and the feature becomes effective automatically as the cluster is upgraded.

Pre-existing groups created before the upgrade

Consumer groups created before the broker was upgraded to support KIP-1327 have no recorded CreationTimeMs (the value is -1). During classification, an unknown group.creationTime is conservatively treated as -∞. In practice, this means:

• Partitions created after the broker upgrade have a recorded partition.creationTime and are classified as new under the -∞ rule. As a result, auto.offset.reset.new.partitions is applied, providing partition-expansion protection.
• Partitions created before the broker upgrade have partition.creationTime == -1 and are classified as pre-existing, so the base auto.offset.reset policy applies.

No operator intervention is required, and no log message is emitted. Once the brokers are upgraded, pre-existing consumer groups automatically benefit from partition-expansion protection for any newly created partitions.

We intentionally do not expose an admin API to mutate the group creation timestamp. Doing so would invite misuse — for example, setting it to the current time solely to force new partitions to reset to earliest — without addressing any legitimate use case.

Pre-existing partitions created before the upgrade

Partitions created before the broker upgrade have no recorded CreationTimeMs (the tagged field is absent from their PartitionRecord) and read as -1 on the wire. For these partitions, classification falls back to the base auto.offset.reset.

This fallback is the correct outcome, not a degradation: a partition that predates the upgrade also predates any post-upgrade consumer group, making it genuinely pre-existing from every reasonable group's perspective. No WARN log is emitted, and no remediation is required.

Classic consumer

This KIP does not extend auto.offset.reset.new.partitions support to classic consumer groups. If a user sets this config while using a classic consumer group, the consumer throws a ConfigException at startup, clearly indicating that the feature requires the modern consumer group protocol (group.protocol=consumer).

Kafka Streams

This KIP extends partition-expansion protection to Kafka Streams applications that use the Streams rebalance protocol (KIP-1071). For these deployments, the Streams group coordinator computes NewPartitions using the same classification algorithm. It resolves source topics for each TaskId from the stored topology, then evaluates each (topic, partition) pair via PartitionRecord.CreationTimeMs. Partitions with creationTime > group.creationTime are included in NewPartitions.

Results are returned as (topicName, partitionId) entries rather than TaskId, since classification is defined per topic-partition.

Streams applications that continue to use the classic consumer group protocol are not covered. If auto.offset.reset.new.partitions is configured (either through StreamsConfig or via AutoOffsetReset.withNewPartitionsResetPolicy(...) for a specific source topic) while the Streams rebalance protocol is not enabled, the application throws ConfigException at startup.

Startup validation

To provide fail-fast behavior, Kafka Streams validates the configuration before any StreamThread is started. The validation applies to both configuration paths:

• Global configuration. If StreamsConfig contains auto.offset.reset.new.partitions, Kafka Streams verifies that the Streams rebalance protocol (KIP-1071) is enabled via group.protocol. Otherwise, it throws ConfigException during KafkaStreams construction.
• Per-source-topic configuration. If AutoOffsetReset.withNewPartitionsResetPolicy(...) is used with Consumed.withOffsetResetPolicy(...) or Topology.addSource(...), Kafka Streams performs the same validation when building the topology. If the Streams rebalance protocol is not enabled, it throws ConfigException during StreamsBuilder.build() or KafkaStreams construction, before any consumer is created.

Upgrade behavior

Before the cluster's active MetadataVersion reaches IBP_4_4_IV1, newly created partitions do not carry partition.creationTime and therefore read as -1 (unknown). As a result, they are classified as pre-existing, and the base auto.offset.reset policy is applied, just as on the consumer-group side.

Streams groups created before the broker upgrade have no recorded StreamsGroupMetadataValue.CreationTimeMs. During classification, an unknown group.creationTime is treated as -∞, so once the MetadataVersion upgrade enables newly created PartitionRecords to carry partition.creationTime, those partitions are automatically classified as new. This behavior is identical to that of consumer groups.

Share consumer

This KIP extends partition-expansion protection to share groups (KIP-932). The share-group coordinator classifies share-partitions during initialization and propagates the result to the partition leader via the existing share-state persister path, without introducing new RPCs. Operators control the behavior through a new group-level dynamic configuration, share.auto.offset.reset.new.partitions, analogous to the existing share.auto.offset.reset.

Pre-existing share groups and partitions

The same conservative classification rules used for consumer groups apply to share groups:

• Share groups created before the upgrade have no recorded ShareGroupMetadataValue.CreationTimeMs and therefore read as -1 (unknown). During classification, an unknown group.creationTime is treated as -∞, so any partition with a known partition.creationTime is classified as new. Once metadata.version reaches IBP_4_4_IV1 and group.version reaches SV_3, pre-existing share groups automatically gain partition-expansion protection for partitions created after the broker upgrade.
• Share-partitions initialized before the upgrade either persist IsNewPartition == false or have no IsNewPartition field, which deserializes to false. In either case, the SPSO has already been initialized and persisted, so the classification result is no longer relevant in practice.
• Share-partitions whose snapshots predate the upgrade but whose SPSO has not yet been initialized are treated as pre-existing because IsNewPartition is absent (and therefore defaults to false). This conservative default matches the existing share-group behavior.

Future default consideration (Kafka 5.0)

For backward compatibility, this KIP defaults auto.offset.reset.new.partitions to null, meaning the consumer applies the base auto.offset.reset uniformly and partition-expansion protection remains opt-in.

A future KIP may revisit this default in a major release such as Kafka 5.0, changing it to earliest so that partition-expansion safety becomes the out-of-the-box behavior. The rationale:

• Silent data loss is the current default failure mode. Users running auto.offset.reset=latest — the most common base policy — silently miss records on every partition expansion unless protection has been explicitly enabled. A safe default eliminates this footgun.
• earliest aligns with user intent in nearly all cases. Selecting latest typically expresses "I do not want to reprocess history that already existed when I started." Newly expanded partitions carry no such history — every record on them was produced during the consumer group's active lifetime — so reading from the beginning is consistent with that same intent.
• Opting out is straightforward. Users who prefer the previous uniform behavior can set auto.offset.reset.new.partitions=latest (or any value matching their base policy), which is semantically a no-op overlay on the base policy.

Test Plan

All existing tests must continue to pass. New tests will cover:

Consumer

• Coordinator emits NewPartitions iff partition.creationTime > group.creationTime. Unknown values default to −∞ (group) and existing (partition).

• auto.offset.reset.new.partitions applies only to NewPartitions; base auto.offset.reset applies to the rest. Out-of-range offsets fall back to base policy.

• CreationTimeMs fields are write-once and never overwritten.

• Classic protocol rejects auto.offset.reset.new.partitions with ConfigException.

Streams

• Applies separate reset policies to new vs existing partitions via Consumed.withOffsetResetPolicy. Non-source topics are unaffected.

• Rejects auto.offset.reset.new.partitions without Streams rebalance protocol (ConfigException).

Share

• Same classification as Consumer; IsNewPartition is persisted in ShareSnapshotValue and returned in all reads.

• Initialization rule: new partitions follow share.auto.offset.reset.new.partitions if set, otherwise base share.auto.offset.reset; existing partitions always use base policy.

• CreationTimeMs is write-once; missing group creation time defaults to −∞ in SV_3+.

Rejected Alternatives

New auto.offset.reset=to_start_time policy (KIP-1282)

The controller records a group creation timestamp (GroupCreationTimeMs) when a consumer group is first created. This timestamp is returned to the consumer via the ConsumerGroupHeartbeatResponse.

When auto.offset.reset=to_start_time and the consumer encounters a partition with no committed offset or an out-of-range offset, it issues a ListOffsetsRequest using the group creation timestamp as the target. The broker returns the earliest offset at or after that timestamp. This produces the following behavior:

• New group, first start: the group creation timestamp is approximately "now", so ListOffsets resolves to the log end offset, equivalent to latest. The consumer skips the historical backlog.
• Partition expansion: the group creation timestamp predates the new partition, so ListOffsets resolves to offset 0. The consumer reads all records from the new partition, preventing data loss.
• Out-of-range (stale offset, log truncation): ListOffsets resolves to the earliest surviving offset at or after the group creation time, effectively falling back to earliest for the group's lifetime.

Why reject

• No consensus on out-of-range semantics: The community could not agree on how to_start_time should behave for out-of-range offsets. The main issue is whether it should remain consistent with latest (skip backlog) or fall back to earliest (ensure completeness). to_start_time applies different behaviors in these cases, which some see as inconsistent. A proposed “Smarter Latest” variant was rejected for introducing branching logic and violating the single-rule model.
• Overkill for the core problem: The goal—preventing data loss during partition expansion—does not require a new reset policy with additional semantics. A simpler approach, such as refining latest via configuration, would be more focused and easier to adopt. Introducing a new policy was considered disproportionate to the problem.

auto.offset.reset.max.age.ms based on partition age

This alternative proposed classifying partitions using a different signal: partition age, computed server-side as broker_current_time − partition_creation_time and returned via MetadataResponse.PartitionAgeMs. A new consumer config, auto.offset.reset.max.age.ms, would compare this age against a user-defined threshold:

• partition_age ≤ threshold → newly expanded → apply auto.offset.reset.new.partitions
• partition_age > threshold → pre-existing → apply the base auto.offset.reset

Why rejected:

• Threshold tuning leaks implementation details. The correct threshold is determined by the consumer's metadata refresh delay — an internal client-side concern. Users would be forced to reason about internal mechanics such as refresh intervals and retry behavior simply to configure the feature correctly.
• Boundary instability. A partition near the threshold can flip from "new" to "pre-existing" between successive metadata fetches as the broker's clock advances. Classification can change based purely on the timing of the next metadata refresh, with no corresponding user action.
• Group creation time is the more intuitive classifier. Users naturally reason about "new vs. pre-existing" relative to their consumer group's lifetime, not relative to broker wall-clock time. The adopted design reflects this framing directly.