You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by re...@apache.org on 2023/03/11 16:53:36 UTC
[beam] branch master updated: Merge pull request #25790: update to new BOM
This is an automated email from the ASF dual-hosted git repository.
reuvenlax pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new fb040ff9c10 Merge pull request #25790: update to new BOM
fb040ff9c10 is described below
commit fb040ff9c1059f97a29096597b42e864d8c87a82
Author: Reuven Lax <re...@google.com>
AuthorDate: Sat Mar 11 08:53:21 2023 -0800
Merge pull request #25790: update to new BOM
---
.../org/apache/beam/gradle/BeamModulePlugin.groovy | 10 +++----
.../container/license_scripts/dep_urls_java.yaml | 2 +-
.../io/gcp/pubsublite/internal/PubsubLiteSink.java | 3 +--
.../pubsublite/internal/SubscribeTransform.java | 6 +----
.../pubsublite/internal/PubsubLiteSinkTest.java | 31 ++++++++++++----------
5 files changed, 25 insertions(+), 27 deletions(-)
diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index 8c4071b70c7..7e8ef20953e 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -466,15 +466,15 @@ class BeamModulePlugin implements Plugin<Project> {
def dbcp2_version = "2.8.0"
def errorprone_version = "2.10.0"
// Try to keep gax_version consistent with gax-grpc version in google_cloud_platform_libraries_bom
- def gax_version = "2.23.0"
+ def gax_version = "2.23.2"
def google_clients_version = "2.0.0"
def google_cloud_bigdataoss_version = "2.2.6"
// Try to keep google_cloud_spanner_version consistent with google_cloud_spanner_bom in google_cloud_platform_libraries_bom
- def google_cloud_spanner_version = "6.36.0"
+ def google_cloud_spanner_version = "6.37.0"
def google_code_gson_version = "2.9.1"
def google_oauth_clients_version = "1.34.1"
// Try to keep grpc_version consistent with gRPC version in google_cloud_platform_libraries_bom
- def grpc_version = "1.52.1"
+ def grpc_version = "1.53.0"
def guava_version = "31.1-jre"
def hadoop_version = "2.10.2"
def hamcrest_version = "2.1"
@@ -609,9 +609,9 @@ class BeamModulePlugin implements Plugin<Project> {
google_cloud_pubsub : "com.google.cloud:google-cloud-pubsub", // google_cloud_platform_libraries_bom sets version
google_cloud_pubsublite : "com.google.cloud:google-cloud-pubsublite", // google_cloud_platform_libraries_bom sets version
// The release notes shows the versions set by the BOM:
- // https://github.com/googleapis/java-cloud-bom/releases/tag/v26.8.0
+ // https://github.com/googleapis/java-cloud-bom/releases/tag/v26.10.0
// Update libraries-bom version on sdks/java/container/license_scripts/dep_urls_java.yaml
- google_cloud_platform_libraries_bom : "com.google.cloud:libraries-bom:26.8.0",
+ google_cloud_platform_libraries_bom : "com.google.cloud:libraries-bom:26.10.0",
google_cloud_spanner : "com.google.cloud:google-cloud-spanner", // google_cloud_platform_libraries_bom sets version
google_cloud_spanner_test : "com.google.cloud:google-cloud-spanner:$google_cloud_spanner_version:tests",
google_code_gson : "com.google.code.gson:gson:$google_code_gson_version",
diff --git a/sdks/java/container/license_scripts/dep_urls_java.yaml b/sdks/java/container/license_scripts/dep_urls_java.yaml
index 2a46fd84477..acf8a6bef23 100644
--- a/sdks/java/container/license_scripts/dep_urls_java.yaml
+++ b/sdks/java/container/license_scripts/dep_urls_java.yaml
@@ -42,7 +42,7 @@ jaxen:
'1.1.6':
type: "3-Clause BSD"
libraries-bom:
- '26.8.0':
+ '26.10.0':
license: "https://raw.githubusercontent.com/GoogleCloudPlatform/cloud-opensource-java/master/LICENSE"
type: "Apache License 2.0"
paranamer:
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSink.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSink.java
index f22a3204021..dbf312ae5a4 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSink.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSink.java
@@ -22,7 +22,6 @@ import static java.util.concurrent.TimeUnit.MINUTES;
import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.api.gax.rpc.ApiException;
-import com.google.cloud.pubsublite.Message;
import com.google.cloud.pubsublite.MessageMetadata;
import com.google.cloud.pubsublite.internal.CheckedApiException;
import com.google.cloud.pubsublite.internal.Publisher;
@@ -59,7 +58,7 @@ public class PubsubLiteSink extends DoFn<PubSubMessage, Void> {
}
void publish(PubSubMessage message) {
- futures.add(publisher.publish(Message.fromProto(message)));
+ futures.add(publisher.publish(message));
}
void waitForDone() throws Exception {
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/SubscribeTransform.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/SubscribeTransform.java
index 493d7a72d7c..f3ffbb13c24 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/SubscribeTransform.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/SubscribeTransform.java
@@ -78,11 +78,7 @@ public class SubscribeTransform extends PTransform<PBegin, PCollection<Sequenced
return new SubscriberAssembler(options, partition)
.getSubscriberFactory(initialOffset)
.newSubscriber(
- messages ->
- consumer.accept(
- messages.stream()
- .map(message -> message.toProto())
- .collect(Collectors.toList())));
+ messages -> consumer.accept(messages.stream().collect(Collectors.toList())));
} catch (Throwable t) {
throw toCanonical(t).underlying;
}
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSinkTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSinkTest.java
index 5c4e731a212..ee277b8f8ce 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSinkTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSinkTest.java
@@ -32,7 +32,6 @@ import com.google.api.core.SettableApiFuture;
import com.google.api.gax.rpc.StatusCode.Code;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudZone;
-import com.google.cloud.pubsublite.Message;
import com.google.cloud.pubsublite.MessageMetadata;
import com.google.cloud.pubsublite.Offset;
import com.google.cloud.pubsublite.Partition;
@@ -43,6 +42,7 @@ import com.google.cloud.pubsublite.internal.CheckedApiException;
import com.google.cloud.pubsublite.internal.ExtractStatus;
import com.google.cloud.pubsublite.internal.Publisher;
import com.google.cloud.pubsublite.internal.testing.FakeApiService;
+import com.google.cloud.pubsublite.proto.PubSubMessage;
import com.google.protobuf.ByteString;
import java.util.Arrays;
import java.util.Optional;
@@ -88,12 +88,12 @@ public class PubsubLiteSinkTest {
private final PubsubLiteSink sink = new PubsubLiteSink(defaultOptions());
@Captor
- final ArgumentCaptor<Message> publishedMessageCaptor = ArgumentCaptor.forClass(Message.class);
+ final ArgumentCaptor<PubSubMessage> publishedMessageCaptor =
+ ArgumentCaptor.forClass(PubSubMessage.class);
- private void runWith(Message... messages) {
+ private void runWith(PubSubMessage... messages) {
pipeline
- .apply(
- Create.of(Arrays.stream(messages).map(Message::toProto).collect(Collectors.toList())))
+ .apply(Create.of(Arrays.stream(messages).collect(Collectors.toList())))
.apply(ParDo.of(sink));
pipeline.run();
}
@@ -106,16 +106,17 @@ public class PubsubLiteSinkTest {
@Test
public void singleMessagePublishes() throws Exception {
- when(publisher.publish(Message.builder().build()))
+ when(publisher.publish(PubSubMessage.newBuilder().build()))
.thenReturn(ApiFutures.immediateFuture(MessageMetadata.of(Partition.of(1), Offset.of(2))));
- runWith(Message.builder().build());
- verify(publisher).publish(Message.builder().build());
+ runWith(PubSubMessage.newBuilder().build());
+ verify(publisher).publish(PubSubMessage.newBuilder().build());
}
@Test
public void manyMessagePublishes() throws Exception {
- Message message1 = Message.builder().build();
- Message message2 = Message.builder().setKey(ByteString.copyFromUtf8("abc")).build();
+ PubSubMessage message1 = PubSubMessage.newBuilder().build();
+ PubSubMessage message2 =
+ PubSubMessage.newBuilder().setKey(ByteString.copyFromUtf8("abc")).build();
when(publisher.publish(message1))
.thenReturn(ApiFutures.immediateFuture(MessageMetadata.of(Partition.of(1), Offset.of(2))));
when(publisher.publish(message2))
@@ -127,7 +128,7 @@ public class PubsubLiteSinkTest {
@Test
public void singleExceptionWhenProcessing() {
- Message message1 = Message.builder().build();
+ PubSubMessage message1 = PubSubMessage.newBuilder().build();
when(publisher.publish(message1))
.thenReturn(
ApiFutures.immediateFailedFuture(new CheckedApiException(Code.INTERNAL).underlying));
@@ -141,9 +142,11 @@ public class PubsubLiteSinkTest {
@Test
public void exceptionMixedWithOK() throws Exception {
- Message message1 = Message.builder().build();
- Message message2 = Message.builder().setKey(ByteString.copyFromUtf8("abc")).build();
- Message message3 = Message.builder().setKey(ByteString.copyFromUtf8("def")).build();
+ PubSubMessage message1 = PubSubMessage.newBuilder().build();
+ PubSubMessage message2 =
+ PubSubMessage.newBuilder().setKey(ByteString.copyFromUtf8("abc")).build();
+ PubSubMessage message3 =
+ PubSubMessage.newBuilder().setKey(ByteString.copyFromUtf8("def")).build();
SettableApiFuture<MessageMetadata> future1 = SettableApiFuture.create();
SettableApiFuture<MessageMetadata> future2 = SettableApiFuture.create();
SettableApiFuture<MessageMetadata> future3 = SettableApiFuture.create();