You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2025/09/22 17:10:39 UTC
(pekko-connectors-kafka) branch main updated: Remove some deprecated code (#369)
This is an automated email from the ASF dual-hosted git repository.
fanningpj 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 fc625a47 Remove some deprecated code (#369)
fc625a47 is described below
commit fc625a47197c190103e390808c564c5581a1fea7
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Mon Sep 22 18:10:34 2025 +0100
Remove some deprecated code (#369)
* remove deprecated methods
* compile issues
* more
* Create remove-deprecated-code.excludes
* scalafmt
* remove old imports
---
.../remove-deprecated-code.excludes | 41 +++++
.../org/apache/pekko/kafka/ConsumerMessage.scala | 9 --
.../scala/org/apache/pekko/kafka/Metadata.scala | 24 ---
.../org/apache/pekko/kafka/ProducerSettings.scala | 5 -
.../apache/pekko/kafka/RestrictedConsumer.scala | 6 -
.../org/apache/pekko/kafka/Subscriptions.scala | 8 +-
.../pekko/kafka/internal/KafkaConsumerActor.scala | 9 --
.../pekko/kafka/internal/MessageBuilder.scala | 6 -
.../pekko/kafka/javadsl/MetadataClient.scala | 6 -
.../org/apache/pekko/kafka/javadsl/Producer.scala | 169 ---------------------
.../apache/pekko/kafka/javadsl/SendProducer.scala | 5 -
.../pekko/kafka/scaladsl/MetadataClient.scala | 10 --
.../org/apache/pekko/kafka/scaladsl/Producer.scala | 158 -------------------
.../apache/pekko/kafka/scaladsl/SendProducer.scala | 5 -
.../org/apache/pekko/kafka/testkit/KafkaTest.java | 11 --
.../testkit/TestcontainersKafkaJunit4Test.java | 11 --
.../kafka/testkit/TestcontainersKafkaTest.java | 10 --
.../pekko/kafka/testkit/javadsl/BaseKafkaTest.java | 14 +-
.../kafka/testkit/javadsl/KafkaJunit4Test.java | 11 --
.../scala/docs/scaladsl/proto/OrderProto.scala | 2 -
20 files changed, 47 insertions(+), 473 deletions(-)
diff --git a/core/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-code.excludes b/core/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-code.excludes
new file mode 100644
index 00000000..b1b812e9
--- /dev/null
+++ b/core/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-code.excludes
@@ -0,0 +1,41 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Remove deprecated code
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.ConsumerMessage#Committable.commitScaladsl")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.ConsumerMessage#Committable.commitJavadsl")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.Metadata.createGetCommittedOffset")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.kafka.Metadata$CommittedOffset")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.kafka.Metadata$CommittedOffset$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.kafka.Metadata$GetCommittedOffset")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.kafka.Metadata$GetCommittedOffset$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.ProducerSettings.producerFactory")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.RestrictedConsumer.committed")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.MetadataClient.getCommittedOffset")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.Producer.flowWithContext")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.Producer.flexiFlow")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.Producer.flow")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.Producer.committableSink")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.Producer.plainSink")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.javadsl.SendProducer.this")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.MetadataClient.getCommittedOffset")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.Producer.flowWithContext")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.Producer.flexiFlow")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.Producer.flow")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.Producer.committableSink")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.Producer.plainSink")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.kafka.scaladsl.SendProducer.apply")
diff --git a/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala b/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala
index c6bedd92..4f19a5fc 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala
@@ -57,15 +57,6 @@ object ConsumerMessage {
* or a number of offsets aggregated as [[CommittableOffsetBatch]].
*/
@DoNotInherit trait Committable {
- @deprecated("use `Committer.flow` or `Committer.sink` instead of direct usage", "Alpakka Kafka 2.0.0")
- def commitScaladsl(): Future[Done]
-
- /**
- * @deprecated use `Committer.flow` or `Committer.sink` instead of direct usage, since Alpakka Kafka 2.0.0
- */
- @java.lang.Deprecated
- @deprecated("use `Committer.flow` or `Committer.sink` instead of direct usage", "Alpakka Kafka 2.0.0")
- def commitJavadsl(): CompletionStage[Done]
@InternalApi
private[kafka] def commitInternal(): Future[Done]
diff --git a/core/src/main/scala/org/apache/pekko/kafka/Metadata.scala b/core/src/main/scala/org/apache/pekko/kafka/Metadata.scala
index 453a6d03..3c744b0d 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/Metadata.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/Metadata.scala
@@ -176,30 +176,6 @@ object Metadata {
def createGetOffsetForTimes(timestampsToSearch: java.util.Map[TopicPartition, java.lang.Long]): GetOffsetsForTimes =
GetOffsetsForTimes(timestampsToSearch.asScala.view.mapValues(_.toLong).toMap)
- /**
- * [[org.apache.kafka.clients.consumer.KafkaConsumer#committed()]]
- */
- @deprecated("use `GetCommittedOffsets`", "Alpakka Kafka 2.0.3")
- final case class GetCommittedOffset(partition: TopicPartition) extends Request with NoSerializationVerificationNeeded
-
- @deprecated("use `CommittedOffsets`", "Alpakka Kafka 2.0.3")
- final case class CommittedOffset(response: Try[OffsetAndMetadata], requestedPartition: TopicPartition)
- extends Response
- with NoSerializationVerificationNeeded {
-
- /**
- * Java API
- */
- def getResponse: Optional[OffsetAndMetadata] = Optional.ofNullable(response.toOption.orNull)
- }
-
- /**
- * Java API:
- * [[org.apache.kafka.clients.consumer.KafkaConsumer#committed()]]
- */
- @deprecated("use `createGetCommittedOffsets`", "Alpakka Kafka 2.0.3")
- def createGetCommittedOffset(partition: TopicPartition): GetCommittedOffset = GetCommittedOffset(partition)
-
/**
* [[org.apache.kafka.clients.consumer.KafkaConsumer#committed()]]
*/
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 eee0dfc9..936da91e 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala
@@ -229,11 +229,6 @@ class ProducerSettings[K, V] @InternalApi private[kafka] (
val enrichAsync: Option[ProducerSettings[K, V] => Future[ProducerSettings[K, V]]],
val producerFactorySync: Option[ProducerSettings[K, V] => Producer[K, V]]) {
- @deprecated(
- "Use createKafkaProducer(), createKafkaProducerAsync(), or createKafkaProducerCompletionStage() to get a new KafkaProducer",
- "Alpakka Kafka 2.0.0")
- def producerFactory: ProducerSettings[K, V] => Producer[K, V] = _ => createKafkaProducer()
-
/**
* An id string to pass to the server when making requests. The purpose of this is to be able to track the source
* of requests beyond just ip/port by allowing a logical application name to be included in server-side request logging.
diff --git a/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala b/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala
index fcaccf60..1b5c2f14 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/RestrictedConsumer.scala
@@ -43,12 +43,6 @@ final class RestrictedConsumer(consumer: Consumer[_, _], duration: java.time.Dur
def commitSync(offsets: java.util.Map[TopicPartition, OffsetAndMetadata]): Unit =
consumer.commitSync(offsets, duration)
- /**
- * See [[org.apache.kafka.clients.consumer.KafkaConsumer#committed(TopicPartition,java.time.Duration)]]
- */
- @deprecated("use `committed(java.util.Set[TopicPartition])`", "Alpakka Kafka 2.0.5")
- def committed(tp: TopicPartition): OffsetAndMetadata = consumer.committed(tp, duration)
-
/**
* See [[org.apache.kafka.clients.consumer.KafkaConsumer#committed(java.util.Set[TopicPartition],java.time.Duration)]]
*/
diff --git a/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala b/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala
index 07a913f1..147dc590 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/Subscriptions.scala
@@ -42,12 +42,12 @@ sealed trait Subscription {
*/
sealed trait ManualSubscription extends Subscription {
- /** @deprecated Manual subscriptions do never rebalance, since Alpakka Kafka 1.0-RC1 */
- @deprecated("Manual subscription does never rebalance", "Alpakka Kafka 1.0-RC1")
+ /** @deprecated Manual subscriptions never rebalances, since Alpakka Kafka 1.0-RC1 */
+ @deprecated("Manual subscription never rebalances", "Alpakka Kafka 1.0-RC1")
def rebalanceListener: Option[ActorRef] = None
- /** @deprecated Manual subscriptions do never rebalance, since Alpakka Kafka 1.0-RC1 */
- @deprecated("Manual subscription does never rebalance", "Alpakka Kafka 1.0-RC1")
+ /** @deprecated Manual subscriptions never rebalances, since Alpakka Kafka 1.0-RC1 */
+ @deprecated("Manual subscription never rebalances", "Alpakka Kafka 1.0-RC1")
def withRebalanceListener(ref: ActorRef): ManualSubscription
}
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 df6e6b61..b837dd42 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
@@ -713,15 +713,6 @@ import scala.util.control.NonFatal
.filterNot(_._2 == null)
.toMap
})
-
- case req: Metadata.GetCommittedOffset @nowarn("cat=deprecation") =>
- @nowarn("cat=deprecation") val resp = Metadata.CommittedOffset(
- Try {
- @nowarn("cat=deprecation") val offset = consumer.committed(req.partition, settings.getMetadataRequestTimeout)
- offset
- },
- req.partition)
- resp
}
private def stopFromMessage(msg: StopLike) = msg match {
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 9c060377..4ca02f29 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
@@ -142,8 +142,6 @@ private[kafka] trait OffsetContextBuilder[K, V]
override val partitionOffset: ConsumerMessage.PartitionOffset,
override val metadata: String)(
val committer: KafkaAsyncConsumerCommitterRef) extends CommittableOffsetMetadata {
- override def commitScaladsl(): Future[Done] = commitInternal()
- override def commitJavadsl(): CompletionStage[Done] = commitInternal().asJava
override def commitInternal(): Future[Done] = KafkaAsyncConsumerCommitterRef.commit(this)
override val batchSize: Long = 1
}
@@ -223,8 +221,6 @@ private[kafka] final class CommittableOffsetBatchImpl(
override def toString: String =
s"CommittableOffsetBatch(batchSize=$batchSize, ${offsets.mkString(", ")})"
- override def commitScaladsl(): Future[Done] = commitInternal()
-
override def commitInternal(): Future[Done] = KafkaAsyncConsumerCommitterRef.commit(this)
override def tellCommit(): CommittableOffsetBatch = tellCommitWithPriority(emergency = false)
@@ -242,8 +238,6 @@ private[kafka] final class CommittableOffsetBatchImpl(
new CommittableOffsetBatchImpl(newOffsets, newCommitters, newOffsets.size.toLong)
}
- 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/MetadataClient.scala b/core/src/main/scala/org/apache/pekko/kafka/javadsl/MetadataClient.scala
index 14f7bda7..a9568476 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
@@ -75,12 +75,6 @@ class MetadataClient private (metadataClient: pekko.kafka.scaladsl.MetadataClien
}(ExecutionContext.parasitic)
.asJava
- @deprecated("use `getCommittedOffsets`", "Alpakka Kafka 2.0.3")
- def getCommittedOffset(partition: TopicPartition): CompletionStage[OffsetAndMetadata] =
- metadataClient
- .getCommittedOffset(partition)
- .asJava
-
def getCommittedOffsets(
partitions: java.util.Set[TopicPartition]): CompletionStage[java.util.Map[TopicPartition, OffsetAndMetadata]] =
metadataClient
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 5fcdfdbd..6eacd2a4 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
@@ -45,78 +45,6 @@ object Producer {
.mapMaterializedValue(_.asJava)
.asJava
- /**
- * Create a sink for publishing records to Kafka topics.
- *
- * The [[org.apache.kafka.clients.producer.ProducerRecord Kafka ProducerRecord]] contains the topic name to which the record is being sent, an optional
- * partition number, and an optional key and value.
- *
- * Supports sharing a Kafka Producer instance.
- *
- * @deprecated Pass in external or shared producer using `ProducerSettings.withProducerFactory` or `ProducerSettings.withProducer`, since Alpakka Kafka 2.0.0
- */
- @Deprecated
- def plainSink[K, V](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V]): Sink[ProducerRecord[K, V], CompletionStage[Done]] =
- plainSink(settings.withProducer(producer))
-
- /**
- * Create a sink that is aware of the [[ConsumerMessage.Committable committable offset]]
- * from a [[Consumer.committableSource]]. It will commit the consumer offset when the message has
- * been published successfully to the topic.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
- *
- * Note that there is a risk that something fails after publishing but before
- * committing, so it is "at-least once delivery" semantics.
- *
- * @deprecated use `committableSink(ProducerSettings, CommitterSettings)` instead, since Alpakka Kafka 2.0.0
- */
- @Deprecated
- def committableSink[K, V, IN <: Envelope[K, V, ConsumerMessage.Committable]](
- settings: ProducerSettings[K, V]): Sink[IN, CompletionStage[Done]] = {
- @nowarn("cat=deprecation")
- val sink: Sink[IN, CompletionStage[Done]] = scaladsl.Producer
- .committableSink(settings)
- .mapMaterializedValue(_.asJava)
- .asJava
- sink
- }
-
- /**
- * Create a sink that is aware of the [[ConsumerMessage.Committable committable offset]]
- * from a [[Consumer.committableSource]]. It will commit the consumer offset when the message has
- * been published successfully to the topic.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
- *
- * Note that there is always a risk that something fails after publishing but before
- * committing, so it is "at-least once delivery" semantics.
- *
- * Supports sharing a Kafka Producer instance.
- *
- * @deprecated use `committableSink(ProducerSettings, CommitterSettings)` instead, since Alpakka Kafka 2.0.0
- */
- @Deprecated
- def committableSink[K, V](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : Sink[Envelope[K, V, ConsumerMessage.Committable], CompletionStage[Done]] =
- committableSink(settings.withProducer(producer))
-
/**
* Create a sink that is aware of the [[ConsumerMessage.Committable committable offset]]
* from a [[Consumer.committableSource]]. The offsets are batched and committed regularly.
@@ -164,28 +92,6 @@ object Producer {
override def apply(p: japi.Pair[IN, C]) = p.first.withPassThrough(p.second)
})
- /**
- * Create a flow to publish records to Kafka topics and then pass it on.
- *
- * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
- *
- * The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
- * or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
- * be committed later in the flow.
- *
- * @deprecated use `flexiFlow` instead, since 0.21
- */
- @Deprecated
- def flow[K, V, PassThrough](
- settings: ProducerSettings[K, V]): Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] = {
- @nowarn("cat=deprecation")
- val flow = scaladsl.Producer
- .flow(settings)
- .asJava
- .asInstanceOf[Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed]]
- flow
- }
-
/**
* Create a flow to conditionally publish records to Kafka topics and then pass it on.
*
@@ -230,79 +136,4 @@ object Producer {
settings: ProducerSettings[K, V]): FlowWithContext[Envelope[K, V, NotUsed], C, Results[K, V, C], C, NotUsed] =
scaladsl.Producer.flowWithContext(settings).asJava
- /**
- * Create a flow to publish records to Kafka topics and then pass it on.
- *
- * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
- *
- * The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
- * or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
- * be committed later in the flow.
- *
- * Supports sharing a Kafka Producer instance.
- *
- * @deprecated use `flexiFlow` instead, since 0.21
- */
- @Deprecated
- def flow[K, V, PassThrough](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] =
- flow(settings.withProducer(producer))
-
- /**
- * Create a flow to conditionally publish records to Kafka topics and then pass it on.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
- *
- * The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
- * or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
- * be committed later in the flow.
- *
- * Supports sharing a Kafka Producer instance.
- *
- * @deprecated Pass in external or shared producer using `ProducerSettings.withProducerFactory` or `ProducerSettings.withProducer`, since Alpakka Kafka 2.0.0
- */
- @Deprecated
- def flexiFlow[K, V, PassThrough](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : Flow[Envelope[K, V, PassThrough], Results[K, V, PassThrough], NotUsed] =
- flexiFlow(settings.withProducer(producer))
-
- /**
- * API MAY CHANGE
- *
- * Create a flow to conditionally publish records to Kafka topics and then pass it on.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
- *
- * This flow is intended to be used with Apache Pekko's [flow with context](https://pekko.apache.org/docs/pekko/current/stream/operators/Flow/asFlowWithContext.html).
- *
- * Supports sharing a Kafka Producer instance.
- *
- * @tparam C the flow context type
- *
- * @deprecated Pass in external or shared producer using `ProducerSettings.withProducerFactory` or `ProducerSettings.withProducer`, since Alpakka Kafka 2.0.0
- */
- @Deprecated
- @ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/880")
- def flowWithContext[K, V, C](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : FlowWithContext[Envelope[K, V, NotUsed], C, Results[K, V, C], C, NotUsed] =
- flowWithContext(settings.withProducer(producer))
-
}
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 efd992be..fbee1428 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
@@ -30,11 +30,6 @@ import scala.jdk.FutureConverters._
*/
final class SendProducer[K, V] private (underlying: scaladsl.SendProducer[K, V]) {
- // kept for bin-compatibility
- @deprecated("use the variant with ClassicActorSystemProvider instead", "Alpakka Kafka 2.0.5")
- private[kafka] def this(settings: ProducerSettings[K, V], system: ActorSystem) =
- this(scaladsl.SendProducer(settings)(system))
-
/**
* Utility class for producing to Kafka without using Apache Pekko Streams.
* @param settings producer settings used to create or access the [[org.apache.kafka.clients.producer.Producer]]
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala
index 62e5aecf..bb3c049c 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/MetadataClient.scala
@@ -75,16 +75,6 @@ class MetadataClient private (consumerActor: ActorRef, timeout: Timeout, managed
case Failure(e) => Future.failed(e)
}(ExecutionContext.parasitic)
- @deprecated("use `getCommittedOffsets`", "Alpakka Kafka 2.0.3")
- def getCommittedOffset(partition: TopicPartition): Future[OffsetAndMetadata] =
- (consumerActor ? GetCommittedOffset(partition))(timeout)
- .mapTo[CommittedOffset]
- .map(_.response)
- .flatMap {
- case Success(res) => Future.successful(res)
- case Failure(e) => Future.failed(e)
- }(ExecutionContext.parasitic)
-
def getCommittedOffsets(partitions: Set[TopicPartition]): Future[Map[TopicPartition, OffsetAndMetadata]] =
(consumerActor ? GetCommittedOffsets(partitions))(timeout)
.mapTo[CommittedOffsets]
diff --git a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala
index 0ddab243..181f48cf 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala
@@ -44,70 +44,6 @@ object Producer {
.via(flexiFlow(settings))
.toMat(Sink.ignore)(Keep.right)
- /**
- * Create a sink for publishing records to Kafka topics.
- *
- * The [[org.apache.kafka.clients.producer.ProducerRecord Kafka ProducerRecord]] contains the topic name to which the record is being sent, an optional
- * partition number, and an optional key and value.
- *
- * Supports sharing a Kafka Producer instance.
- */
- @deprecated(
- "Pass in external or shared producer using ProducerSettings.withProducerFactory or ProducerSettings.withProducer",
- "Alpakka Kafka 2.0.0")
- def plainSink[K, V](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V]): Sink[ProducerRecord[K, V], Future[Done]] =
- plainSink(settings.withProducer(producer))
-
- /**
- * Create a sink that is aware of the [[ConsumerMessage.Committable committable offset]]
- * from a [[Consumer.committableSource]]. It will commit the consumer offset when the message has
- * been published successfully to the topic.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
- *
- * Note that there is a risk that something fails after publishing but before
- * committing, so it is "at-least once delivery" semantics.
- */
- @deprecated("use `committableSink(ProducerSettings, CommitterSettings)` instead", "Alpakka Kafka 2.0.0")
- def committableSink[K, V](
- settings: ProducerSettings[K, V]): Sink[Envelope[K, V, ConsumerMessage.Committable], Future[Done]] =
- flexiFlow[K, V, ConsumerMessage.Committable](settings)
- .mapAsync(settings.parallelism)(_.passThrough.commitInternal())
- .toMat(Sink.ignore)(Keep.right)
-
- /**
- * Create a sink that is aware of the [[ConsumerMessage.Committable committable offset]]
- * from a [[Consumer.committableSource]]. It will commit the consumer offset when the message has
- * been published successfully to the topic.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and commits the offset
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, but commits the offset
- *
- * Note that there is always a risk that something fails after publishing but before
- * committing, so it is "at-least once delivery" semantics.
- *
- * Supports sharing a Kafka Producer instance.
- */
- @deprecated("use `committableSink(ProducerSettings, CommitterSettings)` instead", "Alpakka Kafka 2.0.0")
- def committableSink[K, V](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : Sink[Envelope[K, V, ConsumerMessage.Committable], Future[Done]] =
- committableSink(settings.withProducer(producer))
-
/**
* Create a sink that is aware of the [[ConsumerMessage.Committable committable offset]]
* from a [[Consumer.committableSource]]. The offsets are batched and committed regularly.
@@ -153,27 +89,6 @@ object Producer {
env.withPassThrough(offset)
}
- /**
- * Create a flow to publish records to Kafka topics and then pass it on.
- *
- * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
- *
- * The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
- * or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
- * be committed later in the flow.
- */
- @deprecated("prefer flexiFlow over this flow implementation", "0.21")
- def flow[K, V, PassThrough](
- settings: ProducerSettings[K, V]): Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] = {
- val flow = Flow
- .fromGraph(
- new DefaultProducerStage[K, V, PassThrough, Message[K, V, PassThrough], Result[K, V, PassThrough]](
- settings))
- .mapAsync(settings.parallelism)(identity)
-
- flowWithDispatcher(settings, flow)
- }
-
/**
* Create a flow to conditionally publish records to Kafka topics and then pass it on.
*
@@ -225,79 +140,6 @@ object Producer {
case (env, c) => env.withPassThrough(c)
}(res => res.passThrough)
- /**
- * Create a flow to publish records to Kafka topics and then pass it on.
- *
- * The records must be wrapped in a [[pekko.kafka.ProducerMessage.Message Message]] and continue in the stream as [[pekko.kafka.ProducerMessage.Result Result]].
- *
- * The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
- * or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
- * be committed later in the flow.
- *
- * Supports sharing a Kafka Producer instance.
- */
- @deprecated("prefer flexiFlow over this flow implementation", "0.21")
- def flow[K, V, PassThrough](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] =
- flow(settings.withProducer(producer))
-
- /**
- * Create a flow to conditionally publish records to Kafka topics and then pass it on.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
- *
- * The messages support the possibility to pass through arbitrary data, which can for example be a [[ConsumerMessage.CommittableOffset CommittableOffset]]
- * or [[ConsumerMessage.CommittableOffsetBatch CommittableOffsetBatch]] that can
- * be committed later in the flow.
- *
- * Supports sharing a Kafka Producer instance.
- */
- @deprecated(
- "Pass in external or shared producer using ProducerSettings.withProducerFactory or ProducerSettings.withProducer",
- "Alpakka Kafka 2.0.0")
- def flexiFlow[K, V, PassThrough](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : Flow[Envelope[K, V, PassThrough], Results[K, V, PassThrough], NotUsed] =
- flexiFlow(settings.withProducer(producer))
-
- /**
- * API MAY CHANGE
- *
- * Create a flow to conditionally publish records to Kafka topics and then pass it on.
- *
- * It publishes records to Kafka topics conditionally:
- *
- * - [[pekko.kafka.ProducerMessage.Message Message]] publishes a single message to its topic, and continues in the stream as [[pekko.kafka.ProducerMessage.Result Result]]
- *
- * - [[pekko.kafka.ProducerMessage.MultiMessage MultiMessage]] publishes all messages in its `records` field, and continues in the stream as [[pekko.kafka.ProducerMessage.MultiResult MultiResult]]
- *
- * - [[pekko.kafka.ProducerMessage.PassThroughMessage PassThroughMessage]] does not publish anything, and continues in the stream as [[pekko.kafka.ProducerMessage.PassThroughResult PassThroughResult]]
- *
- * This flow is intended to be used with Apache Pekko's [flow with context](https://pekko.apache.org/docs/pekko/current/stream/operators/Flow/asFlowWithContext.html).
- *
- * Supports sharing a Kafka Producer instance.
- *
- * @tparam C the flow context type
- */
- @deprecated(
- "Pass in external or shared producer using ProducerSettings.withProducerFactory or ProducerSettings.withProducer",
- "Alpakka Kafka 2.0.0")
- @ApiMayChange(issue = "https://github.com/akka/alpakka-kafka/issues/880")
- def flowWithContext[K, V, C](
- settings: ProducerSettings[K, V],
- producer: org.apache.kafka.clients.producer.Producer[K, V])
- : FlowWithContext[Envelope[K, V, NotUsed], C, Results[K, V, C], C, NotUsed] =
- flowWithContext(settings.withProducer(producer))
-
private def flowWithDispatcher[PassThrough, V, K](
settings: ProducerSettings[K, V],
flow: Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed]) =
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 97fd4f6c..2de38dd0 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
@@ -108,9 +108,4 @@ final class SendProducer[K, V] private (val settings: ProducerSettings[K, V], sy
object SendProducer {
def apply[K, V](settings: ProducerSettings[K, V])(implicit system: ClassicActorSystemProvider): SendProducer[K, V] =
new SendProducer(settings, system.classicSystem)
-
- // kept for bin-compatibility
- @deprecated("use the variant with ClassicActorSystemProvider instead", "Alpakka Kafka 2.0.5")
- def apply[K, V](settings: ProducerSettings[K, V], system: ActorSystem): SendProducer[K, V] =
- new SendProducer(settings, system)
}
diff --git a/testkit/src/main/java/org/apache/pekko/kafka/testkit/KafkaTest.java b/testkit/src/main/java/org/apache/pekko/kafka/testkit/KafkaTest.java
index a3487d6c..38149f13 100644
--- a/testkit/src/main/java/org/apache/pekko/kafka/testkit/KafkaTest.java
+++ b/testkit/src/main/java/org/apache/pekko/kafka/testkit/KafkaTest.java
@@ -14,10 +14,8 @@
package org.apache.pekko.kafka.testkit;
-import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.actor.ClassicActorSystemProvider;
import org.apache.pekko.kafka.testkit.javadsl.BaseKafkaTest;
-import org.apache.pekko.stream.Materializer;
import org.apache.pekko.stream.testkit.javadsl.StreamTestKit;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -32,15 +30,6 @@ import org.junit.jupiter.api.BeforeAll;
@SuppressWarnings("unchecked")
public abstract class KafkaTest extends BaseKafkaTest {
- /**
- * @deprecated Materializer no longer necessary in Akka 2.6, use
- * `KafkaTest(ClassicActorSystemProvider, String)` instead, since Alpakka Kafka 2.1.0
- */
- @Deprecated
- protected KafkaTest(ActorSystem system, Materializer mat, String bootstrapServers) {
- super(system, mat, bootstrapServers);
- }
-
protected KafkaTest(ClassicActorSystemProvider system, String bootstrapServers) {
super(system, bootstrapServers);
}
diff --git a/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaJunit4Test.java b/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaJunit4Test.java
index 5d0a06fa..fbdd5d3b 100644
--- a/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaJunit4Test.java
+++ b/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaJunit4Test.java
@@ -18,7 +18,6 @@ import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.actor.ClassicActorSystemProvider;
import org.apache.pekko.kafka.testkit.internal.TestcontainersKafka;
import org.apache.pekko.kafka.testkit.javadsl.KafkaJunit4Test;
-import org.apache.pekko.stream.Materializer;
import org.junit.After;
import org.junit.Before;
@@ -35,16 +34,6 @@ public abstract class TestcontainersKafkaJunit4Test extends KafkaJunit4Test {
private static final KafkaTestkitTestcontainersSettings settings =
TestcontainersKafka.Singleton().testcontainersSettings();
- /**
- * @deprecated Materializer no longer necessary in Akka 2.6, use
- * `TestcontainersKafkaJunit4Test(ClassicActorSystemProvider)` instead, since Alpakka Kafka
- * 2.1.0
- */
- @Deprecated
- protected TestcontainersKafkaJunit4Test(ActorSystem system, Materializer mat) {
- super(system, mat, startKafka(settings));
- }
-
protected TestcontainersKafkaJunit4Test(ClassicActorSystemProvider system) {
super(system, startKafka(settings));
}
diff --git a/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaTest.java b/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaTest.java
index f4ed2e1e..965a4ad3 100644
--- a/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaTest.java
+++ b/testkit/src/main/java/org/apache/pekko/kafka/testkit/TestcontainersKafkaTest.java
@@ -17,7 +17,6 @@ package org.apache.pekko.kafka.testkit;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.actor.ClassicActorSystemProvider;
import org.apache.pekko.kafka.testkit.internal.TestcontainersKafka;
-import org.apache.pekko.stream.Materializer;
/**
* JUnit 5 base class using [[https://www.testcontainers.org/ Testcontainers]] to start a Kafka
@@ -36,15 +35,6 @@ public abstract class TestcontainersKafkaTest extends KafkaTest {
public static final KafkaTestkitTestcontainersSettings settings =
TestcontainersKafka.Singleton().testcontainersSettings();
- /**
- * @deprecated Materializer no longer necessary in Akka 2.6, use
- * `TestcontainersKafkaTest(ClassicActorSystemProvider)` instead, since Alpakka Kafka 2.1.0
- */
- @Deprecated
- protected TestcontainersKafkaTest(ActorSystem system, Materializer mat) {
- super(system, mat, startKafka(settings));
- }
-
protected TestcontainersKafkaTest(ClassicActorSystemProvider system) {
super(system, startKafka(settings));
}
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 660a0e80..031e7f11 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
@@ -31,7 +31,6 @@ import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.pekko.Done;
-import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.actor.ClassicActorSystemProvider;
import org.apache.pekko.japi.Pair;
import org.apache.pekko.kafka.Subscriptions;
@@ -54,18 +53,9 @@ public abstract class BaseKafkaTest extends KafkaTestKitClass {
protected final Materializer materializer;
- /**
- * @deprecated Materializer no longer necessary in Akka 2.6, use
- * `BaseKafkaTest(ClassicActorSystemProvider, String)` instead, since Alpakka Kafka 2.1.0
- */
- @Deprecated
- protected BaseKafkaTest(ActorSystem system, Materializer mat, String bootstrapServers) {
- super(system, bootstrapServers);
- this.materializer = mat;
- }
-
protected BaseKafkaTest(ClassicActorSystemProvider system, String bootstrapServers) {
- this(system.classicSystem(), Materializer.matFromSystem(system), bootstrapServers);
+ super(system.classicSystem(), bootstrapServers);
+ this.materializer = Materializer.matFromSystem(system);
}
@Override
diff --git a/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/KafkaJunit4Test.java b/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/KafkaJunit4Test.java
index 3672f03d..d81a3b1e 100644
--- a/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/KafkaJunit4Test.java
+++ b/testkit/src/main/java/org/apache/pekko/kafka/testkit/javadsl/KafkaJunit4Test.java
@@ -14,9 +14,7 @@
package org.apache.pekko.kafka.testkit.javadsl;
-import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.actor.ClassicActorSystemProvider;
-import org.apache.pekko.stream.Materializer;
import org.apache.pekko.stream.testkit.javadsl.StreamTestKit;
import org.junit.After;
import org.junit.Before;
@@ -25,15 +23,6 @@ import org.junit.Before;
@SuppressWarnings("unchecked")
public abstract class KafkaJunit4Test extends BaseKafkaTest {
- /**
- * @deprecated Materializer no longer necessary in Akka 2.6, use
- * `KafkaJunit4Test(ClassicActorSystemProvider, String)` instead, since Alpakka Kafka 2.1.0
- */
- @Deprecated
- protected KafkaJunit4Test(ActorSystem system, Materializer mat, String bootstrapServers) {
- super(system, mat, bootstrapServers);
- }
-
protected KafkaJunit4Test(ClassicActorSystemProvider system, String bootstrapServers) {
super(system, bootstrapServers);
}
diff --git a/tests/src/test/scala/docs/scaladsl/proto/OrderProto.scala b/tests/src/test/scala/docs/scaladsl/proto/OrderProto.scala
index 5ec03366..b26df6b9 100644
--- a/tests/src/test/scala/docs/scaladsl/proto/OrderProto.scala
+++ b/tests/src/test/scala/docs/scaladsl/proto/OrderProto.scala
@@ -41,6 +41,4 @@ object OrderProto extends _root_.scalapb.GeneratedFileObject {
Array(
))
}
- @deprecated("Use javaDescriptor instead. In a future version this will refer to scalaDescriptor.", "ScalaPB 0.5.47")
- def descriptor: com.google.protobuf.Descriptors.FileDescriptor = javaDescriptor
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org