Repository: redpanda-data/connect (language: go)

Issue:
Add more timeout fields to kafka/redpanda input/outputs
There are some extra fields we should add for customizing the default timeouts, specifically for these franz-go fields: https://github.com/twmb/franz-go/blob/0a757d1a4731998f31807dd84c325abf0620b4cf/pkg/kgo/config.go#L283-L288


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 929c74419e..8d59935fc6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,8 +12,13 @@ All notable changes to this project will be documented in this file.
 - Field `use_enum_numbers` added to `protobuf` processor. (@benwebber)
 - Field `tools` added to `cohere_chat` processor. (@rockwotj)
 - Fields `region`, `endpoint` and `credentials` added to the `dynamodb` configuration section of the `aws_kinesis` input. (@jreyeshdez, @mihaitodor)
-- Field `isolation_level` added to `kafka`, `kafka_franz`, `ockam_kafka`, `redpanda`, `redpanda_common`, and `redpanda_migrator` inputs. (@rockwotj)
+- Field `transaction_isolation_level` added to `kafka_franz`, `ockam_kafka`, `redpanda`, `redpanda_common`, and `redpanda_migrator` inputs. (@rockwotj)
 - New `cohere_rerank` processor to rerank documents in RAG pipelines using Cohere. (@rockwotj)
+- Fields `request_timeout_overhead`, `conn_idle_timeout` and `start_offset` added to the `kafka_franz`, `ockam_kafka`, `redpanda`, `redpanda_common`, and `redpanda_migrator` inputs. (@mihaitodor)
+
+### Changed
+
+- Field `start_from_oldest` for the `kafka_franz`, `ockam_kafka`, `redpanda`, `redpanda_common`, and `redpanda_migrator` inputs is now deprecated in favour of `start_offset`. (@mihaitodor)
 
 ## 4.52.0 - 2025-04-03
 
diff --git a/docs/modules/components/pages/inputs/kafka_franz.adoc b/docs/modules/components/pages/inputs/kafka_franz.adoc
index 0fc0550faf..88255a0bdc 100644
--- a/docs/modules/components/pages/inputs/kafka_franz.adoc
+++ b/docs/modules/components/pages/inputs/kafka_franz.adoc
@@ -68,6 +68,8 @@ input:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topics: [] # No default (required)
     regexp_topics: false
     rack_id: ""
@@ -75,7 +77,7 @@ input:
     rebalance_timeout: 45s
     session_timeout: 1m
     heartbeat_interval: 3s
-    start_from_oldest: true
+    start_offset: earliest
     fetch_max_bytes: 50MiB
     fetch_max_wait: 5s
     fetch_min_bytes: 1B
