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 7f8690ca use lazy slf4j placeholder logging in main sources (#653)
7f8690ca is described below
commit 7f8690cafbc6d3ff2f408d4b295a16392a184ba8
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 22:19:33 2026 +0100
use lazy slf4j placeholder logging in main sources (#653)
Motivation:
Several log calls in main sources built their message with Scala string
interpolation, so the message was formatted even when the level was
disabled. A few others used an `s` prefix with no substitutions at all.
Modification:
Converted the slf4j call sites in `testkit` to `{}` placeholders so the
arguments are only rendered when the level is enabled - this matters most
in `KafkaTestKitChecks.periodicalCheck`, which called
`ex.getClass.getName` and `ex.getMessage` on every failed poll of a retry
loop. Dropped the redundant `s` prefix on interpolation-free literals in
`TransactionalSources` and `KafkaConsumerBenchmarks`. Replaced the manual
`isDebugEnabled` guard in `KafkaConsumerActor` with a placeholder call,
since the logging adapter already checks the level before formatting.
Left unchanged: `KafkaClusterSharding` and `KafkaContainerCluster` already
use placeholders, the `benchmarks` module's scala-logging `LazyLogging`
macros already wrap interpolation in a level check, and the
`Partitions drained` calls in `TransactionalSources` would evaluate
`mkString` eagerly either way and would change their rendered output if
the collection were passed through the adapter instead.
Result:
No message or argument is rendered unless the log level is enabled, and no
call site pays for interpolation it does not use. Log output text is
unchanged.
Tests:
- sbt "core/compile" "testkit/compile" "benchmarks/compile" - success
(Scala 2.13.18)
- sbt "++3.3.8 core/compile" "++3.3.8 testkit/compile" - success (Scala
3.3.8)
- sbt "core/scalafmtCheck" "testkit/scalafmtCheck"
"benchmarks/scalafmtCheck" - success
- sbt "core/mimaReportBinaryIssues" "testkit/mimaReportBinaryIssues" -
success
- No directional test added: this changes only the shape of logging calls,
with no behavior or API change.
References:
None - logging cleanup found while reviewing main sources
---
.../org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala | 6 +++---
.../scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala | 3 +--
.../org/apache/pekko/kafka/internal/TransactionalSources.scala | 4 ++--
.../org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala | 4 ++--
.../apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala | 2 +-
.../scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala | 2 +-
6 files changed, 10 insertions(+), 11 deletions(-)
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala
index 2dde81a4..c0f3acd3 100644
---
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala
+++
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala
@@ -38,7 +38,7 @@ object KafkaConsumerBenchmarks extends LazyLogging {
if (readSoFar >= readLimit)
readSoFar
else {
- logger.debug(s"Polling")
+ logger.debug("Polling")
meter.mark()
pollInLoop(readLimit, readSoFar + 1)
}
@@ -57,7 +57,7 @@ object KafkaConsumerBenchmarks extends LazyLogging {
if (readSoFar >= readLimit)
readSoFar
else {
- logger.debug(s"Polling")
+ logger.debug("Polling")
val records = consumer.poll(pollTimeoutMs)
val recordCount = records.count()
records.iterator().asScala.toList // ensure records are processed
@@ -182,7 +182,7 @@ object KafkaConsumerBenchmarks extends LazyLogging {
if (readSoFar >= readLimit)
readSoFar
else {
- logger.debug(s"Polling")
+ logger.debug("Polling")
consumer.pause(assignment)
val records = consumer.poll(pollTimeoutMs)
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
index 6faa41ae..d4a8fe80 100644
---
a/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
+++
b/core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala
@@ -430,8 +430,7 @@ import scala.util.control.NonFatal
commitRefreshing = CommitRefreshing(settings.commitRefreshInterval,
progressTrackingFactory)
resetProtection = ConsumerResetProtection(log,
settings.resetProtectionSettings, progressTrackingFactory)
try {
- if (log.isDebugEnabled)
- log.debug(s"Creating Kafka consumer with ${settings.toString}")
+ log.debug("Creating Kafka consumer with {}", settings)
consumer = settings.consumerFactory.apply(settings)
context.become(regularReceive)
unstashAll()
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
index 907f88d9..ecc8a759 100644
---
a/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
+++
b/core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala
@@ -130,7 +130,7 @@ private[internal] abstract class
TransactionalSourceLogic[K, V, Msg](shape: Sour
log.debug(s"Partitions drained ${partitions.mkString(",")}")
ack.getOrElse(sender).tell(msg, sourceActor.ref)
} else {
- log.debug(s"Draining partitions {}", partitions)
+ log.debug("Draining partitions {}", partitions)
materializer.scheduleOnce(
consumerSettings.drainingCheckInterval,
() => sourceActor.ref.tell(Drain(partitions,
ack.orElse(Some(sender)), msg), sourceActor.ref))
@@ -428,7 +428,7 @@ private final class TransactionalSubSourceStageLogic[K, V](
log.debug(s"Partitions drained ${partitions.mkString(",")}")
ack.getOrElse(sender) ! msg
} else {
- log.debug(s"Draining partitions {}", partitions)
+ log.debug("Draining partitions {}", partitions)
materializer.scheduleOnce(
consumerSettings.drainingCheckInterval,
() => subSourceActor.ref.tell(Drain(partitions,
ack.orElse(Some(sender)), msg), stageActor.ref))
diff --git
a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
index 72669f2d..623367cd 100644
---
a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
+++
b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala
@@ -183,12 +183,12 @@ trait KafkaTestKit {
}
def sleepMillis(ms: Long, msg: String): Unit = {
- log.debug(s"sleeping $ms ms $msg")
+ log.debug("sleeping {} ms {}", ms, msg)
Thread.sleep(ms)
}
def sleepSeconds(s: Int, msg: String): Unit = {
- log.debug(s"sleeping $s s $msg")
+ log.debug("sleeping {} s {}", s, msg)
Thread.sleep(s * 1000L)
}
}
diff --git
a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
index 93478018..952cf049 100644
---
a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
+++
b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala
@@ -59,7 +59,7 @@ object KafkaTestKitChecks {
@tailrec def check(triesLeft: Int): Unit =
Try(predicate(data())).recover {
case ex =>
- log.debug(s"Ignoring [${ex.getClass.getName}: ${ex.getMessage}]
while waiting for desired state")
+ log.debug("Ignoring [{}: {}] while waiting for desired state",
ex.getClass.getName, ex.getMessage)
false
} match {
case Success(false) if triesLeft > 0 =>
diff --git
a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
index 016dee93..a47ef989 100644
---
a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
+++
b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/scaladsl/KafkaSpec.scala
@@ -73,7 +73,7 @@ abstract class KafkaSpec(_kafkaPort: Int, val zooKeeperPort:
Int, actorSystem: A
}
def sleep(time: FiniteDuration, msg: String = ""): Unit = {
- log.debug(s"sleeping $time $msg")
+ log.debug("sleeping {} {}", time, msg)
Thread.sleep(time.toMillis)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
