You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by md...@apache.org on 2023/05/10 14:19:38 UTC
[incubator-pekko-connectors-kafka] branch main updated: Apply compat changes from latest Pekko
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new e2c660ef Apply compat changes from latest Pekko
e2c660ef is described below
commit e2c660ef5208569e34b5be601c8068e6b946f19d
Author: Matthew de Detrich <ma...@aiven.io>
AuthorDate: Wed May 10 13:01:03 2023 +0200
Apply compat changes from latest Pekko
---
.../kafka/cluster/sharding/KafkaClusterSharding.scala | 7 +++----
.../org/apache/pekko/kafka/ConsumerSettings.scala | 14 +++++++-------
.../org/apache/pekko/kafka/ProducerSettings.scala | 14 +++++++-------
.../pekko/kafka/internal/ControlImplementations.scala | 12 ++++++------
.../apache/pekko/kafka/internal/MessageBuilder.scala | 6 +++---
.../org/apache/pekko/kafka/javadsl/Committer.scala | 8 ++++----
.../org/apache/pekko/kafka/javadsl/Consumer.scala | 10 +++++-----
.../apache/pekko/kafka/javadsl/DiscoverySupport.scala | 8 ++++----
.../apache/pekko/kafka/javadsl/MetadataClient.scala | 18 +++++++++---------
.../org/apache/pekko/kafka/javadsl/Producer.scala | 8 ++++----
.../org/apache/pekko/kafka/javadsl/SendProducer.scala | 9 ++++-----
.../org/apache/pekko/kafka/javadsl/Transactional.scala | 7 +++----
project/Versions.scala | 2 +-
.../pekko/kafka/testkit/javadsl/BaseKafkaTest.java | 5 ++---
.../kafka/testkit/internal/TestcontainersKafka.scala | 7 +++----
.../org/apache/pekko/kafka/javadsl/ControlSpec.scala | 18 +++++++++---------
16 files changed, 74 insertions(+), 79 deletions(-)
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala b/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala
index 56545cc8..3a323218 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala
@@ -35,11 +35,10 @@ import org.apache.kafka.common.utils.Utils
import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContextExecutor, Future }
import scala.util.{ Failure, Success }
+import pekko.util.FutureConverters._
import pekko.util.JavaDurationConverters._
import org.slf4j.LoggerFactory
-import scala.compat.java8.FutureConverters._
-
/**
* API MAY CHANGE
*
@@ -91,7 +90,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
settings: ConsumerSettings[_, _]): CompletionStage[KafkaShardingMessageExtractor[M]] =
getPartitionCount(topic, timeout.asScala, settings)
.map(new KafkaShardingMessageExtractor[M](_))(system.dispatcher)
- .toJava
+ .asJava
/**
* API MAY CHANGE
@@ -157,7 +156,7 @@ final class KafkaClusterSharding(system: ExtendedActorSystem) extends Extension
getPartitionCount(topic, timeout.asScala, settings)
.map(partitions => new KafkaShardingNoEnvelopeExtractor[M](partitions, e => entityIdExtractor.apply(e)))(
system.dispatcher)
- .toJava
+ .asJava
/**
* API MAY CHANGE
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 26729e9f..c6842b7c 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala
@@ -22,12 +22,12 @@ import pekko.annotation.InternalApi
import pekko.kafka.internal._
import pekko.util.JavaDurationConverters._
import pekko.util.ccompat.JavaConverters._
+import pekko.util.OptionConverters._
+import pekko.util.FutureConverters._
import com.typesafe.config.Config
import org.apache.kafka.clients.consumer.{ Consumer, ConsumerConfig, KafkaConsumer }
import org.apache.kafka.common.serialization.Deserializer
-import scala.compat.java8.OptionConverters._
-import scala.compat.java8.FutureConverters._
import scala.concurrent.{ ExecutionContext, Future }
import scala.concurrent.duration._
@@ -165,7 +165,7 @@ object ConsumerSettings {
system: pekko.actor.ActorSystem,
keyDeserializer: Optional[Deserializer[K]],
valueDeserializer: Optional[Deserializer[V]]): ConsumerSettings[K, V] =
- apply(system, keyDeserializer.asScala, valueDeserializer.asScala)
+ apply(system, keyDeserializer.toScala, valueDeserializer.toScala)
/**
* Java API: Create settings from the default configuration
@@ -178,7 +178,7 @@ object ConsumerSettings {
system: pekko.actor.ClassicActorSystemProvider,
keyDeserializer: Optional[Deserializer[K]],
valueDeserializer: Optional[Deserializer[V]]): ConsumerSettings[K, V] =
- apply(system, keyDeserializer.asScala, valueDeserializer.asScala)
+ apply(system, keyDeserializer.toScala, valueDeserializer.toScala)
/**
* Java API: Create settings from a configuration with the same layout as
@@ -189,7 +189,7 @@ object ConsumerSettings {
config: Config,
keyDeserializer: Optional[Deserializer[K]],
valueDeserializer: Optional[Deserializer[V]]): ConsumerSettings[K, V] =
- apply(config, keyDeserializer.asScala, valueDeserializer.asScala)
+ apply(config, keyDeserializer.toScala, valueDeserializer.toScala)
/**
* Java API: Create settings from the default configuration
@@ -516,7 +516,7 @@ class ConsumerSettings[K, V] @InternalApi private[kafka] (
def withEnrichCompletionStage(
value: java.util.function.Function[ConsumerSettings[K, V], CompletionStage[ConsumerSettings[K, V]]])
: ConsumerSettings[K, V] =
- copy(enrichAsync = Some((s: ConsumerSettings[K, V]) => value.apply(s).toScala))
+ copy(enrichAsync = Some((s: ConsumerSettings[K, V]) => value.apply(s).asScala))
/**
* Replaces the default Kafka consumer creation logic.
@@ -622,7 +622,7 @@ class ConsumerSettings[K, V] @InternalApi private[kafka] (
* (without blocking for `enriched`).
*/
def createKafkaConsumerCompletionStage(executor: Executor): CompletionStage[Consumer[K, V]] =
- enriched.map(consumerFactory)(ExecutionContext.fromExecutor(executor)).toJava
+ enriched.map(consumerFactory)(ExecutionContext.fromExecutor(executor)).asJava
override def toString: String = {
val kafkaClients = properties.toSeq
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 816768d7..75a48c4c 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
@@ -24,13 +24,13 @@ import com.typesafe.config.Config
import org.apache.kafka.clients.producer.{ KafkaProducer, Producer, ProducerConfig }
import org.apache.kafka.common.serialization.Serializer
-import scala.compat.java8.OptionConverters._
import scala.concurrent.duration._
import pekko.util.ccompat.JavaConverters._
+import pekko.util.FutureConverters._
import pekko.util.JavaDurationConverters._
+import pekko.util.OptionConverters._
import scala.concurrent.{ ExecutionContext, Future }
-import scala.compat.java8.FutureConverters._
object ProducerSettings {
@@ -140,7 +140,7 @@ object ProducerSettings {
system: org.apache.pekko.actor.ActorSystem,
keySerializer: Optional[Serializer[K]],
valueSerializer: Optional[Serializer[V]]): ProducerSettings[K, V] =
- apply(system, keySerializer.asScala, valueSerializer.asScala)
+ apply(system, keySerializer.toScala, valueSerializer.toScala)
/**
* Java API: Create settings from the default configuration
@@ -153,7 +153,7 @@ object ProducerSettings {
system: org.apache.pekko.actor.ClassicActorSystemProvider,
keySerializer: Optional[Serializer[K]],
valueSerializer: Optional[Serializer[V]]): ProducerSettings[K, V] =
- apply(system, keySerializer.asScala, valueSerializer.asScala)
+ apply(system, keySerializer.toScala, valueSerializer.toScala)
/**
* Java API: Create settings from a configuration with the same layout as
@@ -164,7 +164,7 @@ object ProducerSettings {
config: Config,
keySerializer: Optional[Serializer[K]],
valueSerializer: Optional[Serializer[V]]): ProducerSettings[K, V] =
- apply(config, keySerializer.asScala, valueSerializer.asScala)
+ apply(config, keySerializer.toScala, valueSerializer.toScala)
/**
* Java API: Create settings from the default configuration
@@ -348,7 +348,7 @@ class ProducerSettings[K, V] @InternalApi private[kafka] (
def withEnrichCompletionStage(
value: java.util.function.Function[ProducerSettings[K, V], CompletionStage[ProducerSettings[K, V]]])
: ProducerSettings[K, V] =
- copy(enrichAsync = Some((s: ProducerSettings[K, V]) => value.apply(s).toScala))
+ copy(enrichAsync = Some((s: ProducerSettings[K, V]) => value.apply(s).asScala))
/**
* Replaces the default Kafka producer creation logic with an external producer. This will also set
@@ -460,5 +460,5 @@ class ProducerSettings[K, V] @InternalApi private[kafka] (
* @param executor Executor for asynchronous producer creation
*/
def createKafkaProducerCompletionStage(executor: Executor): CompletionStage[Producer[K, V]] =
- createKafkaProducerAsync()(ExecutionContext.fromExecutor(executor)).toJava
+ createKafkaProducerAsync()(ExecutionContext.fromExecutor(executor)).asJava
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala
index 57b2c5b0..70ab3f5f 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/ControlImplementations.scala
@@ -25,10 +25,10 @@ import pekko.kafka.{ javadsl, scaladsl }
import pekko.stream.SourceShape
import pekko.stream.stage.GraphStageLogic
import pekko.util.ccompat.JavaConverters._
+import pekko.util.FutureConverters._
import pekko.util.Timeout
import org.apache.kafka.common.{ Metric, MetricName }
-import scala.compat.java8.FutureConverters.{ CompletionStageOps, FutureOps }
import scala.concurrent.{ ExecutionContext, Future, Promise }
private object PromiseControl {
@@ -105,17 +105,17 @@ private trait MetricsControl extends scaladsl.Consumer.Control {
@InternalApi
final private[kafka] class ConsumerControlAsJava(underlying: scaladsl.Consumer.Control)
extends javadsl.Consumer.Control {
- override def stop(): CompletionStage[Done] = underlying.stop().toJava
+ override def stop(): CompletionStage[Done] = underlying.stop().asJava
- override def shutdown(): CompletionStage[Done] = underlying.shutdown().toJava
+ override def shutdown(): CompletionStage[Done] = underlying.shutdown().asJava
override def drainAndShutdown[T](streamCompletion: CompletionStage[T], ec: Executor): CompletionStage[T] =
- underlying.drainAndShutdown(streamCompletion.toScala)(ExecutionContext.fromExecutor(ec)).toJava
+ underlying.drainAndShutdown(streamCompletion.asScala)(ExecutionContext.fromExecutor(ec)).asJava
- override def isShutdown: CompletionStage[Done] = underlying.isShutdown.toJava
+ override def isShutdown: CompletionStage[Done] = underlying.isShutdown.asJava
override def getMetrics: CompletionStage[java.util.Map[MetricName, Metric]] =
- underlying.metrics.map(_.asJava)(ExecutionContexts.parasitic).toJava
+ underlying.metrics.map(_.asJava)(ExecutionContexts.parasitic).asJava
}
/** Internal API */
diff --git a/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala b/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala
index af5b7748..cae6c6f9 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala
@@ -29,11 +29,11 @@ import pekko.kafka.ConsumerMessage.{
}
import pekko.util.ccompat._
import pekko.util.ccompat.JavaConverters._
+import pekko.util.FutureConverters._
import org.apache.kafka.clients.consumer.{ ConsumerRecord, OffsetAndMetadata }
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.requests.OffsetFetchResponse
-import scala.compat.java8.FutureConverters.FutureOps
import scala.concurrent.Future
/** Internal API */
@@ -144,7 +144,7 @@ private[kafka] trait OffsetContextBuilder[K, V]
override val metadata: String)(
val committer: KafkaAsyncConsumerCommitterRef) extends CommittableOffsetMetadata {
override def commitScaladsl(): Future[Done] = commitInternal()
- override def commitJavadsl(): CompletionStage[Done] = commitInternal().toJava
+ override def commitJavadsl(): CompletionStage[Done] = commitInternal().asJava
override def commitInternal(): Future[Done] = KafkaAsyncConsumerCommitterRef.commit(this)
override val batchSize: Long = 1
}
@@ -243,7 +243,7 @@ private[kafka] final class CommittableOffsetBatchImpl(
new CommittableOffsetBatchImpl(newOffsets, newCommitters, newOffsets.size.toLong)
}
- override def commitJavadsl(): CompletionStage[Done] = commitInternal().toJava
+ override def commitJavadsl(): CompletionStage[Done] = commitInternal().asJava
/**
* @return true if the batch contains no commits.
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala
index 99febc06..8d7c2eb0 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Committer.scala
@@ -13,6 +13,7 @@
*/
package org.apache.pekko.kafka.javadsl
+
import java.util.concurrent.CompletionStage
import org.apache.pekko
@@ -22,8 +23,7 @@ import pekko.{ Done, NotUsed }
import pekko.kafka.ConsumerMessage.{ Committable, CommittableOffsetBatch }
import pekko.kafka.{ scaladsl, CommitterSettings }
import pekko.stream.javadsl.{ Flow, FlowWithContext, Sink }
-
-import scala.compat.java8.FutureConverters.FutureOps
+import pekko.util.FutureConverters._
object Committer {
@@ -54,7 +54,7 @@ object Committer {
* Batches offsets and commits them to Kafka.
*/
def sink[C <: Committable](settings: CommitterSettings): Sink[C, CompletionStage[Done]] =
- scaladsl.Committer.sink(settings).mapMaterializedValue(_.toJava).asJava
+ scaladsl.Committer.sink(settings).mapMaterializedValue(_.asJava).asJava
/**
* API MAY CHANGE
@@ -68,6 +68,6 @@ object Committer {
.Flow[Pair[E, C]]
.map(_.toScala)
.toMat(scaladsl.Committer.sinkWithOffsetContext(settings))(pekko.stream.scaladsl.Keep.right)
- .mapMaterializedValue[CompletionStage[Done]](_.toJava)
+ .mapMaterializedValue[CompletionStage[Done]](_.asJava)
.asJava[Pair[E, C]]
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala
index 0043ff8c..f737e6b4 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala
@@ -26,11 +26,11 @@ import pekko.kafka._
import pekko.kafka.internal.{ ConsumerControlAsJava, SourceWithOffsetContext }
import pekko.stream.javadsl.{ Source, SourceWithContext }
import pekko.{ Done, NotUsed }
+import pekko.util.FutureConverters._
import pekko.util.ccompat.JavaConverters._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.{ Metric, MetricName, TopicPartition }
-import scala.compat.java8.FutureConverters._
import scala.concurrent.duration.FiniteDuration
/**
@@ -286,7 +286,7 @@ object Consumer {
settings,
subscription,
(tps: Set[TopicPartition]) =>
- getOffsetsOnAssign(tps.asJava).toScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
+ getOffsetsOnAssign(tps.asJava).asScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
_ => ())
.map {
case (tp, source) => Pair(tp, source.asJava)
@@ -315,7 +315,7 @@ object Consumer {
settings,
subscription,
(tps: Set[TopicPartition]) =>
- getOffsetsOnAssign(tps.asJava).toScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
+ getOffsetsOnAssign(tps.asJava).asScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
(tps: Set[TopicPartition]) => onRevoke.accept(tps.asJava))
.map {
case (tp, source) => Pair(tp, source.asJava)
@@ -352,7 +352,7 @@ object Consumer {
settings,
subscription,
(tps: Set[TopicPartition]) =>
- getOffsetsOnAssign(tps.asJava).toScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
+ getOffsetsOnAssign(tps.asJava).asScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
_ => ())
.map {
case (tp, source) => Pair(tp, source.asJava)
@@ -375,7 +375,7 @@ object Consumer {
settings,
subscription,
(tps: Set[TopicPartition]) =>
- getOffsetsOnAssign(tps.asJava).toScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
+ getOffsetsOnAssign(tps.asJava).asScala.map(_.asScala.toMap)(ExecutionContexts.parasitic),
(tps: Set[TopicPartition]) => onRevoke.accept(tps.asJava))
.map {
case (tp, source) => Pair(tp, source.asJava)
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala
index 11883b87..37d4ad53 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/DiscoverySupport.scala
@@ -19,10 +19,10 @@ import java.util.concurrent.CompletionStage
import org.apache.pekko
import pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
import pekko.kafka.{ scaladsl, ConsumerSettings, ProducerSettings }
+import pekko.util.FunctionConverters._
+import pekko.util.FutureConverters._
import com.typesafe.config.Config
-import scala.compat.java8.FunctionConverters._
-import scala.compat.java8.FutureConverters
import scala.concurrent.Future
/**
@@ -43,7 +43,7 @@ object DiscoverySupport {
implicit val sys: ClassicActorSystemProvider = system
val function: ConsumerSettings[K, V] => Future[ConsumerSettings[K, V]] =
scaladsl.DiscoverySupport.consumerBootstrapServers(config)
- function.andThen(FutureConverters.toJava).asJava
+ function.andThen(_.asJava).asJava
}
// kept for bin-compatibility
@@ -66,7 +66,7 @@ object DiscoverySupport {
implicit val sys: ClassicActorSystemProvider = system
val function: ProducerSettings[K, V] => Future[ProducerSettings[K, V]] =
scaladsl.DiscoverySupport.producerBootstrapServers(config)
- function.andThen(FutureConverters.toJava).asJava
+ function.andThen(_.asJava).asJava
}
// kept for bin-compatibility
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
index 03963843..6ffed653 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
@@ -22,11 +22,11 @@ import pekko.dispatch.ExecutionContexts
import pekko.kafka.ConsumerSettings
import pekko.util.ccompat._
import pekko.util.ccompat.JavaConverters._
+import pekko.util.FutureConverters._
import pekko.util.Timeout
import org.apache.kafka.clients.consumer.OffsetAndMetadata
import org.apache.kafka.common.{ PartitionInfo, TopicPartition }
-import scala.compat.java8.FutureConverters._
import scala.concurrent.ExecutionContextExecutor
class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClient) {
@@ -38,13 +38,13 @@ class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClien
.map { beginningOffsets =>
beginningOffsets.view.mapValues(Long.box).toMap.asJava
}(ExecutionContexts.parasitic)
- .toJava
+ .asJava
def getBeginningOffsetForPartition[K, V](partition: TopicPartition): CompletionStage[java.lang.Long] =
metadataClient
.getBeginningOffsetForPartition(partition)
.map(Long.box)(ExecutionContexts.parasitic)
- .toJava
+ .asJava
def getEndOffsets(
partitions: java.util.Set[TopicPartition]): CompletionStage[java.util.Map[TopicPartition, java.lang.Long]] =
@@ -53,13 +53,13 @@ class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClien
.map { endOffsets =>
endOffsets.view.mapValues(Long.box).toMap.asJava
}(ExecutionContexts.parasitic)
- .toJava
+ .asJava
def getEndOffsetForPartition(partition: TopicPartition): CompletionStage[java.lang.Long] =
metadataClient
.getEndOffsetForPartition(partition)
.map(Long.box)(ExecutionContexts.parasitic)
- .toJava
+ .asJava
def listTopics(): CompletionStage[java.util.Map[java.lang.String, java.util.List[PartitionInfo]]] =
metadataClient
@@ -67,7 +67,7 @@ class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClien
.map { topics =>
topics.view.mapValues(partitionsInfo => partitionsInfo.asJava).toMap.asJava
}(ExecutionContexts.parasitic)
- .toJava
+ .asJava
def getPartitionsFor(topic: java.lang.String): CompletionStage[java.util.List[PartitionInfo]] =
metadataClient
@@ -75,13 +75,13 @@ class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClien
.map { partitionsInfo =>
partitionsInfo.asJava
}(ExecutionContexts.parasitic)
- .toJava
+ .asJava
@deprecated("use `getCommittedOffsets`", "2.0.3")
def getCommittedOffset(partition: TopicPartition): CompletionStage[OffsetAndMetadata] =
metadataClient
.getCommittedOffset(partition)
- .toJava
+ .asJava
def getCommittedOffsets(
partitions: java.util.Set[TopicPartition]): CompletionStage[java.util.Map[TopicPartition, OffsetAndMetadata]] =
@@ -90,7 +90,7 @@ class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClien
.map { committedOffsets =>
committedOffsets.asJava
}(ExecutionContexts.parasitic)
- .toJava
+ .asJava
def close(): Unit =
metadataClient.close()
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala
index 4aca27c0..73352d08 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Producer.scala
@@ -22,10 +22,10 @@ import pekko.kafka.ProducerMessage._
import pekko.kafka.{ scaladsl, CommitterSettings, ConsumerMessage, ProducerSettings }
import pekko.stream.javadsl.{ Flow, FlowWithContext, Sink }
import pekko.{ japi, Done, NotUsed }
+import pekko.util.FutureConverters._
import org.apache.kafka.clients.producer.ProducerRecord
import scala.annotation.nowarn
-import scala.compat.java8.FutureConverters._
/**
* Apache Pekko Stream connector for publishing messages to Kafka topics.
@@ -41,7 +41,7 @@ object Producer {
def plainSink[K, V](settings: ProducerSettings[K, V]): Sink[ProducerRecord[K, V], CompletionStage[Done]] =
scaladsl.Producer
.plainSink(settings)
- .mapMaterializedValue(_.toJava)
+ .mapMaterializedValue(_.asJava)
.asJava
/**
@@ -84,7 +84,7 @@ object Producer {
@nowarn("cat=deprecation")
val sink: Sink[IN, CompletionStage[Done]] = scaladsl.Producer
.committableSink(settings)
- .mapMaterializedValue(_.toJava)
+ .mapMaterializedValue(_.asJava)
.asJava
sink
}
@@ -136,7 +136,7 @@ object Producer {
committerSettings: CommitterSettings): Sink[IN, CompletionStage[Done]] =
scaladsl.Producer
.committableSink(producerSettings, committerSettings)
- .mapMaterializedValue(_.toJava)
+ .mapMaterializedValue(_.asJava)
.asJava
/**
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala
index 774603f0..49f95592 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/SendProducer.scala
@@ -21,10 +21,9 @@ import pekko.Done
import pekko.actor.{ ActorSystem, ClassicActorSystemProvider }
import pekko.kafka.ProducerMessage._
import pekko.kafka.{ scaladsl, ProducerSettings }
+import pekko.util.FutureConverters._
import org.apache.kafka.clients.producer.{ ProducerRecord, RecordMetadata }
-import scala.compat.java8.FutureConverters._
-
/**
* Utility class for producing to Kafka without using Apache Pekko Streams.
*/
@@ -58,18 +57,18 @@ final class SendProducer[K, V] private (underlying: scaladsl.SendProducer[K, V])
* The messages support passing through arbitrary data.
*/
def sendEnvelope[PT](envelope: Envelope[K, V, PT]): CompletionStage[Results[K, V, PT]] =
- underlying.sendEnvelope(envelope).toJava
+ underlying.sendEnvelope(envelope).asJava
/**
* Send a raw Kafka [[org.apache.kafka.clients.producer.ProducerRecord]] and complete a future with the resulting metadata.
*/
def send(record: ProducerRecord[K, V]): CompletionStage[RecordMetadata] =
- underlying.send(record).toJava
+ underlying.send(record).asJava
/**
* Close the underlying producer (depending on the "close producer on stop" setting).
*/
- def close(): CompletionStage[Done] = underlying.close().toJava
+ def close(): CompletionStage[Done] = underlying.close().asJava
override def toString: String = s"SendProducer(${underlying.settings})"
}
diff --git a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala
index fb6fe2c8..594e72b5 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/javadsl/Transactional.scala
@@ -26,10 +26,9 @@ import pekko.kafka.internal.{ ConsumerControlAsJava, TransactionalSourceWithOffs
import pekko.kafka.javadsl.Consumer.Control
import pekko.stream.javadsl._
import pekko.{ Done, NotUsed }
+import pekko.util.FutureConverters._
import org.apache.kafka.clients.consumer.ConsumerRecord
-import scala.compat.java8.FutureConverters.FutureOps
-
/**
* Apache Pekko Stream connector to support transactions between Kafka topics.
*/
@@ -97,7 +96,7 @@ object Transactional {
transactionalId: String): Sink[IN, CompletionStage[Done]] =
scaladsl.Transactional
.sink(settings, transactionalId)
- .mapMaterializedValue(_.toJava)
+ .mapMaterializedValue(_.asJava)
.asJava
/**
@@ -115,7 +114,7 @@ object Transactional {
.map(_.toScala)
.toMat(scaladsl.Transactional.sinkWithOffsetContext(settings, transactionalId))(
pekko.stream.scaladsl.Keep.right)
- .mapMaterializedValue(_.toJava)
+ .mapMaterializedValue(_.asJava)
.asJava
/**
diff --git a/project/Versions.scala b/project/Versions.scala
index 3b349025..e296514e 100644
--- a/project/Versions.scala
+++ b/project/Versions.scala
@@ -21,7 +21,7 @@ object Versions {
val Scala3 = "3.2.2"
val pekkoVersionForDocs = "current"
- val pekkoVersion = "0.0.0+26630-2c4d0ee0-SNAPSHOT"
+ val pekkoVersion = "0.0.0+26656-898c6970-SNAPSHOT"
// Keep .scala-steward.conf pin in sync
val kafkaVersion = "3.0.1"
diff --git a/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/BaseKafkaTest.java b/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/BaseKafkaTest.java
index fe01439f..8a5ad22d 100644
--- a/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/BaseKafkaTest.java
+++ b/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/BaseKafkaTest.java
@@ -36,7 +36,6 @@ import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.compat.java8.functionConverterImpls.FromJavaPredicate;
import java.time.Duration;
import java.util.Arrays;
@@ -133,7 +132,7 @@ public abstract class BaseKafkaTest extends KafkaTestKitClass {
settings().clusterTimeout(),
settings().checkInterval(),
adminClient(),
- new FromJavaPredicate<DescribeClusterResult>(predicate),
+ predicate::test,
log());
}
@@ -149,7 +148,7 @@ public abstract class BaseKafkaTest extends KafkaTestKitClass {
settings().consumerGroupTimeout(),
settings().checkInterval(),
adminClient(),
- new FromJavaPredicate<ConsumerGroupDescription>(predicate),
+ predicate::test,
log());
}
diff --git a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala
index 339555ec..3ea3df38 100644
--- a/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala
+++ b/testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala
@@ -18,12 +18,11 @@ import org.apache.pekko
import pekko.kafka.testkit.KafkaTestkitTestcontainersSettings
import pekko.kafka.testkit.scaladsl.KafkaSpec
import pekko.util.JavaDurationConverters._
+import pekko.util.OptionConverters._
import pekko.util.ccompat.JavaConverters._
import org.testcontainers.containers.GenericContainer
import org.testcontainers.utility.DockerImageName
-import scala.compat.java8.OptionConverters._
-
object TestcontainersKafka {
trait Spec extends KafkaSpec {
private var cluster: KafkaContainerCluster = _
@@ -61,12 +60,12 @@ object TestcontainersKafka {
def schemaRegistryContainer: Option[SchemaRegistryContainer] = {
requireStarted()
- cluster.getSchemaRegistry.asScala
+ cluster.getSchemaRegistry.toScala
}
def getSchemaRegistryUrl: String = {
requireStarted()
- cluster.getSchemaRegistry.asScala
+ cluster.getSchemaRegistry.toScala
.map(_.getSchemaRegistryUrl)
.getOrElse(
throw new RuntimeException("Did you enable schema registry in your KafkaTestkitTestcontainersSettings?"))
diff --git a/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala b/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala
index a074a833..541b0f76 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/javadsl/ControlSpec.scala
@@ -22,12 +22,12 @@ import org.apache.pekko
import pekko.Done
import pekko.kafka.internal.ConsumerControlAsJava
import pekko.kafka.tests.scaladsl.LogCapturing
+import pekko.util.FutureConverters._
import org.apache.kafka.common.{ Metric, MetricName }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
-import scala.compat.java8.FutureConverters._
import scala.concurrent.Future
import scala.language.reflectiveCalls
@@ -67,8 +67,8 @@ class ControlSpec extends AnyWordSpec with ScalaFutures with Matchers with LogCa
"drain to stream result" in {
val control = createControl()
val drainingControl =
- Consumer.createDrainingControl(control, Future.successful("expected").toJava)
- drainingControl.drainAndShutdown(ec).toScala.futureValue should be("expected")
+ Consumer.createDrainingControl(control, Future.successful("expected").asJava)
+ drainingControl.drainAndShutdown(ec).asScala.futureValue should be("expected")
control.shutdownCalled.get() should be(true)
}
@@ -77,8 +77,8 @@ class ControlSpec extends AnyWordSpec with ScalaFutures with Matchers with LogCa
val drainingControl = Consumer.createDrainingControl(
control,
- Future.failed[String](new RuntimeException("expected")).toJava)
- val value = drainingControl.drainAndShutdown(ec).toScala.failed.futureValue
+ Future.failed[String](new RuntimeException("expected")).asJava)
+ val value = drainingControl.drainAndShutdown(ec).asScala.failed.futureValue
value shouldBe a[RuntimeException]
value.getMessage should be("expected")
control.shutdownCalled.get() should be(true)
@@ -89,8 +89,8 @@ class ControlSpec extends AnyWordSpec with ScalaFutures with Matchers with LogCa
val drainingControl = Consumer.createDrainingControl(
control,
- Future.failed[String](new RuntimeException("expected")).toJava)
- val value = drainingControl.drainAndShutdown(ec).toScala.failed.futureValue
+ Future.failed[String](new RuntimeException("expected")).asJava)
+ val value = drainingControl.drainAndShutdown(ec).asScala.failed.futureValue
value shouldBe a[RuntimeException]
value.getMessage should be("expected")
control.shutdownCalled.get() should be(true)
@@ -99,8 +99,8 @@ class ControlSpec extends AnyWordSpec with ScalaFutures with Matchers with LogCa
"drain to shutdown failure when stream succeeds" in {
val control = createControl(shutdownFuture = Future.failed(new RuntimeException("expected")))
- val drainingControl = Consumer.createDrainingControl(control, Future.successful(Done).toJava)
- val value = drainingControl.drainAndShutdown(ec).toScala.failed.futureValue
+ val drainingControl = Consumer.createDrainingControl(control, Future.successful(Done).asJava)
+ val value = drainingControl.drainAndShutdown(ec).asScala.failed.futureValue
value shouldBe a[RuntimeException]
value.getMessage should be("expected")
control.shutdownCalled.get() should be(true)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org