@@ -496,6 +498,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topics`
 
 A list of topics to consume from. Multiple comma separated topics can be listed in a single element. When a `consumer_group` is specified partitions are automatically distributed across consumers of a topic, otherwise all partitions are consumed.
@@ -587,14 +607,26 @@ When using a consumer group, `heartbeat_interval` sets how long a group member g
 
 *Default*: `"3s"`
 
-=== `start_from_oldest`
+=== `start_offset`
 
-Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists.
+Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.
 
 
-*Type*: `bool`
+*Type*: `string`
 
-*Default*: `true`
+*Default*: `"earliest"`
+
+|===
+| Option | Summary
+
+| `committed`
+| Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka's `auto.offset.reset=none` option
+| `earliest`
+| Start from the earliest offset. Corresponds to Kafka's `auto.offset.reset=earliest` option.
+| `latest`
+| Start from the latest offset. Corresponds to Kafka's `auto.offset.reset=latest` option.
+
+|===
 
 === `fetch_max_bytes`
 
diff --git a/docs/modules/components/pages/inputs/ockam_kafka.adoc b/docs/modules/components/pages/inputs/ockam_kafka.adoc
index 0b34565fdd..3549c4ba6e 100644
--- a/docs/modules/components/pages/inputs/ockam_kafka.adoc
+++ b/docs/modules/components/pages/inputs/ockam_kafka.adoc
@@ -80,7 +80,7 @@ input:
       rebalance_timeout: 45s
       session_timeout: 1m
       heartbeat_interval: 3s
-      start_from_oldest: true
+      start_offset: earliest
       fetch_max_bytes: 50MiB
       fetch_max_wait: 5s
       fetch_min_bytes: 1B
@@ -393,14 +393,26 @@ When using a consumer group, `heartbeat_interval` sets how long a group member g
 
 *Default*: `"3s"`
 
-=== `kafka.start_from_oldest`
+=== `kafka.start_offset`
 
-Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists.
+Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.
 
 
-*Type*: `bool`
+*Type*: `string`
+
+*Default*: `"earliest"`
+
+|===
+| Option | Summary
 
-*Default*: `true`
+| `committed`
+| Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka's `auto.offset.reset=none` option
+| `earliest`
+| Start from the earliest offset. Corresponds to Kafka's `auto.offset.reset=earliest` option.
+| `latest`
+| Start from the latest offset. Corresponds to Kafka's `auto.offset.reset=latest` option.
+
+|===
 
 === `kafka.fetch_max_bytes`
 
diff --git a/docs/modules/components/pages/inputs/redpanda.adoc b/docs/modules/components/pages/inputs/redpanda.adoc
index 9181849fca..4a08ec8d71 100644
--- a/docs/modules/components/pages/inputs/redpanda.adoc
+++ b/docs/modules/components/pages/inputs/redpanda.adoc
@@ -66,6 +66,8 @@ input:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topics: [] # No default (required)
     regexp_topics: false
     rack_id: ""
@@ -73,7 +75,7 @@ input:
     rebalance_timeout: 45s
     session_timeout: 1m
     heartbeat_interval: 3s
-    start_from_oldest: true
+    start_offset: earliest
     fetch_max_bytes: 50MiB
     fetch_max_wait: 5s
     fetch_min_bytes: 1B
@@ -522,6 +524,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topics`
 
 A list of topics to consume from. Multiple comma separated topics can be listed in a single element. When a `consumer_group` is specified partitions are automatically distributed across consumers of a topic, otherwise all partitions are consumed.
@@ -613,14 +633,26 @@ When using a consumer group, `heartbeat_interval` sets how long a group member g
 
 *Default*: `"3s"`
 
-=== `start_from_oldest`
+=== `start_offset`
 
-Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists.
+Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.
 
 
-*Type*: `bool`
+*Type*: `string`
 
-*Default*: `true`
+*Default*: `"earliest"`
+
+|===
+| Option | Summary
+
+| `committed`
+| Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka's `auto.offset.reset=none` option
+| `earliest`
+| Start from the earliest offset. Corresponds to Kafka's `auto.offset.reset=earliest` option.
+| `latest`
+| Start from the latest offset. Corresponds to Kafka's `auto.offset.reset=latest` option.
+
+|===
 
 === `fetch_max_bytes`
 
diff --git a/docs/modules/components/pages/inputs/redpanda_common.adoc b/docs/modules/components/pages/inputs/redpanda_common.adoc
index 2b676da099..45bb8679eb 100644
--- a/docs/modules/components/pages/inputs/redpanda_common.adoc
+++ b/docs/modules/components/pages/inputs/redpanda_common.adoc
@@ -61,7 +61,7 @@ input:
     rebalance_timeout: 45s
     session_timeout: 1m
     heartbeat_interval: 3s
-    start_from_oldest: true
+    start_offset: earliest
     fetch_max_bytes: 50MiB
     fetch_max_wait: 5s
     fetch_min_bytes: 1B
@@ -222,14 +222,26 @@ When using a consumer group, `heartbeat_interval` sets how long a group member g
 
 *Default*: `"3s"`
 
-=== `start_from_oldest`
+=== `start_offset`
 
-Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists.
+Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.
 
 
-*Type*: `bool`
+*Type*: `string`
 
-*Default*: `true`
+*Default*: `"earliest"`
+
+|===
+| Option | Summary
+
+| `committed`
+| Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka's `auto.offset.reset=none` option
+| `earliest`
+| Start from the earliest offset. Corresponds to Kafka's `auto.offset.reset=earliest` option.
+| `latest`
+| Start from the latest offset. Corresponds to Kafka's `auto.offset.reset=latest` option.
+
+|===
 
 === `fetch_max_bytes`
 
diff --git a/docs/modules/components/pages/inputs/redpanda_migrator.adoc b/docs/modules/components/pages/inputs/redpanda_migrator.adoc
index 5c7ec18152..b42119cbb0 100644
--- a/docs/modules/components/pages/inputs/redpanda_migrator.adoc
+++ b/docs/modules/components/pages/inputs/redpanda_migrator.adoc
@@ -68,6 +68,8 @@ input:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topics: [] # No default (required)
     regexp_topics: false
     rack_id: ""
