Added metadata_max_age_ms option for both Consumer and Producer

faust/transport/drivers/aiokafka.py

@@ -532,6 +532,7 @@ def _create_worker_consumer(
             rebalance_timeout_ms=int(rebalance_timeout * 1000.0),
             heartbeat_interval_ms=int(conf.broker_heartbeat_interval * 1000.0),
             isolation_level=isolation_level,
+            metadata_max_age_ms=conf.consumer_metadata_max_age_ms,
             # traced_from_parent_span=self.traced_from_parent_span,
             # start_rebalancing_span=self.start_rebalancing_span,
             # start_coordinator_span=self.start_coordinator_span,
@@ -1101,6 +1102,7 @@ def _settings_default(self) -> Mapping[str, Any]:
             "partitioner": self.partitioner,
             "request_timeout_ms": int(self.request_timeout * 1000),
             "api_version": self._api_version,
+            "metadata_max_age_ms": self.app.conf.producer_metadata_max_age_ms,
         }
 
     def _settings_auth(self) -> Mapping[str, Any]:

faust/types/settings/settings.py

@@ -112,6 +112,7 @@ def __init__(
         consumer_max_fetch_size: Optional[int] = None,
         consumer_auto_offset_reset: Optional[str] = None,
         consumer_group_instance_id: Optional[str] = None,
+        consumer_metadata_max_age_ms: Optional[int] = None,
         # Topic serialization settings:
         key_serializer: CodecArg = None,
         value_serializer: CodecArg = None,
@@ -128,6 +129,7 @@ def __init__(
         producer_partitioner: SymbolArg[PartitionerT] = None,
         producer_request_timeout: Optional[Seconds] = None,
         producer_threaded: bool = False,
+        producer_metadata_max_age_ms: Optional[int] = None,
         # RPC settings:
         reply_create_topic: Optional[bool] = None,
         reply_expires: Optional[Seconds] = None,
@@ -1114,6 +1116,22 @@ def consumer_group_instance_id(self) -> str:
         each consumer instance has to have a unique id.
         """
 
+    @sections.Consumer.setting(
+        params.Int,
+        version_introduced="0.8.5",
+        env_name="CONSUMER_METADATA_MAX_AGE_MS",
+        default=5 * 60 * 1000,
+    )
+    def consumer_metadata_max_age_ms(self) -> int:
+        """Consumer metadata max age milliseconds
+
+        The period of time in milliseconds after which we force
+        a refresh of metadata even if we haven’t seen any partition
+        leadership changes to proactively discover any new brokers or partitions.
+
+        Default: 300000
+        """
+
     @sections.Serialization.setting(
         params.Codec,
         env_name="APP_KEY_SERIALIZER",
@@ -1344,6 +1362,23 @@ def producer_threaded(self) -> bool:
         send_soon function.
         """
 
+    @sections.Producer.setting(
+        params.Int,
+        version_introduced="0.8.5",
+        env_name="PRODUCER_METADATA_MAX_AGE_MS",
+        default=5 * 60 * 1000,
+    )
+    def producer_metadata_max_age_ms(self) -> int:
+        """Producer metadata max age milliseconds
+
+        The period of time in milliseconds after which we force
+        a refresh of metadata even if we haven’t seen any partition
+        leadership changes to proactively discover any new brokers or partitions.
+
+        Default: 300000
+
+        """
+
     @sections.Stream.setting(
         params.Bool,
         version_introduced="0.4.7",