You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "pjfanning (via GitHub)" <gi...@apache.org> on 2026/03/25 19:09:31 UTC
[PR] Use CompletionStage in Java DSL [pekko-connectors]
pjfanning opened a new pull request, #1526:
URL: https://github.com/apache/pekko-connectors/pull/1526
Several `javadsl` classes exposed `scala.concurrent.Future` in their public API materialized value types, making them awkward to consume from Java. This PR replaces those with `java.util.concurrent.CompletionStage` equivalents and adds deprecated forwarders to preserve backward compatibility.
## Changes
- **`couchbase3/.../javadsl/CouchbaseSink`** — All 8 sink factory methods now return `Sink[T, CompletionStage[Done]]` (or `CompletionStage[java.lang.Boolean]` for `exists`). Old signatures preserved as `@deprecated` methods with an `F` suffix (e.g. `insertDocF`, `upsertF`, `existsF`, …).
- **`google-cloud-pub-sub/.../javadsl/GooglePubSub`** — `subscribeFlow` now returns `Flow[Done, ReceivedMessage, CompletionStage[NotUsed]]`. Old `Future`-returning variant kept as `@deprecated subscribeFlowWithScalaFuture`.
- **`kinesis/.../javadsl/KinesisSchedulerSource`** — New `createShardedCs` returns `SubSource[CommittableRecord, CompletionStage[Scheduler]]`. Old `createSharded` (returning `Future[Scheduler]`) is now `@deprecated`.
- **`kinesis/.../docs/javadsl/KclSnippets`** — Added `createShardedCs` usage example.
## Example (Couchbase)
```java
// Before — materialized value was scala.concurrent.Future, not usable from Java
Sink<MutationDocument<MyDoc>, Future<Done>> sink = CouchbaseSink.insertDoc(options);
// After — proper Java API
Sink<MutationDocument<MyDoc>, CompletionStage<Done>> sink = CouchbaseSink.insertDoc(options);
// Backward compat (deprecated)
Sink<MutationDocument<MyDoc>, Future<Done>> sink = CouchbaseSink.insertDocF(options);
```
<!-- START COPILOT ORIGINAL PROMPT -->
<details>
<summary>Original prompt</summary>
## Problem
Several `javadsl` classes expose Scala types (`scala.concurrent.Future`, `scala.concurrent.duration.FiniteDuration`) in their public API signatures, making them difficult to use correctly from Java. Java users should receive `java.util.concurrent.CompletionStage` and `java.lang.Boolean` in place of their Scala equivalents.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in Java DSL [pekko-connectors]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #1526:
URL: https://github.com/apache/pekko-connectors/pull/1526#issuecomment-4129641779
The return types on the methods have been changed to Java types but the old methods were retained but under new names (`Future` appended to the function name).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in Java DSL [pekko-connectors]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1526:
URL: https://github.com/apache/pekko-connectors/pull/1526#discussion_r2993796573
##########
couchbase3/src/main/scala/org/apache/pekko/stream/connectors/couchbase3/javadsl/CouchbaseSink.scala:
##########
@@ -24,69 +24,189 @@ import org.apache.pekko.Done
import org.apache.pekko.stream.connectors.couchbase3.scaladsl.{ CouchbaseSink => ScalaCouchbaseSink }
import org.apache.pekko.stream.javadsl.Sink
+import java.util.concurrent.CompletionStage
import scala.concurrent.Future
+import scala.jdk.FutureConverters._
object CouchbaseSink {
/**
* reference to [[CouchbaseFlow.insertDoc]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #insertDocFuture} which works like this method worked in 1.x.
*/
def insertDoc[T](insertOptions: InsertOptions)(
+ implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], CompletionStage[Done]] =
+ ScalaCouchbaseSink.insertDoc[T](insertOptions).mapMaterializedValue(_.asJava).asJava
+
+ /**
+ * reference to [[CouchbaseFlow.insertDoc]]
+ * @deprecated Use insertDoc which returns CompletionStage instead
+ */
+ @deprecated("Use insertDoc which returns CompletionStage instead", since = "2.0.0")
+ def insertDocFuture[T](insertOptions: InsertOptions)(
implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], Future[Done]] =
ScalaCouchbaseSink.insertDoc[T](insertOptions).asJava
/**
- * reference to [[CouchbaseFlow.upsertDoc]]
+ * reference to [[CouchbaseFlow.insert]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #insertFuture} which works like this method worked in 1.x.
*/
def insert[T](applyId: T => String,
+ insertOptions: InsertOptions)(
+ implicit asyncCollection: AsyncCollection): Sink[T, CompletionStage[Done]] =
+ ScalaCouchbaseSink.insert[T](applyId, insertOptions).mapMaterializedValue(_.asJava).asJava
+
+ /**
+ * reference to [[CouchbaseFlow.insert]]
+ * @deprecated Use insert which returns CompletionStage instead
+ */
+ @deprecated("Use insert which returns CompletionStage instead", since = "2.0.0")
+ def insertFuture[T](applyId: T => String,
insertOptions: InsertOptions)(
implicit asyncCollection: AsyncCollection): Sink[T, Future[Done]] =
ScalaCouchbaseSink.insert[T](applyId, insertOptions).asJava
/**
* reference to [[CouchbaseFlow.upsertDoc]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #upsertDocFuture} which works like this method worked in 1.x.
*/
def upsertDoc[T](upsertOptions: UpsertOptions = UpsertOptions.upsertOptions())(
+ implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], CompletionStage[Done]] =
+ ScalaCouchbaseSink.upsertDoc[T](upsertOptions).mapMaterializedValue(_.asJava).asJava
+
+ /**
+ * reference to [[CouchbaseFlow.upsertDoc]]
+ * @deprecated Use upsertDoc which returns CompletionStage instead
+ */
+ @deprecated("Use upsertDoc which returns CompletionStage instead", since = "2.0.0")
+ def upsertDocFuture[T](upsertOptions: UpsertOptions = UpsertOptions.upsertOptions())(
implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], Future[Done]] =
ScalaCouchbaseSink.upsertDoc[T](upsertOptions).asJava
/**
* reference to [[CouchbaseFlow.upsert]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #upsertFuture} which works like this method worked in 1.x.
*/
def upsert[T](applyId: T => String,
Review Comment:
addressed this in #1528
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in Java DSL [pekko-connectors]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1526:
URL: https://github.com/apache/pekko-connectors/pull/1526#discussion_r2992421781
##########
couchbase3/src/main/scala/org/apache/pekko/stream/connectors/couchbase3/javadsl/CouchbaseSink.scala:
##########
@@ -24,69 +24,189 @@ import org.apache.pekko.Done
import org.apache.pekko.stream.connectors.couchbase3.scaladsl.{ CouchbaseSink => ScalaCouchbaseSink }
import org.apache.pekko.stream.javadsl.Sink
+import java.util.concurrent.CompletionStage
import scala.concurrent.Future
+import scala.jdk.FutureConverters._
object CouchbaseSink {
/**
* reference to [[CouchbaseFlow.insertDoc]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #insertDocFuture} which works like this method worked in 1.x.
*/
def insertDoc[T](insertOptions: InsertOptions)(
+ implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], CompletionStage[Done]] =
+ ScalaCouchbaseSink.insertDoc[T](insertOptions).mapMaterializedValue(_.asJava).asJava
+
+ /**
+ * reference to [[CouchbaseFlow.insertDoc]]
+ * @deprecated Use insertDoc which returns CompletionStage instead
+ */
+ @deprecated("Use insertDoc which returns CompletionStage instead", since = "2.0.0")
+ def insertDocFuture[T](insertOptions: InsertOptions)(
implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], Future[Done]] =
ScalaCouchbaseSink.insertDoc[T](insertOptions).asJava
/**
- * reference to [[CouchbaseFlow.upsertDoc]]
+ * reference to [[CouchbaseFlow.insert]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #insertFuture} which works like this method worked in 1.x.
*/
def insert[T](applyId: T => String,
+ insertOptions: InsertOptions)(
+ implicit asyncCollection: AsyncCollection): Sink[T, CompletionStage[Done]] =
+ ScalaCouchbaseSink.insert[T](applyId, insertOptions).mapMaterializedValue(_.asJava).asJava
+
+ /**
+ * reference to [[CouchbaseFlow.insert]]
+ * @deprecated Use insert which returns CompletionStage instead
+ */
+ @deprecated("Use insert which returns CompletionStage instead", since = "2.0.0")
+ def insertFuture[T](applyId: T => String,
insertOptions: InsertOptions)(
implicit asyncCollection: AsyncCollection): Sink[T, Future[Done]] =
ScalaCouchbaseSink.insert[T](applyId, insertOptions).asJava
/**
* reference to [[CouchbaseFlow.upsertDoc]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #upsertDocFuture} which works like this method worked in 1.x.
*/
def upsertDoc[T](upsertOptions: UpsertOptions = UpsertOptions.upsertOptions())(
+ implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], CompletionStage[Done]] =
+ ScalaCouchbaseSink.upsertDoc[T](upsertOptions).mapMaterializedValue(_.asJava).asJava
+
+ /**
+ * reference to [[CouchbaseFlow.upsertDoc]]
+ * @deprecated Use upsertDoc which returns CompletionStage instead
+ */
+ @deprecated("Use upsertDoc which returns CompletionStage instead", since = "2.0.0")
+ def upsertDocFuture[T](upsertOptions: UpsertOptions = UpsertOptions.upsertOptions())(
implicit asyncCollection: AsyncCollection): Sink[MutationDocument[T], Future[Done]] =
ScalaCouchbaseSink.upsertDoc[T](upsertOptions).asJava
/**
* reference to [[CouchbaseFlow.upsert]]
+ * <p>
+ * This function's return type changed in 2.0.0 to return a Sink with a CompletionStage instead of a
+ * Scala Future, to be more consistent with Java usage.
+ * </p>
+ * @see {@link #upsertFuture} which works like this method worked in 1.x.
*/
def upsert[T](applyId: T => String,
Review Comment:
We need java functions here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in Java DSL [pekko-connectors]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning merged PR #1526:
URL: https://github.com/apache/pekko-connectors/pull/1526
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org