@@ -75,7 +77,7 @@ input:
     rebalance_timeout: 45s
     session_timeout: 1m
     heartbeat_interval: 3s
-    start_from_oldest: true
+    start_offset: earliest
     fetch_max_bytes: 50MiB
     fetch_max_wait: 5s
     fetch_min_bytes: 1B
@@ -499,6 +501,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topics`
 
 A list of topics to consume from. Multiple comma separated topics can be listed in a single element. When a `consumer_group` is specified partitions are automatically distributed across consumers of a topic, otherwise all partitions are consumed.
@@ -590,14 +610,26 @@ When using a consumer group, `heartbeat_interval` sets how long a group member g
 
 *Default*: `"3s"`
 
-=== `start_from_oldest`
+=== `start_offset`
 
-Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists.
+Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.
 
 
-*Type*: `bool`
+*Type*: `string`
 
-*Default*: `true`
+*Default*: `"earliest"`
+
+|===
+| Option | Summary
+
+| `committed`
+| Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka's `auto.offset.reset=none` option
+| `earliest`
+| Start from the earliest offset. Corresponds to Kafka's `auto.offset.reset=earliest` option.
+| `latest`
+| Start from the latest offset. Corresponds to Kafka's `auto.offset.reset=latest` option.
+
+|===
 
 === `fetch_max_bytes`
 
diff --git a/docs/modules/components/pages/inputs/redpanda_migrator_offsets.adoc b/docs/modules/components/pages/inputs/redpanda_migrator_offsets.adoc
index 26a837c211..644ff36625 100644
--- a/docs/modules/components/pages/inputs/redpanda_migrator_offsets.adoc
+++ b/docs/modules/components/pages/inputs/redpanda_migrator_offsets.adoc
@@ -67,6 +67,8 @@ input:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topics: [] # No default (required)
     regexp_topics: false
     rack_id: ""
@@ -482,6 +484,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topics`
 
 A list of topics to consume from. Multiple comma separated topics can be listed in a single element. When a `consumer_group` is specified partitions are automatically distributed across consumers of a topic, otherwise all partitions are consumed.
diff --git a/docs/modules/components/pages/outputs/kafka_franz.adoc b/docs/modules/components/pages/outputs/kafka_franz.adoc
index 6bdaf4c40d..c60a03a512 100644
--- a/docs/modules/components/pages/outputs/kafka_franz.adoc
+++ b/docs/modules/components/pages/outputs/kafka_franz.adoc
@@ -75,6 +75,8 @@ output:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topic: "" # No default (required)
     key: "" # No default (optional)
     partition: ${! meta("partition") } # No default (optional)
@@ -484,6 +486,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topic`
 
 A topic to write messages to.
diff --git a/docs/modules/components/pages/outputs/redpanda.adoc b/docs/modules/components/pages/outputs/redpanda.adoc
index bf0d94a62d..1eed194a6f 100644
--- a/docs/modules/components/pages/outputs/redpanda.adoc
+++ b/docs/modules/components/pages/outputs/redpanda.adoc
@@ -68,6 +68,8 @@ output:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topic: "" # No default (required)
     key: "" # No default (optional)
     partition: ${! meta("partition") } # No default (optional)
@@ -469,6 +471,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topic`
 
 A topic to write messages to.
diff --git a/docs/modules/components/pages/outputs/redpanda_migrator.adoc b/docs/modules/components/pages/outputs/redpanda_migrator.adoc
index 148beb1810..c00ff0eb15 100644
--- a/docs/modules/components/pages/outputs/redpanda_migrator.adoc
+++ b/docs/modules/components/pages/outputs/redpanda_migrator.adoc
@@ -70,6 +70,8 @@ output:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     topic: "" # No default (required)
     key: "" # No default (optional)
     partition: ${! meta("partition") } # No default (optional)
@@ -515,6 +517,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `topic`
 
 A topic to write messages to.
diff --git a/docs/modules/components/pages/outputs/redpanda_migrator_offsets.adoc b/docs/modules/components/pages/outputs/redpanda_migrator_offsets.adoc
index 913baedbed..d830dcdb22 100644
--- a/docs/modules/components/pages/outputs/redpanda_migrator_offsets.adoc
+++ b/docs/modules/components/pages/outputs/redpanda_migrator_offsets.adoc
@@ -69,6 +69,8 @@ output:
       client_certs: []
     sasl: [] # No default (optional)
     metadata_max_age: 5m
