This is an automated email from the ASF dual-hosted git repository.
He-Pin 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 32019348 refactor: use Optional.stream() (Java 9+) (#576)
32019348 is described below
commit 320193485fa7feea9397af053f9848a3ac5ce8be
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Jun 26 21:06:17 2026 +0800
refactor: use Optional.stream() (Java 9+) (#576)
Motivation:
Replace Java 8-era Optional-to-Stream conversion pattern with the
built-in Optional.stream() method available since Java 9.
Modification:
Replace the manual isPresent/get/Stream.of pattern with option.stream().
Result:
Simpler, more idiomatic code using the standard library API.
---
.../pekko/kafka/testkit/internal/KafkaContainerCluster.java | 12 +++---------
1 file changed, 3 insertions(+), 9 deletions(-)
diff --git
a/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
b/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
index 1e154198..0bd1cf46 100644
---
a/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
+++
b/testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java
@@ -188,16 +188,10 @@ public class KafkaContainerCluster implements Startable {
.collect(Collectors.joining(","));
}
- /** for backwards compatibility with Java 8 */
- private <T> Stream<T> optionalStream(Optional<T> option) {
- if (option.isPresent()) return Stream.of(option.get());
- else return Stream.empty();
- }
-
private Stream<GenericContainer> allContainers() {
return Stream.concat(
- Stream.concat(this.brokers.stream(), optionalStream(this.zookeeper)),
- optionalStream(this.schemaRegistry));
+ Stream.concat(this.brokers.stream(), this.zookeeper.stream()),
+ this.schemaRegistry.stream());
}
@Override
@@ -219,7 +213,7 @@ public class KafkaContainerCluster implements Startable {
waitForClusterFormation();
// start schema registry if the container is initialized
- Startables.deepStart(optionalStream(this.schemaRegistry))
+ Startables.deepStart(this.schemaRegistry.stream())
.get(clusterStartTimeout.getSeconds(), SECONDS);
} catch (Exception ex) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
