librdkafka: expose retry and reconnect backoff settings (#33973)

Exposes librdkafka settings as system settings:

- `retry.backoff.ms`
- `retry.backoff.max.ms`
- `reconnect.backoff.ms`
- `reconnect.backoff.max.ms`

MZ will provide the default values, most of which are set based on the
current defaults described
[here](https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html).

I increased the default value for `reconnect.backoff.max.ms` to `30s`
based on local testing. `librdkafka` would have tried reconnecting a few
times quickly before reaching the `30s` time. If those didn't work, it's
likely that the cause is longer lived. Happy to set this to the
librdkafka default of `10s` if folks disagree.

### Motivation

exposes settings to help alleviate
MaterializeInc/database-issues#9801

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: martykulma

misc/python/materialize/mzcompose/__init__.py

@@ -578,6 +578,10 @@ def get_default_system_parameters(
     "compute_peek_stash_batch_size",
     "storage_statistics_retention_duration",
     "enable_paused_cluster_readhold_downgrade",
+    "kafka_retry_backoff",
+    "kafka_retry_backoff_max",
+    "kafka_reconnect_backoff",
+    "kafka_reconnect_backoff_max",
 ]
 
 

misc/python/materialize/parallel_workload/action.py

@@ -1512,6 +1512,10 @@ def __init__(
             "storage_statistics_retention_duration",
             "enable_paused_cluster_readhold_downgrade",
             "enable_with_ordinality_legacy_fallback",
+            "kafka_retry_backoff",
+            "kafka_retry_backoff_max",
+            "kafka_reconnect_backoff",
+            "kafka_reconnect_backoff_max",
         ]
 
     def run(self, exe: Executor) -> bool:

src/storage-types/src/connections.rs

@@ -60,7 +60,8 @@ use crate::connections::string_or_secret::StringOrSecret;
 use crate::controller::AlterError;
 use crate::dyncfgs::{
     ENFORCE_EXTERNAL_ADDRESSES, KAFKA_CLIENT_ID_ENRICHMENT_RULES,
-    KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM,
+    KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM, KAFKA_RECONNECT_BACKOFF,
+    KAFKA_RECONNECT_BACKOFF_MAX, KAFKA_RETRY_BACKOFF, KAFKA_RETRY_BACKOFF_MAX,
 };
 use crate::errors::{ContextCreationError, CsrConnectError};
 
@@ -852,6 +853,35 @@ impl KafkaConnection {
             }
         }
 
+        options.insert(
+            "retry.backoff.ms".into(),
+            KAFKA_RETRY_BACKOFF
+                .get(storage_configuration.config_set())
+                .as_millis()
+                .into(),
+        );
+        options.insert(
+            "retry.backoff.max.ms".into(),
+            KAFKA_RETRY_BACKOFF_MAX
+                .get(storage_configuration.config_set())
+                .as_millis()
+                .into(),
+        );
+        options.insert(
+            "reconnect.backoff.ms".into(),
+            KAFKA_RECONNECT_BACKOFF
+                .get(storage_configuration.config_set())
+                .as_millis()
+                .into(),
+        );
+        options.insert(
+            "reconnect.backoff.max.ms".into(),
+            KAFKA_RECONNECT_BACKOFF_MAX
+                .get(storage_configuration.config_set())
+                .as_millis()
+                .into(),
+        );
+
         let mut config = mz_kafka_util::client::create_new_client_config(
             storage_configuration
                 .connection_context

src/storage-types/src/dyncfgs.rs

@@ -129,6 +129,40 @@ pub const KAFKA_BUFFERED_EVENT_RESIZE_THRESHOLD_ELEMENTS: Config<usize> = Config
         most this number of elements.",
 );
 
+/// Sets retry.backoff.ms in librdkafka for sources and sinks.
+/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
+pub const KAFKA_RETRY_BACKOFF: Config<Duration> = Config::new(
+    "kafka_retry_backoff",
+    Duration::from_millis(100),
+    "Sets retry.backoff.ms in librdkafka for sources and sinks.",
+);
+
+/// Sets retry.backoff.max.ms in librdkafka for sources and sinks.
+/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
+pub const KAFKA_RETRY_BACKOFF_MAX: Config<Duration> = Config::new(
+    "kafka_retry_backoff_max",
+    Duration::from_secs(1),
+    "Sets retry.backoff.max.ms in librdkafka for sources and sinks.",
+);
+
+/// Sets reconnect.backoff.ms in librdkafka for sources and sinks.
+/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
+pub const KAFKA_RECONNECT_BACKOFF: Config<Duration> = Config::new(
+    "kafka_reconnect_backoff",
+    Duration::from_millis(100),
+    "Sets reconnect.backoff.ms in librdkafka for sources and sinks.",
+);
+
+/// Sets reconnect.backoff.max.ms in librdkafka for sources and sinks.
+/// We default to 30s instead of 10s to avoid constant reconnection attempts in the event of
+/// auth changes or unavailability.
+/// See <https://docs.confluent.io/platform/current/clients/librdkafka/html/md_CONFIGURATION.html>
+pub const KAFKA_RECONNECT_BACKOFF_MAX: Config<Duration> = Config::new(
+    "kafka_reconnect_backoff_max",
+    Duration::from_secs(30),
+    "Sets reconnect.backoff.max.ms in librdkafka for sources and sinks.",
+);
+
 // MySQL
 
 /// Replication heartbeat interval requested from the MySQL server.
@@ -301,6 +335,10 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM)
         .add(&KAFKA_METADATA_FETCH_INTERVAL)
         .add(&KAFKA_POLL_MAX_WAIT)
+        .add(&KAFKA_RETRY_BACKOFF)
+        .add(&KAFKA_RETRY_BACKOFF_MAX)
+        .add(&KAFKA_RECONNECT_BACKOFF)
+        .add(&KAFKA_RECONNECT_BACKOFF_MAX)
         .add(&MYSQL_OFFSET_KNOWN_INTERVAL)
         .add(&MYSQL_REPLICATION_HEARTBEAT_INTERVAL)
         .add(&ORE_OVERFLOWING_BEHAVIOR)