+    request_timeout_overhead: 10s
+    conn_idle_timeout: 20s
     offset_topic: ${! @kafka_offset_topic }
     offset_group: ${! @kafka_offset_group }
     offset_partition: ${! @kafka_offset_partition }
@@ -469,6 +471,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `offset_topic`
 
 Kafka offset topic.
diff --git a/docs/modules/components/pages/redpanda/about.adoc b/docs/modules/components/pages/redpanda/about.adoc
index 6dc3f4ee6a..1aa1cf951f 100644
--- a/docs/modules/components/pages/redpanda/about.adoc
+++ b/docs/modules/components/pages/redpanda/about.adoc
@@ -52,6 +52,8 @@ redpanda:
     client_certs: []
   sasl: [] # No default (optional)
   metadata_max_age: 5m
+  request_timeout_overhead: 10s
+  conn_idle_timeout: 20s
   pipeline_id: ""
   logs_topic: ""
   logs_level: info
@@ -446,6 +448,24 @@ The maximum age of metadata before it is refreshed.
 
 *Default*: `"5m"`
 
+=== `request_timeout_overhead`
+
+The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.
+
+
+*Type*: `string`
+
+*Default*: `"10s"`
+
+=== `conn_idle_timeout`
+
+The rough amount of time to allow connections to idle before they are closed.
+
+
+*Type*: `string`
+
+*Default*: `"20s"`
+
 === `pipeline_id`
 
 An optional identifier for the pipeline, this will be present in logs and status updates sent to topics.
diff --git a/internal/impl/kafka/enterprise/redpanda_common_input.go b/internal/impl/kafka/enterprise/redpanda_common_input.go
index dc4d74a7f2..8de2fea783 100644
--- a/internal/impl/kafka/enterprise/redpanda_common_input.go
+++ b/internal/impl/kafka/enterprise/redpanda_common_input.go
@@ -85,16 +85,7 @@ This input adds the following metadata fields to each message:
 - All record headers
 ` + "```" + `
 `).
-		LintRule(`
-let has_topic_partitions = this.topics.any(t -> t.contains(":"))
-root = if $has_topic_partitions {
-  if this.consumer_group.or("") != "" {
-    "this input does not support both a consumer group and explicit topic partitions"
-  } else if this.regexp_topics {
-    "this input does not support both regular expression topics and explicit topic partitions"
-  }
-}
-`)
+		LintRule(kafka.FranzConsumerFieldLintRules)
 }
 
 func init() {
diff --git a/internal/impl/kafka/enterprise/redpanda_migrator_input.go b/internal/impl/kafka/enterprise/redpanda_migrator_input.go
index 0649c88785..d3ffc08cd9 100644
--- a/internal/impl/kafka/enterprise/redpanda_migrator_input.go
+++ b/internal/impl/kafka/enterprise/redpanda_migrator_input.go
@@ -67,20 +67,7 @@ This input adds the following metadata fields to each message:
 ` + "```" + `
 `).
 		Fields(redpandaMigratorInputConfigFields()...).
