This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new afb18032 add require guards for config settings (#627)
afb18032 is described below
commit afb18032102dc072ae2eb090432c9cd41777caf5
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 17 10:25:44 2026 +0100
add require guards for config settings (#627)
---
.../scala/org/apache/pekko/kafka/CommitterSettings.scala | 3 +++
.../scala/org/apache/pekko/kafka/ConsumerSettings.scala | 16 ++++++++++++++++
.../scala/org/apache/pekko/kafka/ProducerSettings.scala | 3 +++
3 files changed, 22 insertions(+)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
b/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
index 9ba19bd8..f720f08c 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/CommitterSettings.scala
@@ -135,6 +135,9 @@ object CommitterSettings {
val maxBatch = config.getLong("max-batch")
val maxInterval = config.getDuration("max-interval",
TimeUnit.MILLISECONDS).millis
val parallelism = config.getInt("parallelism")
+ require(maxBatch > 0, "max-batch must be positive")
+ require(maxInterval > Duration.Zero, "max-interval must be positive")
+ require(parallelism > 0, "parallelism must be positive")
val delivery = CommitDelivery.valueOf(config.getString("delivery"))
val when = CommitWhen.valueOf(config.getString("when"))
new CommitterSettings(maxBatch, maxInterval, parallelism, delivery, when)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
b/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
index 0d4e952a..1a3d25db 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
@@ -94,6 +94,22 @@ object ConsumerSettings {
val drainingCheckInterval =
config.getDuration("eos-draining-check-interval").toScala
val connectionCheckerSettings =
ConnectionCheckerSettings(config.getConfig(ConnectionCheckerSettings.configPath))
val partitionHandlerWarning =
config.getDuration("partition-handler-warning").toScala
+ require(pollInterval > Duration.Zero, "poll-interval must be positive")
+ require(pollTimeout > Duration.Zero, "poll-timeout must be positive")
+ require(stopTimeout > Duration.Zero, "stop-timeout must be positive")
+ require(closeTimeout > Duration.Zero, "close-timeout must be positive")
+ require(commitTimeout > Duration.Zero, "commit-timeout must be positive")
+ require(commitTimeWarning > Duration.Zero, "commit-time-warning must be
positive")
+ commitRefreshInterval match {
+ case fd: FiniteDuration => require(fd > Duration.Zero,
"commit-refresh-interval must be positive when finite")
+ case _ => // infinite is valid
+ }
+ require(waitClosePartition > Duration.Zero, "wait-close-partition must be
positive")
+ require(positionTimeout > Duration.Zero, "position-timeout must be
positive")
+ require(offsetForTimesTimeout > Duration.Zero, "offset-for-times-timeout
must be positive")
+ require(metadataRequestTimeout > Duration.Zero, "metadata-request-timeout
must be positive")
+ require(drainingCheckInterval > Duration.Zero,
"eos-draining-check-interval must be positive")
+ require(partitionHandlerWarning > Duration.Zero,
"partition-handler-warning must be positive")
val resetProtectionThreshold = OffsetResetProtectionSettings(
config.getConfig(OffsetResetProtectionSettings.configPath))
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
b/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
index 2f63d4da..a93b7c01 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
@@ -82,6 +82,9 @@ object ProducerSettings {
val parallelism = config.getInt("parallelism")
val dispatcher = config.getString("use-dispatcher")
val eosCommitInterval = config.getDuration("eos-commit-interval").toScala
+ require(closeTimeout > Duration.Zero, "close-timeout must be positive")
+ require(parallelism > 0, "parallelism must be positive")
+ require(eosCommitInterval > Duration.Zero, "eos-commit-interval must be
positive")
new ProducerSettings[K, V](
properties,
keySerializer,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
