pjfanning opened a new issue, #2043:
URL: https://github.com/apache/pekko-connectors/issues/2043
A review of the `javadsl` packages across all connectors, plus the
settings/model classes reachable from them, found a number of places where the
Java API exposes Scala types or is otherwise hard to use from Java. This is a
tracking issue; items can be split into per-module PRs.
Notes:
- Findings came from reading the source (plus `javap` on existing compiled
classes where available). Items marked *(unverified)* are inferred from JLS
overload rules / erasure and should be confirmed with a small Java compile test
before fixing.
- Most fixes can be additive (new Java-friendly overloads/getters, deprecate
the old ones) to stay binary compatible; 2.0.0 also gives room to
remove/replace where that is cleaner.
## 1. Bugs (behavioural, not just API shape)
- [ ] **google-cloud-bigquery** `InsertAllRetryPolicy.scala:47` — `def
withoutDeduplication = WithDeduplication`; Java users silently get
deduplication.
- [ ] **google-fcm** `models/AndroidConfig.scala:192` —
`LightSettings.light_off_duration(value)` sets `light_on_duration`.
- [ ] **aws-event-bridge** `javadsl/EventBridgePublisher.scala:58` — 2-arg
`flowSeq(settings, client)` delegates to scaladsl `flow`, so it is identical to
`flow` and does not batch.
- [ ] **ironmq** `javadsl/IronMqProducer.scala:33,51,74` —
`asInstanceOf[Flow[_, String, _]]` over the value class `Message.Id`. Elements
are boxed `Message$Id` at runtime, so Java code reading them as `String` should
get a `ClassCastException` *(unverified; Java tests only check `.size()`)*. Use
`.map(_.value)`.
- [ ] **google-common** `GoogleSettings.scala:299` —
`ForwardProxy.create(connectionContext: HttpConnectionContext, ...)` casts to
`HttpsConnectionContext`; a plain HTTP context throws `ClassCastException`.
- [ ] **mqtt-streaming** `model.scala:1129` — `Command(..., completed:
Optional[CompletionStage[Done]], ...)` calls `toCompletableFuture.complete`,
which fails or completes a copy for non-`CompletableFuture` stages; failures
are never propagated.
- [ ] **testkit** `javadsl/LogCapturingExtension.scala:32`,
`LogCapturingJunit4.scala:37` — logback snippet names
`org.apache.pekko.actor.testkit.typed.internal.CapturingAppender` instead of
`org.apache.pekko.stream.connectors.testkit.CapturingAppender`.
## 2. High — Java cannot reasonably call it
### Ambiguous overloads (Java functional interface vs `scala.FunctionN` of
the same arity) *(unverified)*
Since Scala 2.12 `FunctionN` is SAM-compatible, so an implicitly typed Java
lambda matches both overloads.
- [ ] **cassandra** `CassandraSessionSettings.scala:42/50` — `withInit`
- [ ] **couchbase3** `javadsl/CouchbaseFlow.scala` — `insert` 156/177,
`replace` 202/223, `upsert` 248/269, `remove` 296/319, `exists` 378/390,
`touchDuration` 415/438, `touchInstant` 449/472 (deprecated `T => String`
variants; rename like `CouchbaseSink` did with `*Future`, or remove)
- [ ] **hbase** `HTableSettings.scala:43/49` — `withConverter`
- [ ] **jms / jakartams** `Destinations.scala:69-74` — `CustomDestination`
constructors (Java tests work around it with a typed helper); add
`CustomDestination.create(name, java.util.function.Function)`
- [ ] **mqttv5** `settings.scala:295/309` — `withAuth`; the Java variant
also uses `BiFunction[Int, MqttProperties, Pair[Int, MqttProperties]]`, which
erases to `Object`
### Scala types in javadsl signatures
- [ ] **unix-domain-socket** `javadsl/UnixDomainSocket.scala:190` —
`outgoingConnection(..., connectTimeout: scala.concurrent.duration.Duration)`;
add `java.time.Duration` overload
- [ ] **s3** `javadsl/S3.scala:744` —
`listObjectVersionsAndCommonPrefixes(..., prefix: Option[String], ...)`;
siblings use `Optional`
- [ ] **aws-event-bridge** `javadsl/EventBridgePublisher.scala:69` —
`flowSeq(client): Flow[scala Seq[PutEventsRequestEntry], ...]`
- [ ] **google-cloud-bigquery-storage** javadsl `BigQueryArrowStorage`,
`BigQueryAvroStorage`, `BigQueryStorage` — ~20 methods return `scala.Tuple2`
instead of `pekko.japi.Pair`
- [ ] **google-cloud-bigquery-storage**
`javadsl/BigQueryArrowStorage.scala:22`, `BigQueryAvroStorage.scala:21` — take
ScalaPB `TableReadOptions` (needs a Scala `Seq`), while
`javadsl/BigQueryStorage.scala` takes the Java protobuf type
- [ ] **dynamodb** `javadsl/DynamoDb.scala:56` — `flowWithContext` emits
`scala.util.Try` (`@ApiMayChange`)
### Varargs / static access
- [ ] **couchbase** `javadsl/CouchbaseSession.scala:280` — `createIndex(...,
fields: AnyRef*)` lacks `@varargs` (Java sees `immutable.Seq<Object>`); also
returns `CompletionStage[Boolean]` → `CompletionStage<Object>`
- [ ] **google-common** `javadsl/Google.scala:34` — `object Google extends
Google` with a trait companion, so no static forwarders; Java must use
`Google$.MODULE$`
- [ ] Top-level `case object`s needing `X$.MODULE$` from Java:
- **mqtt-streaming** `model.scala:123,125,479` — `PingReq`, `PingResp`,
`Disconnect`
- **google-fcm** `models/AndroidConfig.scala:146-178` — priority /
visibility values
- **solr** `SolrMessages.scala:131-136` — `Operation` values (low)
- **elasticsearch** `ElasticsearchWriteSettings.scala:25` — `RetryNever`
(low)
### Settings / models Java cannot build or read
- [ ] **sqs** `SqsSourceSettings.scala:113` —
`withVisibilityTimeout(FiniteDuration)` only
- [ ] **hdfs** `model.scala:209` — `RotationStrategy.time(FiniteDuration)`
only
- [ ] **s3** `S3Headers.scala:86` — `withCustomHeaders(scala Map)` only
- [ ] **s3** `settings.scala:789` — "Java API" `S3Settings.create(...,
allowedHeaders: Map[String, Set[String]])`; no `withAllowedHeaders` / Java
getter
- [ ] **s3** `model.scala:136` — `BucketVersioningResult.getStatus` returns
`scala.Option`
- [ ] **google-common** `GoogleSettings.scala:170` — `getForwardProxy`
returns `scala.Option`
- [ ] **google-cloud-storage** `StorageObject.scala:441` — "Java API"
`create` takes `Optional[Boolean]`, `Optional[scala Map]`, `Optional[Int]`,
`Optional[scala List]`
- [ ] **google-cloud-storage** `Owner.scala`, `ProjectTeam.scala`,
`CustomerEncryption.scala`, `ObjectAccessControls.scala` — constructor params
are not vals / no getters; contents unreadable
- [ ] **huawei-push-kit** `models/NotificationTarget.scala:21`
(`Tokens(Seq)`), `AndroidConfig.scala:98-123`, `WebConfig.scala:24-64` — Scala
`Seq`/`Map` only (repo Java example uses `new Set.Set1<>(...).toSeq()`)
- [ ] **google-fcm** `models/FcmNotification.scala:43`,
`AndroidConfig.scala:37,98,102,125`, `ApnsConfig.scala:24`,
`WebPushConfig.scala:24` — Scala `Seq`/`Map` only
- [ ] **solr** `SolrMessages.scala:122` — `WriteResult` (javadsl flow
output) exposes only `Option`/`Map`
- [ ] **google-cloud-bigquery-storage** `BigQueryRecordMapImpl.scala:20` —
`BigQueryRecord.get` returns `Option`
- [ ] **mqtt-streaming** `model.scala:1185` — `Event.carry: Option`;
`model.scala:307` — `Publish.packetId: Option[PacketId]` (`@InternalApi` value
class; Java test does `packetId().get().underlying()`)
## 3. Medium
### Scala primitives in generic positions (erase to `Object`)
- [ ] **couchbase** `javadsl/CouchbaseSession.scala:252,262` — `counter:
CompletionStage[Long]`
- [ ] **couchbase3** `javadsl/CouchbaseFlow.scala:367,378` — `exists:
Flow[T, Boolean, _]` (`CouchbaseSink.exists` uses `java.lang.Boolean`)
- [ ] **elasticsearch** `ReadResult.scala:32` — `getVersion: Optional[Long]`
- [ ] **google-cloud-bigquery** `model/QueryJsonProtocol.scala:66-67` —
`getMaximumBytesBilled: Optional[Long]`; `getLabels` returns `Optional[scala
Map]` (also `JobJsonProtocol.scala:88`)
- [ ] **s3** `model.scala:139,187` — `getMfaDelete` / `create(mfaDelete:
Optional[Boolean])`
- [ ] **google-cloud-storage** `StorageObject.scala:106-107` —
`getTemporaryHold` / `getEventBasedHold: Optional[Boolean]`
- [ ] **amqp** `AmqpConnectionProvider.scala:94,224,337` —
`java.util.List[Pair[String, Int]]`
- [ ] **unix-domain-socket** `javadsl/UnixDomainSocket.scala:50` —
`unbind(): CompletionStage[Unit]` → `BoxedUnit`
### Scala default args / implicit param lists in javadsl
- [ ] **couchbase3** `javadsl/CouchbaseSource.scala` — every method
(defaults + `implicit` collection/scope); give it the explicit overloads
`CouchbaseFlow`/`CouchbaseSink` already have
- [ ] **s3** `javadsl/S3.scala:68` `request`;
`:1102,1118,1438,1453,1504,1518` `completeMultipartUpload`, `deleteUpload`,
`putBucketVersioning` (`(implicit system, attributes = Attributes())`, no `(…,
system)` overload unlike siblings)
- [ ] **amqp** `javadsl/AmqpRpcFlow.scala:79` `committableFlow(...,
repliesPerMessage = 1)`
- [ ] **orientdb** `javadsl/OrientDbSource.scala:53` `typed(..., query =
null)`
- [ ] **mongodb** `javadsl/MongoFlow.scala:161` `replaceOne(collection,
options = ...)` with no 1-arg overload
- [ ] Redundant defaults: hdfs `HdfsSource.scala:81`, ftp
`FtpApi.scala:182,314,407,501`, mongodb `MongoFlow.scala:111`, s3 `S3.scala:628`
### Element/model getters with no Java alternative (on types Java users
receive)
- [ ] **google-cloud-pub-sub** `model.scala:96` —
`PubSubMessage.data/attributes/orderingKey`
- [ ] **influxdb** `InfluxDbWriteMessage.scala:86` —
`InfluxDbWriteResult.error`
- [ ] **kinesis** `CommittableRecord.scala:39` — `shutdownReason`
- [ ] **elasticsearch** `WriteMessage.scala:44-50` —
id/source/version/indexName/customMetadata
- [ ] **file** `model.scala:75` — `TarArchiveMetadata.filePathPrefix`
- [ ] **ftp** `model.scala:36` — `FtpFile.permissions: scala Set`, no factory
- [ ] **xml** `model.scala:119` `findAttribute: Option`; `:96`
`attributesList`
- [ ] **mqtt** `model.scala:20` — `MqttMessage.qos` (mqttv5 has `getQoS`)
- [ ] **mqtt-streaming** `model.scala:378,409,442` —
`Subscribe.topicFilters`, `SubAck.returnCodes`, `Unsubscribe.topicFilters`;
`Connect` can't set keepAlive/will from Java
- [ ] **couchbase3** `Document.scala:22,29` — `MutationDocument` /
`MutationBinaryDocument` need `scala.Option` to construct; add `create` +
`getResult: Optional`
- [ ] **solr** `SolrMessages.scala:61` — `WriteMessage` accessors
- [ ] **sqs** `SqsSourceSettings.scala:28` getters; `SqsModel.scala:142,155`
`getResult` erases to `Object`; `SqsModel.scala:152` typo `geResponseMetadata`
- [ ] **huawei-push-kit** `HmsSettings.scala:158,203` —
`getForwardProxy`/`ForwardProxy.create` use `Option`
### Missing parity with scaladsl
- [ ] **s3** — no `chunkSize` / `chunkingParallelism` on javadsl
`multipartUpload*`, `resumeMultipartUpload*`, `multipartCopy`
- [ ] **file** — no javadsl `LogRotatorSink.withTypedSinkFactory`
- [ ] **slick** — no `forDbAndProfile`, no `SlickWithTryResult` equivalent,
`source` only takes a raw SQL string
- [ ] **sse** `javadsl/EventSource.scala:87` — no `retryDelay`
- [ ] **unix-domain-socket** — no javadsl `bindAndHandle`
- [ ] **google-cloud-bigquery-storage** — no javadsl way to supply a custom
`GrpcBigQueryStorageReader`
- [ ] **google-cloud-bigquery** `javadsl/BigQuery.scala:441` —
`createLoadJob[Job]` shadows `model.Job` and `BigQueryMarshallers` has no `Job`
marshaller
### Other
- [ ] **cassandra** `javadsl/CassandraSession.scala:60,79` constructors take
`ExecutionContext`; `CqlSessionProvider.connect()(implicit ec): Future` SPI
- [ ] **amqp** `AmqpConnectionProvider.scala:257` —
`withSSLContext(Option[SSLContext])` only
- [ ] **ironmq** `IronMqSettings.scala:35` `withEndpoint(scaladsl Uri)`
only; `:69` `ConsumerSettings` only reachable via
`IronMqSettings.ConsumerSettings$.MODULE$`
- [ ] **google-cloud-pub-sub-grpc** `PubSubSettings.scala` — no
`withUseTls`, `create(host, port)` hard-codes `useTls = false`
- [ ] **kinesis** `javadsl/KinesisSchedulerSource.scala:31` —
`SchedulerBuilder` is an abstract class, so no lambdas
- [ ] **elasticsearch** `ElasticsearchWriteSettings.scala:19` — `RetryLogic`
SPI requires `FiniteDuration`
- [ ] **s3** `settings.scala:302` — `RetrySettings.default` is uncallable
from Java (`default` is a keyword); no `create(Config)`
- [ ] **s3** `S3Headers.scala:34` — `MetaHeaders.withMetaHeaders(scala Map)`
only
- [ ] **google-cloud-storage** `StorageObject.scala:153,158` —
`withMetadata` / `withAcl` Scala collections only
- [ ] **jms / jakartams** `JmsMessages.scala:881` —
`JmsObjectMessagePassThrough.withProperties(scala Map)` only
- [ ] **huawei-push-kit** `NotificationTarget.scala:35`, **google-fcm**
`NotificationTarget.scala:38` — `ConditionBuilder` only has symbolic `&&` /
`||` / `unary_!`; add `and` / `or` / `not`
- [ ] **slick** `javadsl/package.scala:73` — `nextBytes():
Array[java.lang.Byte]` (boxed `Byte[]`)
- [ ] **mqtt-streaming** `model.scala:97-259` — value classes (incl.
`@InternalApi` `PacketId`, `ControlPacketFlags`) erase to `int` in public
constructors
## 4. Low (summary)
About 120 smaller items, mostly:
- Settings getters returning `Option` / `Seq` / `FiniteDuration` / Scala
`Map` with no `getX` returning `Optional` / `java.util.*` /
`java.time.Duration` (amqp, cassandra, couchbase, elasticsearch, ftp, hbase,
influxdb, ironmq, jms, kinesis, mqtt, mqttv5, mqtt-streaming, s3, sqs,
google-fcm).
- Naming inconsistencies: `apply` vs `create` (kinesisfirehose, influxdb
`Default()`), `grouped` vs `createGrouped` (sqs),
`getObjectMetadataWithHeaders` (s3), `createTableListResponse` (bigquery),
`withOrientDBCredentials(ODatabasePool)` (orientdb).
- Missing convenience overloads (amqp `createSimple(settings)`, pub-sub
`publish` without parallelism, GCS bucket ops without `attributes`, sse without
`lastEventId`, xml `Namespace.create(uri)`, `parserWithContext(Consumer)`).
- `CompletableFuture` vs `CompletionStage` inconsistency (pub-sub-grpc
`subscribe`), `CompletionStage[NotUsed]` mat values (bigquery-storage).
- Deprecated Scala-`Future` variants still present (`subscribeFlowFuture`,
`createShardedFuture`) — candidates for removal in 2.0.0.
- Doc slips (text `TextFlow.encoding` scaladoc), redundant `ActorSystem`
overloads (udp), undeclared return types on public methods (reference, udp).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