-		LintRule(`
-let has_topic_partitions = this.topics.any(t -> t.contains(":"))
-root = if $has_topic_partitions {
-  if this.consumer_group.or("") != "" {
-    "this input does not support both a consumer group and explicit topic partitions"
-  } else if this.regexp_topics {
-    "this input does not support both regular expression topics and explicit topic partitions"
-  }
-} else {
-  if this.consumer_group.or("") == "" {
-    "a consumer group is mandatory when not using explicit topic partitions"
-  }
-}
-`)
+		LintRule(kafka.FranzConsumerFieldLintRules)
 }
 
 func redpandaMigratorInputConfigFields() []*service.ConfigField {
diff --git a/internal/impl/kafka/franz_client.go b/internal/impl/kafka/franz_client.go
index d569cba5c5..1e09f16276 100644
--- a/internal/impl/kafka/franz_client.go
+++ b/internal/impl/kafka/franz_client.go
@@ -27,10 +27,12 @@ import (
 
 const (
 	// Connection fields
-	kfcFieldSeedBrokers    = "seed_brokers"
-	kfcFieldClientID       = "client_id"
-	kfcFieldTLS            = "tls"
-	kfcFieldMetadataMaxAge = "metadata_max_age"
+	kfcFieldSeedBrokers            = "seed_brokers"
+	kfcFieldClientID               = "client_id"
+	kfcFieldTLS                    = "tls"
+	kfcFieldMetadataMaxAge         = "metadata_max_age"
+	kfcFieldRequestTimeoutOverhead = "request_timeout_overhead"
+	kfcFieldConnIdleTimeout        = "conn_idle_timeout"
 )
 
 // FranzConnectionFields returns a slice of fields specifically for establishing
@@ -52,18 +54,28 @@ func FranzConnectionFields() []*service.ConfigField {
 			Description("The maximum age of metadata before it is refreshed.").
 			Default("5m").
 			Advanced(),
+		service.NewDurationField(kfcFieldRequestTimeoutOverhead).
+			Description("The request time overhead. Uses the given time as overhead while deadlining requests. Roughly equivalent to request.timeout.ms, but grants additional time to requests that have timeout fields.").
+			Default("10s").
+			Advanced(),
+		service.NewDurationField(kfcFieldConnIdleTimeout).
+			Description("The rough amount of time to allow connections to idle before they are closed.").
+			Default("20s").
+			Advanced(),
 	}
 }
 
 // FranzConnectionDetails describes information required to create a kafka
 // connection.
 type FranzConnectionDetails struct {
-	SeedBrokers []string
-	ClientID    string
-	TLSEnabled  bool
-	TLSConf     *tls.Config
-	SASL        []sasl.Mechanism
-	MetaMaxAge  time.Duration
+	SeedBrokers            []string
+	ClientID               string
+	TLSEnabled             bool
+	TLSConf                *tls.Config
+	SASL                   []sasl.Mechanism
+	MetaMaxAge             time.Duration
+	RequestTimeoutOverhead time.Duration
+	ConnIdleTimeout        time.Duration
 
 	Logger *service.Logger
 }
@@ -99,6 +111,14 @@ func FranzConnectionDetailsFromConfig(conf *service.ParsedConfig, log *service.L
 		return nil, err
 	}
 
+	if d.RequestTimeoutOverhead, err = conf.FieldDuration(kfcFieldRequestTimeoutOverhead); err != nil {
+		return nil, err
+	}
+
+	if d.ConnIdleTimeout, err = conf.FieldDuration(kfcFieldConnIdleTimeout); err != nil {
+		return nil, err
+	}
+
 	return &d, nil
 }
 
