kafka_actions: harden reset-consumer-offsets action (DataDog#24165)

* kafka_actions: harden reset-consumer-offsets action

- Abort before altering offsets when the consumer group still has active
  members (describe_consumer_groups pre-flight check) so the Kafka
  NON_EMPTY_GROUP error never surfaces as a cryptic partial failure.
- Inspect per-partition TopicPartition error codes in the
  alter_consumer_group_offsets result; previously only group-level
  futures were checked, letting silent per-partition failures through.
- Add reset_to: earliest|latest support: resolves the log-start or
  high-watermark via list_offsets at runtime, then commits the concrete
  offset. alter_consumer_group_offsets does not accept symbolic offsets.
- Validate that partition and offset are non-negative; reject entries
  that set both offset and reset_to or neither.
- Clarify in spec.yaml that the cluster field must be the Kafka-internal
  UUID (from AdminClient.list_topics().cluster_id), not a human-readable
  name; update examples and error messages accordingly.
- Log the full offset list in the pre-execution warning for audit trail.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: add changelog entry for PR DataDog#24165

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: fix ruff formatting

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: fix ruff import sort in remaining test files

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Fix update_consumer_group_offsets in kafka_actions

The confluent-kafka AdminClient.alter_consumer_group_offsets() API takes a
single list of ConsumerGroupTopicPartitions, not (group, partitions) as two
positional arguments, which raised "takes 2 positional arguments but 3 were
given". ConsumerGroupTopicPartitions is also exported from confluent_kafka,
not confluent_kafka.admin.

Additionally, future.result() returns a single ConsumerGroupTopicPartitions
object rather than a list of partitions, so iterate its topic_partitions to
collect per-partition errors.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* kafka_actions: add timestamp-based offset reset and sentinel offsets

Replace reset_to: earliest/latest with offset: -2/-1 sentinel values
and add a timestamp field that resolves all partitions of a topic to the
first offset at or after the given millisecond timestamp, with automatic
partition discovery when partition is omitted.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: remove unrelated cluster-UUID doc changes from spec.yaml

Reverts stray documentation claiming the 'cluster' field for
update_consumer_group_offsets must be the Kafka-internal UUID rather
than a human-readable name. That claim is unrelated to this PR's
hardening work and inconsistent with every other action's cluster
docs/examples (prod-kafka-1).

* kafka_actions: sync changelog entry with current update_consumer_group_offsets behavior

* kafka_actions: restore autogenerated file headers and fix formatting

Regenerated config_models were missing their license headers (a stray
regeneration artifact), and kafka_client.py / test_unit.py had two
small ruff formatting/import-order issues.

* kafka_actions: address round-1 review feedback on offset-reset hardening

Adds request timeouts and error handling around list_offsets/describe_consumer_groups
calls, de-dups overlapping offset targets, guards against both offset and timestamp
being set, and splits update_consumer_group_offsets into smaller resolver methods.
Extracts a shared partition-discovery helper and de-dups the config.py partition
validation. Adds unit test coverage for the client-internal sentinel/timestamp/
per-partition-error/active-member logic and parameterizes the offsets validation
tests. Cleans up doc formatting in spec.yaml and splits the changelog entry into
.fixed/.added to reflect the net-new sentinel/timestamp capabilities.

* [kafka_actions] Address round-2 review feedback

Batch the timestamp-fallback offset lookup instead of issuing one
list_offsets RPC per partition, pass the admin client into the resolver
helpers consistently, wrap the describe_consumer_groups result with the
same log-and-reraise pattern as the other admin calls, and tighten a
couple of doc/type-hint/test gaps flagged by the second review pass.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* [kafka_actions] Fix import grouping in test files (ruff I001)

CI's ruff check flagged missing blank-line separation between third-party
and first-party import groups across the test suite.

* [kafka_actions] Address round-3 review nits

Type-hint the admin param on the offset resolver helpers, and key the
fallback test's list_offsets stub on OffsetSpec type instead of request
length so it stays correct if a second fallback partition is added.

* [kafka_actions] Fix import grouping (ruff I001)

Restore the blank line between third-party and first-party import
groups after adding the OffsetSpec import.

* [kafka_actions] Trim redundant offsets doc and merge changelog entries

Remove the duplicated offset/timestamp usage block from the action
description (already covered in the per-field docs), and consolidate
the two changelog fragments into a single added entry.

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: piochelepiotr

kafka_actions/assets/configuration/spec.yaml

@@ -502,7 +502,8 @@ files:
       description: |
         Configuration for updating consumer group offsets.
         WARNING: Can cause duplicate processing or data loss.
-        Consumer group should be stopped (no active members).
+        The consumer group must have no active members before this action runs.
+        The check enforces this and will fail with a clear error if members are found.
       value:
         type: object
         required:
@@ -518,7 +519,9 @@ files:
               offset: 1000
             - topic: orders
               partition: 1
-              offset: 1500
+              offset: -2
+            - topic: payments
+              timestamp: 1735689600000
         properties:
         - name: cluster
           type: string
@@ -530,30 +533,43 @@ files:
           example: order-processor
         - name: offsets
           type: array
-          description: List of topic-partition-offset tuples to update
+          description: |
+            List of offset specifications. Each entry must specify exactly one of `offset`
+            or `timestamp`. See the action description for details.
           items:
             type: object
             required:
               - topic
-              - partition
-              - offset
             properties:
             - name: topic
               type: string
               description: Topic name
             - name: partition
               type: integer
-              description: Partition number
+              description: |
+                Non-negative partition number. Required when `offset` is specified.
+                Optional when `timestamp` is specified. Omit to target all partitions.
             - name: offset
               type: integer
-              description: New offset value
+              description: |
+                Offset to commit. Use -2 for earliest (log-start), -1 for latest
+                (high-watermark), or a non-negative integer for an explicit position.
+                Mutually exclusive with `timestamp`. Requires `partition`.
+            - name: timestamp
+              type: integer
+              description: |
+                Milliseconds since epoch. Resets to the first offset at or after this
+                timestamp. Partitions with no message at or after the timestamp are
+                reset to latest. Mutually exclusive with `offset`.
           example:
             - topic: orders
               partition: 0
               offset: 1000
             - topic: orders
               partition: 1
-              offset: 1500
+              offset: -2
+            - topic: payments
+              timestamp: 1735689600000
       fleet_configurable: false
 
     # ========================================================================

kafka_actions/changelog.d/24165.added

@@ -0,0 +1 @@
+Added earliest, latest, and timestamp-based offset support to the ``update_consumer_group_offsets`` action, along with an inactive-group precondition check and per-partition error reporting.

kafka_actions/datadog_checks/kafka_actions/check.py

@@ -743,10 +743,12 @@ def _action_update_consumer_group_offsets(self):
                 offsets:
                     - topic: orders
                       partition: 0
-                      offset: 1000
+                      offset: 1000       # explicit offset
                     - topic: orders
                       partition: 1
-                      offset: 1500
+                      offset: -2         # earliest
+                    - topic: payments
+                      timestamp: 1735689600000   # all partitions at/after this timestamp
         """
         config = self.config.update_consumer_group_offsets
 
@@ -755,10 +757,14 @@ def _action_update_consumer_group_offsets(self):
         consumer_group = config['consumer_group']
         offsets = config['offsets']
 
+        self.kafka_client.check_consumer_group_inactive(consumer_group)
+
         self.log.warning(
-            "Updating offsets for consumer group '%s' on cluster '%s' - may cause duplicate processing or data loss",
+            "Updating offsets for consumer group '%s' on cluster '%s' - may cause duplicate processing or data loss. "
+            "Offsets: %s",
             consumer_group,
             self.cluster,
+            offsets,
         )
 
         success = self.kafka_client.update_consumer_group_offsets(consumer_group=consumer_group, offsets=offsets)

kafka_actions/datadog_checks/kafka_actions/config.py

@@ -2,6 +2,7 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 import os
+from typing import Any
 
 from datadog_checks.base import ConfigurationError, is_affirmative
 
@@ -289,6 +290,11 @@ def _validate_delete_consumer_group(self):
         if not config.get('consumer_group'):
             raise ConfigurationError("delete_consumer_group action requires 'consumer_group' parameter")
 
+    def _validate_offset_entry_partition(self, offset_entry: dict[str, Any], index: int) -> None:
+        """Validate that offsets[index].partition, if present, is a non-negative integer."""
+        if not isinstance(offset_entry['partition'], int) or offset_entry['partition'] < 0:
+            raise ConfigurationError(f"offsets[{index}].partition must be a non-negative integer")
+
     def _validate_update_consumer_group_offsets(self):
         """Validate update_consumer_group_offsets action configuration."""
         config = self.update_consumer_group_offsets
@@ -310,17 +316,36 @@ def _validate_update_consumer_group_offsets(self):
             if not offset_entry.get('topic'):
                 raise ConfigurationError(f"offsets[{i}] requires 'topic' parameter")
 
-            if 'partition' not in offset_entry:
-                raise ConfigurationError(f"offsets[{i}] requires 'partition' parameter")
+            has_offset = 'offset' in offset_entry
+            has_timestamp = 'timestamp' in offset_entry
+
+            if not has_offset and not has_timestamp:
+                raise ConfigurationError(f"offsets[{i}] requires 'offset' or 'timestamp'")
+
+            if has_offset and has_timestamp:
+                raise ConfigurationError(f"offsets[{i}] cannot specify both 'offset' and 'timestamp'")
 
-            if 'offset' not in offset_entry:
-                raise ConfigurationError(f"offsets[{i}] requires 'offset' parameter")
+            if has_offset:
+                if 'partition' not in offset_entry:
+                    raise ConfigurationError(f"offsets[{i}] requires 'partition' when 'offset' is specified")
 
-            if not isinstance(offset_entry.get('partition'), int):
-                raise ConfigurationError(f"offsets[{i}].partition must be an integer")
+                self._validate_offset_entry_partition(offset_entry, i)
+
+                offset_val = offset_entry['offset']
+                if not isinstance(offset_val, int) or offset_val < -2:
+                    raise ConfigurationError(
+                        f"offsets[{i}].offset must be -2 (earliest), -1 (latest), or a non-negative integer"
+                    )
+
+            if has_timestamp:
+                ts = offset_entry['timestamp']
+                if not isinstance(ts, int) or ts <= 0:
+                    raise ConfigurationError(
+                        f"offsets[{i}].timestamp must be a positive integer (milliseconds since epoch)"
+                    )
 
-            if not isinstance(offset_entry.get('offset'), int):
-                raise ConfigurationError(f"offsets[{i}].offset must be an integer")
+                if 'partition' in offset_entry:
+                    self._validate_offset_entry_partition(offset_entry, i)
 
     def _validate_produce_message(self):
         """Validate produce_message action configuration."""

kafka_actions/datadog_checks/kafka_actions/config_models/instance.py

@@ -205,8 +205,18 @@ class Offset(BaseModel):
         arbitrary_types_allowed=True,
         frozen=True,
     )
-    offset: int = Field(..., description='New offset value')
-    partition: int = Field(..., description='Partition number')
+    offset: Optional[int] = Field(
+        None,
+        description='Offset to commit. Use -2 for earliest (log-start), -1 for latest\n(high-watermark), or a non-negative integer for an explicit position.\nMutually exclusive with `timestamp`. Requires `partition`.\n',
+    )
+    partition: Optional[int] = Field(
+        None,
+        description='Non-negative partition number. Required when `offset` is specified.\nOptional when `timestamp` is specified. Omit to target all partitions.\n',
+    )
+    timestamp: Optional[int] = Field(
+        None,
+        description='Milliseconds since epoch. Resets to the first offset at or after this\ntimestamp. Partitions with no message at or after the timestamp are\nreset to latest. Mutually exclusive with `offset`.\n',
+    )
     topic: str = Field(..., description='Topic name')
 
 
@@ -219,11 +229,12 @@ class UpdateConsumerGroupOffsets(BaseModel):
     consumer_group: str = Field(..., description='Consumer group ID to update', examples=['order-processor'])
     offsets: tuple[Offset, ...] = Field(
         ...,
-        description='List of topic-partition-offset tuples to update',
+        description='List of offset specifications. Each entry must specify exactly one of `offset`\nor `timestamp`. See the action description for details.\n',
         examples=[
             [
                 {'offset': 1000, 'partition': 0, 'topic': 'orders'},
-                {'offset': 1500, 'partition': 1, 'topic': 'orders'},
+                {'offset': -2, 'partition': 1, 'topic': 'orders'},
+                {'timestamp': 1735689600000, 'topic': 'payments'},
             ]
         ],
     )

kafka_actions/datadog_checks/kafka_actions/data/conf.yaml.example

@@ -259,7 +259,8 @@ instances:
     ## @param update_consumer_group_offsets - mapping - optional
     ## Configuration for updating consumer group offsets.
     ## WARNING: Can cause duplicate processing or data loss.
-    ## Consumer group should be stopped (no active members).
+    ## The consumer group must have no active members before this action runs.
+    ## The check enforces this and will fail with a clear error if members are found.
     #
     # update_consumer_group_offsets:
     #   cluster: prod-kafka-1
@@ -270,7 +271,9 @@ instances:
     #     offset: 1000
     #   - topic: orders
     #     partition: 1
-    #     offset: 1500
+    #     offset: -2
+    #   - topic: payments
+    #     timestamp: 1735689600000
 
     ## @param produce_message - mapping - optional
     ## Configuration for producing a message to a Kafka topic.