This is an automated email from the ASF dual-hosted git repository.
ahmedabualsaud 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 3b9d86d6fa6 feat: add pubsub topic validation (#32582)
3b9d86d6fa6 is described below
commit 3b9d86d6fa6f960283e46dece4aa2a21ed84bb96
Author: Hyeonho Kim <[email protected]>
AuthorDate: Fri Oct 18 06:05:37 2024 +0900
feat: add pubsub topic validation (#32582)
---
CHANGES.md | 4 +
.../beam/sdk/io/gcp/pubsub/PubsubClient.java | 3 +
.../beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java | 16 ++
.../apache/beam/sdk/io/gcp/pubsub/PubsubIO.java | 89 +++++++++++
.../beam/sdk/io/gcp/pubsub/PubsubJsonClient.java | 14 ++
.../beam/sdk/io/gcp/pubsub/PubsubTestClient.java | 6 +
.../sdk/io/gcp/pubsub/PubsubGrpcClientTest.java | 40 +++++
.../beam/sdk/io/gcp/pubsub/PubsubIOTest.java | 170 +++++++++++++++++++++
.../sdk/io/gcp/pubsub/PubsubJsonClientTest.java | 24 +++
9 files changed, 366 insertions(+)
diff --git a/CHANGES.md b/CHANGES.md
index 840c60fc114..766f74fc3be 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -106,6 +106,10 @@
* [Managed Iceberg] Added auto-sharding for streaming writes
([#32612](https://github.com/apache/beam/pull/32612))
* [Managed Iceberg] Added support for writing to dynamic destinations
([#32565](https://github.com/apache/beam/pull/32565))
+## I/Os
+
+* PubsubIO can validate that the Pub/Sub topic exists before running the
Read/Write pipeline (Java) ([#32465](https://github.com/apache/beam/pull/32465))
+
## New Features / Improvements
* Dataflow worker can install packages from Google Artifact Registry Python
repositories (Python) ([#32123](https://github.com/apache/beam/issues/32123)).
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubClient.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubClient.java
index 2964a29dbb6..bd01604643e 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubClient.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubClient.java
@@ -507,6 +507,9 @@ public abstract class PubsubClient implements Closeable {
/** Return a list of topics for {@code project}. */
public abstract List<TopicPath> listTopics(ProjectPath project) throws
IOException;
+ /** Return {@literal true} if {@code topic} exists. */
+ public abstract boolean isTopicExists(TopicPath topic) throws IOException;
+
/** Create {@code subscription} to {@code topic}. */
public abstract void createSubscription(
TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds)
throws IOException;
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java
index 93fdd552400..0cfb0668810 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClient.java
@@ -54,6 +54,7 @@ import com.google.pubsub.v1.Topic;
import io.grpc.Channel;
import io.grpc.ClientInterceptors;
import io.grpc.ManagedChannel;
+import io.grpc.StatusRuntimeException;
import io.grpc.auth.ClientAuthInterceptor;
import io.grpc.netty.GrpcSslContexts;
import io.grpc.netty.NegotiationType;
@@ -372,6 +373,21 @@ public class PubsubGrpcClient extends PubsubClient {
return topics;
}
+ @Override
+ public boolean isTopicExists(TopicPath topic) throws IOException {
+ GetTopicRequest request =
GetTopicRequest.newBuilder().setTopic(topic.getPath()).build();
+ try {
+ publisherStub().getTopic(request);
+ return true;
+ } catch (StatusRuntimeException e) {
+ if (e.getStatus().getCode() == io.grpc.Status.Code.NOT_FOUND) {
+ return false;
+ }
+
+ throw e;
+ }
+ }
+
@Override
public void createSubscription(
TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds)
throws IOException {
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
index f59a68c4055..c6c8b3e7181 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
@@ -50,6 +50,7 @@ import
org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.OutgoingMessage;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.SubscriptionPath;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.TopicPath;
import org.apache.beam.sdk.metrics.Lineage;
+import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
@@ -860,6 +861,8 @@ public class PubsubIO {
abstract ErrorHandler<BadRecord, ?> getBadRecordErrorHandler();
+ abstract boolean getValidate();
+
abstract Builder<T> toBuilder();
static <T> Builder<T> newBuilder(SerializableFunction<PubsubMessage, T>
parseFn) {
@@ -871,6 +874,7 @@ public class PubsubIO {
builder.setNeedsOrderingKey(false);
builder.setBadRecordRouter(BadRecordRouter.THROWING_ROUTER);
builder.setBadRecordErrorHandler(new DefaultErrorHandler<>());
+ builder.setValidate(false);
return builder;
}
@@ -918,6 +922,8 @@ public class PubsubIO {
abstract Builder<T> setBadRecordErrorHandler(
ErrorHandler<BadRecord, ?> badRecordErrorHandler);
+ abstract Builder<T> setValidate(boolean validation);
+
abstract Read<T> build();
}
@@ -1097,6 +1103,11 @@ public class PubsubIO {
.build();
}
+ /** Enable validation of the PubSub Read. */
+ public Read<T> withValidation() {
+ return toBuilder().setValidate(true).build();
+ }
+
@VisibleForTesting
/**
* Set's the internal Clock.
@@ -1262,6 +1273,35 @@ public class PubsubIO {
return read.setCoder(getCoder());
}
+ @Override
+ public void validate(PipelineOptions options) {
+ if (!getValidate()) {
+ return;
+ }
+
+ PubsubOptions psOptions = options.as(PubsubOptions.class);
+
+ // Validate the existence of the topic.
+ if (getTopicProvider() != null) {
+ PubsubTopic topic = getTopicProvider().get();
+ boolean topicExists = true;
+ try (PubsubClient pubsubClient =
+ getPubsubClientFactory()
+ .newClient(getTimestampAttribute(), getIdAttribute(),
psOptions)) {
+ topicExists =
+ pubsubClient.isTopicExists(
+ PubsubClient.topicPathFromName(topic.project, topic.topic));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ if (!topicExists) {
+ throw new IllegalArgumentException(
+ String.format("Pubsub topic '%s' does not exist.", topic));
+ }
+ }
+ }
+
@Override
public void populateDisplayData(DisplayData.Builder builder) {
super.populateDisplayData(builder);
@@ -1341,6 +1381,8 @@ public class PubsubIO {
abstract ErrorHandler<BadRecord, ?> getBadRecordErrorHandler();
+ abstract boolean getValidate();
+
abstract Builder<T> toBuilder();
static <T> Builder<T> newBuilder(
@@ -1350,6 +1392,7 @@ public class PubsubIO {
builder.setFormatFn(formatFn);
builder.setBadRecordRouter(BadRecordRouter.THROWING_ROUTER);
builder.setBadRecordErrorHandler(new DefaultErrorHandler<>());
+ builder.setValidate(false);
return builder;
}
@@ -1386,6 +1429,8 @@ public class PubsubIO {
abstract Builder<T> setBadRecordErrorHandler(
ErrorHandler<BadRecord, ?> badRecordErrorHandler);
+ abstract Builder<T> setValidate(boolean validation);
+
abstract Write<T> build();
}
@@ -1396,11 +1441,14 @@ public class PubsubIO {
* {@code topic} string.
*/
public Write<T> to(String topic) {
+ ValueProvider<String> topicProvider = StaticValueProvider.of(topic);
+ validateTopic(topicProvider);
return to(StaticValueProvider.of(topic));
}
/** Like {@code topic()} but with a {@link ValueProvider}. */
public Write<T> to(ValueProvider<String> topic) {
+ validateTopic(topic);
return toBuilder()
.setTopicProvider(NestedValueProvider.of(topic,
PubsubTopic::fromPath))
.setTopicFunction(null)
@@ -1408,6 +1456,13 @@ public class PubsubIO {
.build();
}
+ /** Handles validation of {@code topic}. */
+ private static void validateTopic(ValueProvider<String> topic) {
+ if (topic.isAccessible()) {
+ PubsubTopic.fromPath(topic.get());
+ }
+ }
+
/**
* Provides a function to dynamically specify the target topic per
message. Not compatible with
* any of the other to methods. If {@link #to} is called again specifying
a topic, then this
@@ -1497,6 +1552,11 @@ public class PubsubIO {
.build();
}
+ /** Enable validation of the PubSub Write. */
+ public Write<T> withValidation() {
+ return toBuilder().setValidate(true).build();
+ }
+
@Override
public PDone expand(PCollection<T> input) {
if (getTopicProvider() == null && !getDynamicDestinations()) {
@@ -1566,6 +1626,35 @@ public class PubsubIO {
throw new RuntimeException(); // cases are exhaustive.
}
+ @Override
+ public void validate(PipelineOptions options) {
+ if (!getValidate()) {
+ return;
+ }
+
+ PubsubOptions psOptions = options.as(PubsubOptions.class);
+
+ // Validate the existence of the topic.
+ if (getTopicProvider() != null) {
+ PubsubTopic topic = getTopicProvider().get();
+ boolean topicExists = true;
+ try (PubsubClient pubsubClient =
+ getPubsubClientFactory()
+ .newClient(getTimestampAttribute(), getIdAttribute(),
psOptions)) {
+ topicExists =
+ pubsubClient.isTopicExists(
+ PubsubClient.topicPathFromName(topic.project, topic.topic));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ if (!topicExists) {
+ throw new IllegalArgumentException(
+ String.format("Pubsub topic '%s' does not exist.", topic));
+ }
+ }
+ }
+
@Override
public void populateDisplayData(DisplayData.Builder builder) {
super.populateDisplayData(builder);
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClient.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClient.java
index 386febcf005..0a838da66f6 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClient.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClient.java
@@ -19,6 +19,7 @@ package org.apache.beam.sdk.io.gcp.pubsub;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
+import com.google.api.client.googleapis.json.GoogleJsonResponseException;
import com.google.api.client.http.HttpRequestInitializer;
import com.google.api.services.pubsub.Pubsub;
import com.google.api.services.pubsub.Pubsub.Projects.Subscriptions;
@@ -310,6 +311,19 @@ public class PubsubJsonClient extends PubsubClient {
return topics;
}
+ @Override
+ public boolean isTopicExists(TopicPath topic) throws IOException {
+ try {
+ pubsub.projects().topics().get(topic.getPath()).execute();
+ return true;
+ } catch (GoogleJsonResponseException e) {
+ if (e.getStatusCode() == 404) {
+ return false;
+ }
+ throw e;
+ }
+ }
+
@Override
public void createSubscription(
TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds)
throws IOException {
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java
index a8109d05ec3..3d5a879fce1 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubTestClient.java
@@ -605,6 +605,12 @@ public class PubsubTestClient extends PubsubClient
implements Serializable {
throw new UnsupportedOperationException();
}
+ @Override
+ public boolean isTopicExists(TopicPath topic) throws IOException {
+ // Always return true for testing purposes.
+ return true;
+ }
+
@Override
public void createSubscription(
TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds)
throws IOException {
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClientTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClientTest.java
index 3724e169c61..6c4625f2e07 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClientTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubGrpcClientTest.java
@@ -40,6 +40,7 @@ import com.google.pubsub.v1.SubscriberGrpc.SubscriberImplBase;
import com.google.pubsub.v1.Topic;
import io.grpc.ManagedChannel;
import io.grpc.Server;
+import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
@@ -432,4 +433,43 @@ public class PubsubGrpcClientTest {
server.shutdownNow();
}
}
+
+ @Test
+ public void isTopicExists() throws IOException {
+ initializeClient(null, null);
+ TopicPath topicDoesNotExist =
+
PubsubClient.topicPathFromPath("projects/testProject/topics/dontexist");
+ TopicPath topicExists =
PubsubClient.topicPathFromPath("projects/testProject/topics/exist");
+
+ PublisherImplBase publisherImplBase =
+ new PublisherImplBase() {
+ @Override
+ public void getTopic(GetTopicRequest request, StreamObserver<Topic>
responseObserver) {
+ String topicPath = request.getTopic();
+ if (topicPath.equals(topicDoesNotExist.getPath())) {
+ responseObserver.onError(
+ new
StatusRuntimeException(Status.fromCode(Status.Code.NOT_FOUND)));
+ }
+ if (topicPath.equals(topicExists.getPath())) {
+ responseObserver.onNext(
+ Topic.newBuilder()
+ .setName(topicPath)
+ .setSchemaSettings(
+
SchemaSettings.newBuilder().setSchema(SCHEMA.getPath()).build())
+ .build());
+ responseObserver.onCompleted();
+ }
+ }
+ };
+ Server server =
+
InProcessServerBuilder.forName(channelName).addService(publisherImplBase).build().start();
+ try {
+ assertEquals(false, client.isTopicExists(topicDoesNotExist));
+
+ assertEquals(true, client.isTopicExists(topicExists));
+
+ } finally {
+ server.shutdownNow();
+ }
+ }
}
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
index d4effbae40a..bec157ae83c 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
@@ -83,6 +83,7 @@ import org.apache.beam.sdk.values.ValueInSingleWindow;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
+import org.apache.commons.lang3.RandomStringUtils;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
@@ -97,6 +98,7 @@ import org.junit.runner.Description;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.junit.runners.model.Statement;
+import org.mockito.Mockito;
/** Tests for PubsubIO Read and Write transforms. */
@RunWith(JUnit4.class)
@@ -928,4 +930,172 @@ public class PubsubIOTest {
pipeline.run();
}
}
+
+ @Test
+ public void testReadValidate() throws IOException {
+ PubsubOptions options =
TestPipeline.testingPipelineOptions().as(PubsubOptions.class);
+ TopicPath existingTopic = PubsubClient.topicPathFromName("test-project",
"testTopic");
+ PubsubClient mockClient = Mockito.mock(PubsubClient.class);
+ Mockito.when(mockClient.isTopicExists(existingTopic)).thenReturn(true);
+ PubsubClient.PubsubClientFactory mockFactory =
+ Mockito.mock(PubsubClient.PubsubClientFactory.class);
+ Mockito.when(mockFactory.newClient("myTimestamp", "myId",
options)).thenReturn(mockClient);
+
+ Read<PubsubMessage> read =
+ Read.newBuilder()
+ .setTopicProvider(
+ StaticValueProvider.of(
+
PubsubIO.PubsubTopic.fromPath("projects/test-project/topics/testTopic")))
+ .setTimestampAttribute("myTimestamp")
+ .setIdAttribute("myId")
+ .setPubsubClientFactory(mockFactory)
+ .setCoder(PubsubMessagePayloadOnlyCoder.of())
+ .setValidate(true)
+ .build();
+
+ read.validate(options);
+ }
+
+ @Test
+ public void testReadValidateTopicIsNotExists() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+
+ PubsubOptions options =
TestPipeline.testingPipelineOptions().as(PubsubOptions.class);
+ TopicPath nonExistingTopic =
PubsubClient.topicPathFromName("test-project", "nonExistingTopic");
+ PubsubClient mockClient = Mockito.mock(PubsubClient.class);
+ Mockito.when(mockClient.isTopicExists(nonExistingTopic)).thenReturn(false);
+ PubsubClient.PubsubClientFactory mockFactory =
+ Mockito.mock(PubsubClient.PubsubClientFactory.class);
+ Mockito.when(mockFactory.newClient("myTimestamp", "myId",
options)).thenReturn(mockClient);
+
+ Read<PubsubMessage> read =
+ Read.newBuilder()
+ .setTopicProvider(
+ StaticValueProvider.of(
+
PubsubIO.PubsubTopic.fromPath("projects/test-project/topics/nonExistingTopic")))
+ .setTimestampAttribute("myTimestamp")
+ .setIdAttribute("myId")
+ .setPubsubClientFactory(mockFactory)
+ .setCoder(PubsubMessagePayloadOnlyCoder.of())
+ .setValidate(true)
+ .build();
+
+ read.validate(options);
+ }
+
+ @Test
+ public void testReadWithoutValidation() throws IOException {
+ PubsubOptions options =
TestPipeline.testingPipelineOptions().as(PubsubOptions.class);
+ TopicPath nonExistingTopic =
PubsubClient.topicPathFromName("test-project", "nonExistingTopic");
+ PubsubClient mockClient = Mockito.mock(PubsubClient.class);
+ Mockito.when(mockClient.isTopicExists(nonExistingTopic)).thenReturn(false);
+ PubsubClient.PubsubClientFactory mockFactory =
+ Mockito.mock(PubsubClient.PubsubClientFactory.class);
+ Mockito.when(mockFactory.newClient("myTimestamp", "myId",
options)).thenReturn(mockClient);
+
+ Read<PubsubMessage> read =
+
PubsubIO.readMessages().fromTopic("projects/test-project/topics/nonExistingTopic");
+
+ read.validate(options);
+ }
+
+ @Test
+ public void testWriteTopicValidationSuccess() throws Exception {
+ PubsubIO.writeStrings().to("projects/my-project/topics/abc");
+ PubsubIO.writeStrings().to("projects/my-project/topics/ABC");
+ PubsubIO.writeStrings().to("projects/my-project/topics/AbC-DeF");
+ PubsubIO.writeStrings().to("projects/my-project/topics/AbC-1234");
+
PubsubIO.writeStrings().to("projects/my-project/topics/AbC-1234-_.~%+-_.~%+-_.~%+-abc");
+ PubsubIO.writeStrings()
+ .to(
+ new StringBuilder()
+ .append("projects/my-project/topics/A-really-long-one-")
+ .append(RandomStringUtils.randomAlphanumeric(100))
+ .toString());
+ }
+
+ @Test
+ public void testWriteTopicValidationBadCharacter() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ PubsubIO.writeStrings().to("projects/my-project/topics/abc-*-abc");
+ }
+
+ @Test
+ public void testWriteValidationTooLong() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ PubsubIO.writeStrings()
+ .to(
+ new StringBuilder()
+ .append("projects/my-project/topics/A-really-long-one-")
+ .append(RandomStringUtils.randomAlphanumeric(1000))
+ .toString());
+ }
+
+ @Test
+ public void testWriteValidate() throws IOException {
+ PubsubOptions options =
TestPipeline.testingPipelineOptions().as(PubsubOptions.class);
+ TopicPath existingTopic = PubsubClient.topicPathFromName("test-project",
"testTopic");
+ PubsubClient mockClient = Mockito.mock(PubsubClient.class);
+ Mockito.when(mockClient.isTopicExists(existingTopic)).thenReturn(true);
+ PubsubClient.PubsubClientFactory mockFactory =
+ Mockito.mock(PubsubClient.PubsubClientFactory.class);
+ Mockito.when(mockFactory.newClient("myTimestamp", "myId",
options)).thenReturn(mockClient);
+
+ PubsubIO.Write<PubsubMessage> write =
+ PubsubIO.Write.newBuilder()
+ .setTopicProvider(
+ StaticValueProvider.of(
+
PubsubIO.PubsubTopic.fromPath("projects/test-project/topics/testTopic")))
+ .setTimestampAttribute("myTimestamp")
+ .setIdAttribute("myId")
+ .setDynamicDestinations(false)
+ .setPubsubClientFactory(mockFactory)
+ .setValidate(true)
+ .build();
+
+ write.validate(options);
+ }
+
+ @Test
+ public void testWriteValidateTopicIsNotExists() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+
+ PubsubOptions options =
TestPipeline.testingPipelineOptions().as(PubsubOptions.class);
+ TopicPath nonExistingTopic =
PubsubClient.topicPathFromName("test-project", "nonExistingTopic");
+ PubsubClient mockClient = Mockito.mock(PubsubClient.class);
+ Mockito.when(mockClient.isTopicExists(nonExistingTopic)).thenReturn(false);
+ PubsubClient.PubsubClientFactory mockFactory =
+ Mockito.mock(PubsubClient.PubsubClientFactory.class);
+ Mockito.when(mockFactory.newClient("myTimestamp", "myId",
options)).thenReturn(mockClient);
+
+ PubsubIO.Write<PubsubMessage> write =
+ PubsubIO.Write.newBuilder()
+ .setTopicProvider(
+ StaticValueProvider.of(
+
PubsubIO.PubsubTopic.fromPath("projects/test-project/topics/nonExistingTopic")))
+ .setTimestampAttribute("myTimestamp")
+ .setIdAttribute("myId")
+ .setDynamicDestinations(false)
+ .setPubsubClientFactory(mockFactory)
+ .setValidate(true)
+ .build();
+
+ write.validate(options);
+ }
+
+ @Test
+ public void testWithoutValidation() throws IOException {
+ PubsubOptions options =
TestPipeline.testingPipelineOptions().as(PubsubOptions.class);
+ TopicPath nonExistingTopic =
PubsubClient.topicPathFromName("test-project", "nonExistingTopic");
+ PubsubClient mockClient = Mockito.mock(PubsubClient.class);
+ Mockito.when(mockClient.isTopicExists(nonExistingTopic)).thenReturn(false);
+ PubsubClient.PubsubClientFactory mockFactory =
+ Mockito.mock(PubsubClient.PubsubClientFactory.class);
+ Mockito.when(mockFactory.newClient("myTimestamp", "myId",
options)).thenReturn(mockClient);
+
+ PubsubIO.Write<PubsubMessage> write =
+
PubsubIO.writeMessages().to("projects/test-project/topics/nonExistingTopic");
+
+ write.validate(options);
+ }
}
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClientTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClientTest.java
index 634ad42c937..49681c86257 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClientTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubJsonClientTest.java
@@ -23,6 +23,10 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.mockito.Mockito.when;
+import com.google.api.client.googleapis.json.GoogleJsonError;
+import com.google.api.client.googleapis.json.GoogleJsonResponseException;
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpResponseException;
import com.google.api.services.pubsub.Pubsub;
import com.google.api.services.pubsub.Pubsub.Projects.Subscriptions;
import com.google.api.services.pubsub.Pubsub.Projects.Topics;
@@ -425,4 +429,24 @@ public class PubsubJsonClientTest {
IllegalArgumentException.class,
() -> client.getSchema(SCHEMA));
}
+
+ @Test
+ public void isTopicExists() throws Exception {
+ TopicPath topicExists =
+
PubsubClient.topicPathFromPath("projects/testProject/topics/topicExists");
+ TopicPath topicDoesNotExist =
+
PubsubClient.topicPathFromPath("projects/testProject/topics/topicDoesNotExist");
+ HttpResponseException.Builder builder =
+ new HttpResponseException.Builder(404, "topic is not found", new
HttpHeaders());
+ GoogleJsonError error = new GoogleJsonError();
+ when(mockPubsub.projects().topics().get(topicExists.getPath()).execute())
+ .thenReturn(new Topic().setName(topicExists.getName()));
+
when(mockPubsub.projects().topics().get(topicDoesNotExist.getPath()).execute())
+ .thenThrow(new GoogleJsonResponseException(builder, error));
+
+ client = new PubsubJsonClient(null, null, mockPubsub);
+
+ assertEquals(true, client.isTopicExists(topicExists));
+ assertEquals(false, client.isTopicExists(topicDoesNotExist));
+ }
}