@@ -111,6 +131,8 @@ func (d *FranzConnectionDetails) FranzOpts() []kgo.Opt {
 		kgo.SASL(d.SASL...),
 		kgo.ClientID(d.ClientID),
 		kgo.MetadataMaxAge(d.MetaMaxAge),
+		kgo.RequestTimeoutOverhead(d.RequestTimeoutOverhead),
+		kgo.ConnIdleTimeout(d.ConnIdleTimeout),
 	}
 
 	if d.TLSEnabled {
diff --git a/internal/impl/kafka/franz_reader.go b/internal/impl/kafka/franz_reader.go
index 4547bb0b83..eb3767d5a3 100644
--- a/internal/impl/kafka/franz_reader.go
+++ b/internal/impl/kafka/franz_reader.go
@@ -15,6 +15,7 @@
 package kafka
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
@@ -53,6 +54,7 @@ const (
 	kfrFieldTopics                 = "topics"
 	kfrFieldRegexpTopics           = "regexp_topics"
 	kfrFieldStartFromOldest        = "start_from_oldest"
+	kfrFieldStartOffset            = "start_offset"
 	kfrFieldFetchMaxBytes          = "fetch_max_bytes"
 	kfrFieldFetchMinBytes          = "fetch_min_bytes"
 	kfrFieldFetchMaxPartitionBytes = "fetch_max_partition_bytes"
@@ -73,6 +75,45 @@ const (
 	TransactionIsolationLevelReadCommitted TransactionIsolationLevel = "read_committed"
 )
 
+// startOffsetType describes the offset to start consuming from, or if OffsetOutOfRange is seen while fetching,
+// to restart consuming from.
+type startOffsetType string
+
+const (
+	// startOffsetEarliest corresponds to auto.offset.reset "earliest"
+	startOffsetEarliest startOffsetType = "earliest"
+	// startOffsetLatest corresponds to auto.offset.reset "latest"
+	startOffsetLatest startOffsetType = "latest"
+	// startOffsetCommitted corresponds to auto.offset.reset "none"
+	startOffsetCommitted startOffsetType = "committed"
+)
+
+const (
+	// FranzConsumerFieldLintRules contains the lint rules for the consumer fields.
+	FranzConsumerFieldLintRules = `
+let has_topic_partitions = this.topics.any(t -> t.contains(":"))
+
+root = [
+  if $has_topic_partitions {
+    if this.consumer_group.or("") != "" {
+      "this input does not support both a consumer group and explicit topic partitions"
+    } else if this.regexp_topics {
+      "this input does not support both regular expression topics and explicit topic partitions"
+    }
+  } else {
+    if this.consumer_group.or("") == "" {
+      "a consumer group is mandatory when not using explicit topic partitions"
+    }
+  },
+  # We don't have any way to distinguish between start_from_oldest set explicitly to true and not set at all, so we
+  # assume users will be OK if start_offset overwrites it silently
+  if this.start_from_oldest == false && this.start_offset == "earliest" {
+    "start_from_oldest cannot be set to false when start_offset is set to earliest"
+  }
+]
+`
+)
+
 // FranzConsumerFields returns a slice of fields specifically for customising
 // consumer behaviour via the franz-go library.
 func FranzConsumerFields() []*service.ConfigField {
@@ -116,6 +157,14 @@ Finally, it's also possible to specify an explicit offset to consume from by add
 		service.NewBoolField(kfrFieldStartFromOldest).
 			Description("Determines whether to consume from the oldest available offset, otherwise messages are consumed from the latest offset. The setting is applied when creating a new consumer group or the saved offset no longer exists.").
 			Default(true).
+			Advanced().
+			Deprecated(),
+		service.NewStringAnnotatedEnumField(kfrFieldStartOffset, map[string]string{
+			string(startOffsetEarliest):  "Start from the earliest offset. Corresponds to Kafka's `auto.offset.reset=earliest` option.",
+			string(startOffsetLatest):    "Start from the latest offset. Corresponds to Kafka's `auto.offset.reset=latest` option.",
+			string(startOffsetCommitted): "Prevents consuming a partition in a group if the partition has no prior commits. Corresponds to Kafka's `auto.offset.reset=none` option",
+		}).Description("Sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.").
+			Default(string(startOffsetEarliest)).
 			Advanced(),
 		service.NewStringField(kfrFieldFetchMaxBytes).
 			Description("Sets the maximum amount of bytes a broker will try to send during a fetch. Note that brokers may not obey this limit if it has records larger than this limit. This is the equivalent to the Java fetch.max.bytes setting.").
@@ -151,7 +200,7 @@ type FranzConsumerDetails struct {
 	SessionTimeout         time.Duration
 	RebalanceTimeout       time.Duration
 	HeartbeatInterval      time.Duration
-	InitialOffset          kgo.Offset
+	StartOffset            kgo.Offset
 	Topics                 []string
 	TopicPartitions        map[string]map[int32]kgo.Offset
 	RegexPattern           bool
@@ -199,19 +248,29 @@ func FranzConsumerDetailsFromConfig(conf *service.ParsedConfig) (*FranzConsumerD
 		return nil, fmt.Errorf("invalid transaction isolation level: %v", isolationLevelStr)
 	}
 
-	startFromOldest, err := conf.FieldBool(kfrFieldStartFromOldest)
+	startOffset, err := conf.FieldString(kfrFieldStartOffset)
 	if err != nil {
 		return nil, err
 	}
-	var defaultOffset int64 = -1
-	if startFromOldest {
-		defaultOffset = -2
+
+	switch startOffsetType(startOffset) {
+	case startOffsetEarliest:
+		d.StartOffset = kgo.NewOffset().AtStart()
+	case startOffsetLatest:
+		d.StartOffset = kgo.NewOffset().AtEnd()
+	case startOffsetCommitted:
+		d.StartOffset = kgo.NewOffset().AtCommitted()
+	default:
+		return nil, fmt.Errorf("invalid start offset type: %s", startOffset)
 	}
 
-	if startFromOldest {
-		d.InitialOffset = kgo.NewOffset().AtStart()
-	} else {
-		d.InitialOffset = kgo.NewOffset().AtEnd()
+	startFromOldest, err := conf.FieldBool(kfrFieldStartFromOldest)
+	if err != nil {
+		return nil, err
+	}
+
+	if !startFromOldest && d.StartOffset == kgo.NewOffset().AtStart() {
+		return nil, errors.New("start_from_oldest cannot be set to false when start_offset is set to earliest")
 	}
 
 	topicList, err := conf.FieldStringList(kfrFieldTopics)
@@ -220,7 +279,7 @@ func FranzConsumerDetailsFromConfig(conf *service.ParsedConfig) (*FranzConsumerD
 	}
 
 	var topicPartitionsInts map[string]map[int32]int64
-	if d.Topics, topicPartitionsInts, err = ParseTopics(topicList, defaultOffset, true); err != nil {
+	if d.Topics, topicPartitionsInts, err = ParseTopics(topicList, d.StartOffset.EpochOffset().Offset, true); err != nil {
 		return nil, err
 	}
 
@@ -263,7 +322,7 @@ func (d *FranzConsumerDetails) FranzOpts() []kgo.Opt {
 		kgo.Rack(d.RackID),
 		kgo.ConsumeTopics(d.Topics...),
 		kgo.ConsumePartitions(d.TopicPartitions),
-		kgo.ConsumeResetOffset(d.InitialOffset),
+		kgo.ConsumeResetOffset(d.StartOffset),
 		kgo.FetchMaxBytes(d.FetchMaxBytes),
 		kgo.FetchMinBytes(d.FetchMinBytes),
 		kgo.FetchMaxPartitionBytes(d.FetchMaxPartitionBytes),
diff --git a/internal/impl/kafka/input_kafka_franz.go b/internal/impl/kafka/input_kafka_franz.go
index bb6301e760..e9a9496958 100644
--- a/internal/impl/kafka/input_kafka_franz.go
+++ b/internal/impl/kafka/input_kafka_franz.go
@@ -48,20 +48,7 @@ This input adds the following metadata fields to each message:
 ` + "```" + `
 `).
 		Fields(FranzKafkaInputConfigFields()...).
-		LintRule(`
-let has_topic_partitions = this.topics.any(t -> t.contains(":"))
-root = if $has_topic_partitions {
-  if this.consumer_group.or("") != "" {
-    "this input does not support both a consumer group and explicit topic partitions"
-  } else if this.regexp_topics {
-    "this input does not support both regular expression topics and explicit topic partitions"
-  }
-} else {
-  if this.consumer_group.or("") == "" {
-    "a consumer group is mandatory when not using explicit topic partitions"
-  }
-}
-`)
+		LintRule(FranzConsumerFieldLintRules)
 }
 
 // FranzKafkaInputConfigFields returns the full suite of config fields for a
diff --git a/internal/impl/kafka/input_redpanda.go b/internal/impl/kafka/input_redpanda.go
index b8a6217c35..7639595031 100644
--- a/internal/impl/kafka/input_redpanda.go
+++ b/internal/impl/kafka/input_redpanda.go
@@ -81,20 +81,7 @@ This input adds the following metadata fields to each message:
 ` + "```" + `
 `).
 		Fields(redpandaInputConfigFields()...).
-		LintRule(`
-let has_topic_partitions = this.topics.any(t -> t.contains(":"))
-root = if $has_topic_partitions {
-  if this.consumer_group.or("") != "" {
-    "this input does not support both a consumer group and explicit topic partitions"
-  } else if this.regexp_topics {
-    "this input does not support both regular expression topics and explicit topic partitions"
-  }
-} else {
-  if this.consumer_group.or("") == "" {
-    "a consumer group is mandatory when not using explicit topic partitions"
-  }
-}
-`)
+		LintRule(FranzConsumerFieldLintRules)
 }
 
 func redpandaInputConfigFields() []*service.ConfigField {
diff --git a/internal/impl/ockam/input_kafka.go b/internal/impl/ockam/input_kafka.go
index 32e2f4117b..1693b2a657 100644
--- a/internal/impl/ockam/input_kafka.go
+++ b/internal/impl/ockam/input_kafka.go
@@ -56,7 +56,7 @@ func ockamKafkaInputConfig() *service.ConfigSpec {
 			},
 			kafka.FranzConsumerFields(),
 			kafka.FranzReaderUnorderedConfigFields(),
-		)...)).
+		)...).LintRule(kafka.FranzConsumerFieldLintRules)).
 		Field(service.NewBoolField("disable_content_encryption").Default(false)).
 		Field(service.NewStringField("enrollment_ticket").Optional()).
 		Field(service.NewStringField("identity_name").Optional()).


Test patch: