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 26a94ed4 Replace expressions with Single Abstract Methods (SAMs) (#577)
26a94ed4 is described below
commit 26a94ed4119de1e53fe4e075ffb5cdaef83dea7e
Author: Philippus Baalman <[email protected]>
AuthorDate: Fri Jun 26 16:01:26 2026 +0200
Replace expressions with Single Abstract Methods (SAMs) (#577)
---
.../kafka/benchmarks/KafkaProducerBenchmarks.scala | 4 +---
.../benchmarks/KafkaTransactionBenchmarks.scala | 5 +----
.../kafka/benchmarks/PerfFixtureHelpers.scala | 24 ++++++++++------------
.../pekko/kafka/internal/SingleSourceLogic.scala | 9 ++++----
.../pekko/kafka/internal/SubSourceLogic.scala | 8 +++-----
.../kafka/internal/TransactionalSources.scala | 10 ++-------
.../apache/pekko/kafka/scaladsl/SendProducer.scala | 12 +++++------
7 files changed, 27 insertions(+), 45 deletions(-)
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerBenchmarks.scala
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerBenchmarks.scala
index 1b484c4d..69b10de4 100644
---
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerBenchmarks.scala
+++
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaProducerBenchmarks.scala
@@ -37,9 +37,7 @@ object KafkaProducerBenchmarks extends LazyLogging {
val partition: Int = (i % fixture.numberOfPartitions).toInt
producer.send(
new ProducerRecord[Array[Byte], String](fixture.topic, partition,
null, msg),
- new Callback {
- override def onCompletion(metadata: RecordMetadata, exception:
Exception): Unit = meter.mark()
- })
+ (_: RecordMetadata, _: Exception) => meter.mark())
if (i % logStep == 0) {
val lastPartEnd = System.nanoTime()
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.scala
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.scala
index 94d18090..c0c0e847 100644
---
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.scala
+++
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.scala
@@ -72,10 +72,7 @@ object KafkaTransactionBenchmarks extends LazyLogging {
lastProcessedOffset = record.offset()
val producerRecord = new ProducerRecord(fixture.sinkTopic,
record.partition(), record.key(), record.value())
- producer.send(producerRecord,
- new Callback {
- override def onCompletion(metadata: RecordMetadata, exception:
Exception): Unit = meter.mark()
- })
+ producer.send(producerRecord, (_: RecordMetadata, _: Exception) =>
meter.mark())
if (lastProcessedOffset % loggedStep == 0)
logger.info(
s"Transformed $lastProcessedOffset elements to Kafka (${100 *
lastProcessedOffset / msgCount}%)")
diff --git
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PerfFixtureHelpers.scala
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PerfFixtureHelpers.scala
index 7653ddda..b55c3235 100644
---
a/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PerfFixtureHelpers.scala
+++
b/benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PerfFixtureHelpers.scala
@@ -97,20 +97,18 @@ private[benchmarks] trait PerfFixtureHelpers extends
LazyLogging {
val partition: Int = (i % ft.numberOfPartitions).toInt
producer.send(
new ProducerRecord[Array[Byte], String](ft.topic, partition, null,
msg),
- new Callback {
- override def onCompletion(recordMetadata: RecordMetadata, e:
Exception): Unit =
- if (e == null) {
- if (i % loggedStep == 0)
- logger.info(s"Written $i elements to Kafka (${100 * i /
ft.msgCount}%)")
- if (i >= ft.msgCount - 1 &&
!lastElementStoredPromise.isCompleted)
- lastElementStoredPromise.success(())
- } else {
- if (!lastElementStoredPromise.isCompleted) {
- e.printStackTrace()
- lastElementStoredPromise.failure(e)
- }
+ (_: RecordMetadata, e: Exception) =>
+ if (e == null) {
+ if (i % loggedStep == 0)
+ logger.info(s"Written $i elements to Kafka (${100 * i /
ft.msgCount}%)")
+ if (i >= ft.msgCount - 1 &&
!lastElementStoredPromise.isCompleted)
+ lastElementStoredPromise.success(())
+ } else {
+ if (!lastElementStoredPromise.isCompleted) {
+ e.printStackTrace()
+ lastElementStoredPromise.failure(e)
}
- })
+ })
}
}
val lastElementStoredFuture = lastElementStoredPromise.future
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
index e2b832c6..283a0c4c 100644
---
a/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
+++
b/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
@@ -79,11 +79,10 @@ import scala.concurrent.{ Future, Promise }
}
protected def stopConsumerActor(): Unit =
- materializer.scheduleOnce(settings.stopTimeout,
- new Runnable {
- override def run(): Unit =
- consumerActor.tell(KafkaConsumerActor.Internal.StopFromStage(id),
sourceActor.ref)
- })
+ materializer.scheduleOnce(
+ settings.stopTimeout,
+ () => consumerActor.tell(KafkaConsumerActor.Internal.StopFromStage(id),
sourceActor.ref)
+ )
/**
* Opportunity for subclasses to add a different logic to the partition
assignment callbacks.
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
index 30e7487d..c09acb31 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
@@ -283,10 +283,8 @@ private class SubSourceLogic[K, V, Msg](
}
materializer.scheduleOnce(
settings.stopTimeout,
- new Runnable {
- override def run(): Unit =
- consumerActor.tell(KafkaConsumerActor.Internal.StopFromStage(id),
sourceActor.ref)
- })
+ () => consumerActor.tell(KafkaConsumerActor.Internal.StopFromStage(id),
sourceActor.ref)
+ )
}
/**
@@ -420,7 +418,7 @@ private abstract class SubSourceStageLogic[K, V, Msg](
log.info("Starting. Partition {}", tp)
subSourceActor = getStageActor(messageHandling)
subSourceActor.watch(consumerActor)
- val controlAndActor = ControlAndStageActor(this.asInstanceOf[Control],
subSourceActor.ref)
+ val controlAndActor = ControlAndStageActor(this, subSourceActor.ref)
val started = SubSourceStageLogicControl(tp, controlAndActor,
filterRevokedPartitionsCB)
subSourceStartedCb.invoke(started)
consumerActor.tell(RegisterSubStage(requestMessages.tps),
subSourceActor.ref)
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 22e5c660..c8ee5b43 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
@@ -133,10 +133,7 @@ private[internal] abstract class
TransactionalSourceLogic[K, V, Msg](shape: Sour
log.debug(s"Draining partitions {}", partitions)
materializer.scheduleOnce(
consumerSettings.drainingCheckInterval,
- new Runnable {
- override def run(): Unit =
- sourceActor.ref.tell(Drain(partitions, ack.orElse(Some(sender)),
msg), sourceActor.ref)
- })
+ () => sourceActor.ref.tell(Drain(partitions,
ack.orElse(Some(sender)), msg), sourceActor.ref))
}
}
@@ -442,10 +439,7 @@ private final class TransactionalSubSourceStageLogic[K, V](
log.debug(s"Draining partitions {}", partitions)
materializer.scheduleOnce(
consumerSettings.drainingCheckInterval,
- new Runnable {
- override def run(): Unit =
- subSourceActor.ref.tell(Drain(partitions,
ack.orElse(Some(sender)), msg), stageActor.ref)
- })
+ () => subSourceActor.ref.tell(Drain(partitions,
ack.orElse(Some(sender)), msg), stageActor.ref))
}
case (sender, DrainingComplete) =>
completeStage()
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
index 2de38dd0..e2d4cb22 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/SendProducer.scala
@@ -78,13 +78,11 @@ final class SendProducer[K, V] private (val settings:
ProducerSettings[K, V], sy
val result = Promise[R]()
producer.send(
record,
- new Callback {
- override def onCompletion(metadata: RecordMetadata, exception:
Exception): Unit = {
- if (exception == null)
- result.success(success(metadata))
- else
- result.failure(exception)
- }
+ (metadata: RecordMetadata, exception: Exception) => {
+ if (exception == null)
+ result.success(success(metadata))
+ else
+ result.failure(exception)
})
result.future
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
