diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/AddTimestampAttribute.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/AddTimestampAttribute.java index 0ecafc82f90b..2fd3032c0da1 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/AddTimestampAttribute.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/AddTimestampAttribute.java @@ -18,6 +18,7 @@ package org.apache.beam.sdk.io.gcp.pubsub; import static org.apache.beam.sdk.io.gcp.pubsub.PubsubMessageToRow.TIMESTAMP_FIELD; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; import org.apache.beam.sdk.schemas.transforms.DropFields; import org.apache.beam.sdk.transforms.PTransform; @@ -31,9 +32,6 @@ * Adds a timestamp attribute if desired and filters it out of the underlying row if no timestamp * attribute exists. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) class AddTimestampAttribute extends PTransform, PCollection> { private static final Logger LOG = LoggerFactory.getLogger(AddTimestampAttribute.class); private final boolean useTimestampAttribute; @@ -48,7 +46,14 @@ public PCollection expand(PCollection input) { // element's event time. PubSubIO will populate the attribute from there. PCollection withTimestamp = useTimestampAttribute - ? input.apply(WithTimestamps.of((row) -> row.getDateTime(TIMESTAMP_FIELD).toInstant())) + ? input.apply( + WithTimestamps.of( + (row) -> + checkArgumentNotNull( + row.getDateTime(TIMESTAMP_FIELD), + "Field '%s' must be present and non-null in the input row when writing to PubSub with a timestamp attribute.", + TIMESTAMP_FIELD) + .toInstant())) : input; PCollection rows; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/ExternalWrite.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/ExternalWrite.java index 23b14b68a08b..3bcc83aed78b 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/ExternalWrite.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/ExternalWrite.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; + import com.google.auto.service.AutoService; import java.util.Map; import org.apache.beam.sdk.expansion.ExternalTransformRegistrar; @@ -35,9 +37,6 @@ /** Exposes {@link PubsubIO.Write} as an external transform for cross-language usage. */ @AutoService(ExternalTransformRegistrar.class) -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public final class ExternalWrite implements ExternalTransformRegistrar { public ExternalWrite() {} @@ -50,7 +49,7 @@ public ExternalWrite() {} /** Parameters class to expose the transform to an external SDK. */ public static class Configuration { - private String topic; + private @Nullable String topic; private @Nullable String idAttribute; private @Nullable String timestampAttribute; private boolean publishWithOrderingKey = false; @@ -80,8 +79,9 @@ public WriteBuilder() {} public PTransform, PDone> buildExternal(Configuration config) { PubsubIO.Write.Builder writeBuilder = PubsubIO.Write.newBuilder(new ParsePubsubMessageProtoAsPayloadFromWindowedValue()); - if (config.topic != null) { - StaticValueProvider topic = StaticValueProvider.of(config.topic); + String topicStr = config.topic; + if (topicStr != null) { + StaticValueProvider topic = StaticValueProvider.of(topicStr); writeBuilder.setTopicProvider(NestedValueProvider.of(topic, PubsubTopic::fromPath)); } if (config.idAttribute != null) { @@ -102,7 +102,8 @@ public static class ParsePubsubMessageProtoAsPayloadFromWindowedValue @Override public PubsubMessage apply(ValueInSingleWindow input) { - return INNER.apply(input.getValue()); + return INNER.apply( + checkArgumentNotNull(input.getValue(), "Windowed value input cannot be null")); } } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/NestedRowToMessage.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/NestedRowToMessage.java index 9c56410cda6b..50325f7b39e2 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/NestedRowToMessage.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/NestedRowToMessage.java @@ -22,6 +22,7 @@ import static org.apache.beam.sdk.io.gcp.pubsub.PubsubSchemaIOProvider.ATTRIBUTE_ARRAY_FIELD_TYPE; import static org.apache.beam.sdk.io.gcp.pubsub.PubsubSchemaIOProvider.ATTRIBUTE_MAP_FIELD_TYPE; import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import java.util.Collection; @@ -34,16 +35,17 @@ import org.apache.beam.sdk.transforms.SimpleFunction; import org.apache.beam.sdk.values.Row; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; class NestedRowToMessage extends SimpleFunction { private static final long serialVersionUID = 65176815766314684L; - private final PayloadSerializer serializer; + private final @Nullable PayloadSerializer serializer; private final SerializableFunction> attributesExtractor; private final SerializableFunction payloadExtractor; @SuppressWarnings("methodref.receiver.bound") - NestedRowToMessage(PayloadSerializer serializer, Schema schema) { + NestedRowToMessage(@Nullable PayloadSerializer serializer, Schema schema) { this.serializer = serializer; if (schema.getField(ATTRIBUTES_FIELD).getType().equals(ATTRIBUTE_MAP_FIELD_TYPE)) { attributesExtractor = NestedRowToMessage::getAttributesFromMap; @@ -81,7 +83,7 @@ private static byte[] getPayloadFromBytes(Row row) { } private byte[] getPayloadFromNested(Row row) { - return serializer.serialize(checkArgumentNotNull(row.getRow(PAYLOAD_FIELD))); + return checkStateNotNull(serializer).serialize(checkArgumentNotNull(row.getRow(PAYLOAD_FIELD))); } @Override diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubSubPayloadTranslation.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubSubPayloadTranslation.java index b77928bd957e..eec590a480f7 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubSubPayloadTranslation.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubSubPayloadTranslation.java @@ -19,6 +19,7 @@ import static org.apache.beam.sdk.io.gcp.pubsub.PubsubIO.ENABLE_CUSTOM_PUBSUB_SINK; import static org.apache.beam.sdk.io.gcp.pubsub.PubsubIO.ENABLE_CUSTOM_PUBSUB_SOURCE; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import com.google.auto.service.AutoService; import java.util.Collections; @@ -37,16 +38,13 @@ import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider; import org.apache.beam.sdk.runners.AppliedPTransform; import org.apache.beam.sdk.transforms.PTransform; -import org.apache.beam.sdk.util.Preconditions; import org.apache.beam.sdk.util.construction.PTransformTranslation; import org.apache.beam.sdk.util.construction.PTransformTranslation.TransformPayloadTranslator; import org.apache.beam.sdk.util.construction.SdkComponents; import org.apache.beam.sdk.util.construction.TransformPayloadTranslatorRegistrar; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) /** * Utility methods for translating a {@link Unbounded} which reads from {@link * PubsubUnboundedSource} to {@link RunnerApi} representations. @@ -61,7 +59,7 @@ public String getUrn() { } @Override - public RunnerApi.FunctionSpec translate( + public RunnerApi.@Nullable FunctionSpec translate( AppliedPTransform> transform, SdkComponents components) { if (ExperimentalOptions.hasExperiment( transform.getPipeline().getOptions(), ENABLE_CUSTOM_PUBSUB_SOURCE)) { @@ -93,11 +91,13 @@ public RunnerApi.FunctionSpec translate( } } - if (pubsubUnboundedSource.getTimestampAttribute() != null) { - payloadBuilder.setTimestampAttribute(pubsubUnboundedSource.getTimestampAttribute()); + String timestampAttribute = pubsubUnboundedSource.getTimestampAttribute(); + if (timestampAttribute != null) { + payloadBuilder.setTimestampAttribute(timestampAttribute); } - if (pubsubUnboundedSource.getIdAttribute() != null) { - payloadBuilder.setIdAttribute(pubsubUnboundedSource.getIdAttribute()); + String idAttribute = pubsubUnboundedSource.getIdAttribute(); + if (idAttribute != null) { + payloadBuilder.setIdAttribute(idAttribute); } payloadBuilder.setWithAttributes( pubsubUnboundedSource.getNeedsAttributes() || pubsubUnboundedSource.getNeedsMessageId()); @@ -116,7 +116,7 @@ public String getUrn() { } @Override - public RunnerApi.FunctionSpec translate( + public RunnerApi.@Nullable FunctionSpec translate( AppliedPTransform transform, SdkComponents components) { if (ExperimentalOptions.hasExperiment( @@ -125,19 +125,20 @@ public RunnerApi.FunctionSpec translate( } PubSubWritePayload.Builder payloadBuilder = PubSubWritePayload.newBuilder(); ValueProvider topicProvider = - Preconditions.checkStateNotNull(transform.getTransform().outer.getTopicProvider()); + checkStateNotNull(transform.getTransform().outer.getTopicProvider()); if (topicProvider.isAccessible()) { payloadBuilder.setTopic(topicProvider.get().getFullPath()); } else { payloadBuilder.setTopicRuntimeOverridden( ((NestedValueProvider) topicProvider).propertyName()); } - if (transform.getTransform().outer.getTimestampAttribute() != null) { - payloadBuilder.setTimestampAttribute( - transform.getTransform().outer.getTimestampAttribute()); + String timestampAttribute = transform.getTransform().outer.getTimestampAttribute(); + if (timestampAttribute != null) { + payloadBuilder.setTimestampAttribute(timestampAttribute); } - if (transform.getTransform().outer.getIdAttribute() != null) { - payloadBuilder.setIdAttribute(transform.getTransform().outer.getIdAttribute()); + String idAttribute = transform.getTransform().outer.getIdAttribute(); + if (idAttribute != null) { + payloadBuilder.setIdAttribute(idAttribute); } return FunctionSpec.newBuilder() .setUrn(getUrn(transform.getTransform())) @@ -154,7 +155,7 @@ public String getUrn() { } @Override - public RunnerApi.FunctionSpec translate( + public RunnerApi.@Nullable FunctionSpec translate( AppliedPTransform transform, SdkComponents components) { if (ExperimentalOptions.hasExperiment( @@ -162,12 +163,13 @@ public RunnerApi.FunctionSpec translate( return null; } PubSubWritePayload.Builder payloadBuilder = PubSubWritePayload.newBuilder(); - if (transform.getTransform().outer.getTimestampAttribute() != null) { - payloadBuilder.setTimestampAttribute( - transform.getTransform().outer.getTimestampAttribute()); + String timestampAttribute = transform.getTransform().outer.getTimestampAttribute(); + if (timestampAttribute != null) { + payloadBuilder.setTimestampAttribute(timestampAttribute); } - if (transform.getTransform().outer.getIdAttribute() != null) { - payloadBuilder.setIdAttribute(transform.getTransform().outer.getIdAttribute()); + String idAttribute = transform.getTransform().outer.getIdAttribute(); + if (idAttribute != null) { + payloadBuilder.setIdAttribute(idAttribute); } return FunctionSpec.newBuilder() .setUrn(getUrn(transform.getTransform())) 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 333497462177..3f4d5765b66c 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 @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; @@ -40,13 +42,11 @@ 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.ImmutableMap; import org.checkerframework.checker.nullness.qual.Nullable; +import org.checkerframework.dataflow.qual.Pure; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** An (abstract) helper class for talking to Pubsub via an underlying transport. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public abstract class PubsubClient implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(PubsubClient.class); private static final Map> @@ -82,7 +82,7 @@ PubsubClient newClient( * Return timestamp as ms-since-unix-epoch corresponding to {@code timestamp}. Throw {@link * IllegalArgumentException} if timestamp cannot be recognized. */ - protected static Long parseTimestampAsMsSinceEpoch(String timestamp) { + protected static long parseTimestampAsMsSinceEpoch(String timestamp) { if (timestamp.isEmpty()) { throw new IllegalArgumentException("Empty timestamp."); } @@ -113,17 +113,12 @@ protected static long extractTimestampAttribute( String timestampAttribute, @Nullable Map attributes) { Preconditions.checkState(!timestampAttribute.isEmpty()); String value = attributes == null ? null : attributes.get(timestampAttribute); - checkArgument( - value != null, - "PubSub message is missing a value for timestamp attribute %s", - timestampAttribute); - Long timestampMsSinceEpoch = parseTimestampAsMsSinceEpoch(value); - checkArgument( - timestampMsSinceEpoch != null, - "Cannot interpret value of attribute %s as timestamp: %s", - timestampAttribute, - value); - return timestampMsSinceEpoch; + String nonNullValue = + checkArgumentNotNull( + value, + "PubSub message is missing a value for timestamp attribute %s", + timestampAttribute); + return parseTimestampAsMsSinceEpoch(nonNullValue); } /** Path representing a cloud project id. */ @@ -386,17 +381,21 @@ public static TopicPath topicPathFromName(String projectId, String topicName) { public abstract static class OutgoingMessage implements Serializable { /** Underlying Message. May not have publish timestamp set. */ + @Pure public abstract PubsubMessage getMessage(); /** Timestamp for element (ms since epoch). */ + @Pure public abstract long getTimestampMsSinceEpoch(); /** * If using an id attribute, the record id to associate with this record's metadata so the * receiver can reject duplicates. Otherwise {@literal null}. */ + @Pure public abstract @Nullable String recordId(); + @Pure public abstract @Nullable String topic(); public static OutgoingMessage of( @@ -415,11 +414,13 @@ public static OutgoingMessage of( @Nullable String topic) { PubsubMessage.Builder builder = PubsubMessage.newBuilder().setData(ByteString.copyFrom(message.getPayload())); - if (message.getAttributeMap() != null) { - builder.putAllAttributes(message.getAttributeMap()); + Map attributeMap = message.getAttributeMap(); + if (attributeMap != null) { + builder.putAllAttributes(attributeMap); } - if (message.getOrderingKey() != null) { - builder.setOrderingKey(message.getOrderingKey()); + String orderingKey = message.getOrderingKey(); + if (orderingKey != null) { + builder.setOrderingKey(orderingKey); } return of(builder.build(), timestampMsSinceEpoch, recordId, topic); } @@ -435,21 +436,23 @@ public static OutgoingMessage of( public abstract static class IncomingMessage implements Serializable { /** Underlying Message. */ + @Pure public abstract PubsubMessage message(); - /** - * Timestamp for element (ms since epoch). Either Pubsub's processing time, or the custom - * timestamp associated with the message. - */ + /** Either Pubsub's processing time, or the custom timestamp associated with the message. */ + @Pure public abstract long timestampMsSinceEpoch(); /** Timestamp (in system time) at which we requested the message (ms since epoch). */ + @Pure public abstract long requestTimeMsSinceEpoch(); /** Id to pass back to Pubsub to acknowledge receipt of this message. */ + @Pure public abstract String ackId(); /** Id to pass to the runner to distinguish this message from all others. */ + @Pure public abstract String recordId(); public static IncomingMessage of( @@ -554,7 +557,7 @@ public abstract void createSchema( public abstract void deleteSchema(SchemaPath schemaPath) throws IOException; /** Return {@link SchemaPath} from {@link TopicPath} if exists. */ - public abstract SchemaPath getSchemaPath(TopicPath topicPath) throws IOException; + public abstract @Nullable SchemaPath getSchemaPath(TopicPath topicPath) throws IOException; /** Return a Beam {@link Schema} from the Pub/Sub schema resource, if exists. */ public abstract Schema getSchema(SchemaPath schemaPath) throws IOException; @@ -567,8 +570,10 @@ static Schema fromPubsubSchema(com.google.api.services.pubsub.model.Schema pubsu "Pub/Sub schema type %s is not supported at this time", pubsubSchema.getType())); } SerializableFunction definitionToSchemaFn = - schemaTypeToConversionFnMap.get(pubsubSchema.getType()); - return definitionToSchemaFn.apply(pubsubSchema.getDefinition()); + checkStateNotNull(schemaTypeToConversionFnMap.get(pubsubSchema.getType())); + String definition = + checkNotNull(pubsubSchema.getDefinition(), "Pub/Sub schema definition is null"); + return definitionToSchemaFn.apply(definition); } /** Convert a {@link com.google.pubsub.v1.Schema} to a Beam {@link Schema}. */ @@ -579,7 +584,7 @@ static Schema fromPubsubSchema(com.google.pubsub.v1.Schema pubsubSchema) { String.format("Pub/Sub schema type %s is not supported at this time", typeName)); } SerializableFunction definitionToSchemaFn = - schemaTypeToConversionFnMap.get(typeName); + checkStateNotNull(schemaTypeToConversionFnMap.get(typeName)); return definitionToSchemaFn.apply(pubsubSchema.getDefinition()); } 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 7fb30a1099aa..42fb80b6b42b 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 @@ -17,7 +17,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 static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import com.google.auth.Credentials; import com.google.protobuf.Timestamp; @@ -67,20 +67,17 @@ import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; 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.base.Strings; 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.util.concurrent.ThreadFactoryBuilder; import org.checkerframework.checker.nullness.qual.Nullable; /** A helper class for talking to Pubsub via grpc. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubGrpcClient extends PubsubClient { private static final int LIST_BATCH_SIZE = 1000; private static final int DEFAULT_TIMEOUT_S = 60; + @SuppressWarnings("nullness") // netty requires ciphers(null) but is not annotated private static ManagedChannel channelForRootUrl(String urlString) throws IOException { String format = PubsubOptions.targetForRootUrl(urlString); @@ -104,7 +101,7 @@ public PubsubClient newClient( @Nullable String timestampAttribute, @Nullable String idAttribute, PubsubOptions options, - String rootUrlOverride) + @Nullable String rootUrlOverride) throws IOException { return new PubsubGrpcClient( timestampAttribute, @@ -144,9 +141,9 @@ public String getKind() { /** Cached stubs, or null if not cached. */ private PublisherGrpc.@Nullable PublisherBlockingStub cachedPublisherStub; - private SubscriberGrpc.SubscriberBlockingStub cachedSubscriberStub; + private SubscriberGrpc.@Nullable SubscriberBlockingStub cachedSubscriberStub; - private SchemaServiceGrpc.SchemaServiceBlockingStub cachedSchemaServiceStub; + private SchemaServiceGrpc.@Nullable SchemaServiceBlockingStub cachedSchemaServiceStub; @VisibleForTesting PubsubGrpcClient( @@ -189,7 +186,8 @@ public void close() { /** Return channel with interceptor for returning credentials. */ private Channel newChannel() throws IOException { - checkState(publisherChannel != null, "PubsubGrpcClient has been closed"); + ManagedChannel channel = + checkStateNotNull(publisherChannel, "PubsubGrpcClient has been closed"); ClientAuthInterceptor interceptor = new ClientAuthInterceptor( credentials, @@ -198,7 +196,7 @@ private Channel newChannel() throws IOException { .setDaemon(true) .setNameFormat("PubsubGrpcClient-thread") .build())); - return ClientInterceptors.intercept(publisherChannel, interceptor); + return ClientInterceptors.intercept(channel, interceptor); } /** Return a stub for making a publish request with a timeout. */ @@ -237,8 +235,9 @@ public int publish(TopicPath topic, List outgoingMessages) thro timestampAttribute, String.valueOf(outgoingMessage.getTimestampMsSinceEpoch())); } - if (idAttribute != null && !Strings.isNullOrEmpty(outgoingMessage.recordId())) { - message.putAttributes(idAttribute, outgoingMessage.recordId()); + String recordId = outgoingMessage.recordId(); + if (idAttribute != null && recordId != null && !recordId.isEmpty()) { + message.putAttributes(idAttribute, recordId); } request.addMessages(message); @@ -267,12 +266,18 @@ public List pull( } List incomingMessages = new ArrayList<>(response.getReceivedMessagesCount()); for (ReceivedMessage message : response.getReceivedMessagesList()) { + if (!message.hasMessage()) { + continue; + } PubsubMessage pubsubMessage = message.getMessage(); Map attributes = pubsubMessage.getAttributes(); // Timestamp. long timestampMsSinceEpoch; - if (Strings.isNullOrEmpty(timestampAttribute)) { + if (timestampAttribute == null || timestampAttribute.isEmpty()) { + if (!pubsubMessage.hasPublishTime()) { + throw new IllegalStateException("Received message missing publishTime"); + } Timestamp timestampProto = pubsubMessage.getPublishTime(); timestampMsSinceEpoch = timestampProto.getSeconds() * 1000 + timestampProto.getNanos() / 1000L / 1000L; @@ -282,14 +287,16 @@ public List pull( // Ack id. String ackId = message.getAckId(); - checkState(!Strings.isNullOrEmpty(ackId)); + if (ackId.isEmpty()) { + throw new IllegalStateException("Received message missing ackId"); + } // Record id, if any. - @Nullable String recordId = null; + String recordId = null; if (idAttribute != null) { recordId = attributes.get(idAttribute); } - if (Strings.isNullOrEmpty(recordId)) { + if (recordId == null || recordId.isEmpty()) { // Fall back to the Pubsub provided message id. recordId = pubsubMessage.getMessageId(); } @@ -475,15 +482,15 @@ public void deleteSchema(SchemaPath schemaPath) throws IOException { /** Return {@link SchemaPath} from {@link TopicPath} if exists. */ @Override - public SchemaPath getSchemaPath(TopicPath topicPath) throws IOException { + public @Nullable SchemaPath getSchemaPath(TopicPath topicPath) throws IOException { GetTopicRequest request = GetTopicRequest.newBuilder().setTopic(topicPath.getPath()).build(); Topic topic = publisherStub().getTopic(request); - SchemaSettings schemaSettings = topic.getSchemaSettings(); - if (schemaSettings.getSchema().isEmpty()) { + if (!topic.hasSchemaSettings()) { return null; } + SchemaSettings schemaSettings = topic.getSchemaSettings(); String schemaPath = schemaSettings.getSchema(); - if (schemaPath.equals(SchemaPath.DELETED_SCHEMA_PATH)) { + if (schemaPath.isEmpty() || schemaPath.equals(SchemaPath.DELETED_SCHEMA_PATH)) { return null; } return PubsubClient.schemaPathFromPath(schemaPath); 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 57005745044b..ee4e283cc2f2 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 @@ -18,6 +18,8 @@ package org.apache.beam.sdk.io.gcp.pubsub; import static org.apache.beam.sdk.transforms.errorhandling.BadRecordRouter.BAD_RECORD_TAG; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.google.api.client.util.Clock; @@ -71,7 +73,6 @@ import org.apache.beam.sdk.transforms.errorhandling.ErrorHandler.DefaultErrorHandler; import org.apache.beam.sdk.transforms.windowing.AfterWatermark; import org.apache.beam.sdk.util.CoderUtils; -import org.apache.beam.sdk.util.Preconditions; import org.apache.beam.sdk.values.EncodableThrowable; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PBegin; @@ -112,20 +113,21 @@ *

Updates to the I/O connector code

* * For any significant updates to this I/O connector, please consider involving corresponding code - * reviewers mentioned - * here. + * reviewers mentioned here. * *

Example PubsubIO read usage

* *
{@code
- * // Read from a specific topic; a subscription will be created at pipeline start time.
+ * // Read from a specific topic; a subscription will be created at pipeline
+ * // start time.
  * PCollection messages = PubsubIO.readMessages().fromTopic(topic);
  *
  * // Read from a subscription.
  * PCollection messages = PubsubIO.readMessages().fromSubscription(subscription);
  *
- * // Read messages including attributes. All PubSub attributes will be included in the PubsubMessage.
+ * // Read messages including attributes. All PubSub attributes will be included
+ * // in the PubsubMessage.
  * PCollection messages = PubsubIO.readMessagesWithAttributes().fromTopic(topic);
  *
  * // Examples of reading different types from PubSub.
@@ -161,10 +163,11 @@
  * topic and writing using the {@link PubsubIO#writeMessagesDynamic()} method. For example:
  *
  * 
{@code
- * events.apply(MapElements.into(new TypeDescriptor() {})
- *                         .via(e -> new PubsubMessage(
- *                             e.toByteString(), Collections.emptyMap()).withTopic(e.getCountry())))
- * .apply(PubsubIO.writeMessagesDynamic());
+ * events.apply(MapElements.into(new TypeDescriptor() {
+ * })
+ *     .via(e -> new PubsubMessage(
+ *         e.toByteString(), Collections.emptyMap()).withTopic(e.getCountry())))
+ *     .apply(PubsubIO.writeMessagesDynamic());
  * }
* *

Custom timestamps

@@ -175,9 +178,6 @@ * to be used, that timestamp must be published in a PubSub attribute and specified using {@link * PubsubIO.Read#withTimestampAttribute}. See the Javadoc for that method for the timestamp format. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubIO { private static final Logger LOG = LoggerFactory.getLogger(PubsubIO.class); @@ -251,9 +251,9 @@ private static void validatePubsubName(String name) { /** Populate common {@link DisplayData} between Pubsub source and sink. */ private static void populateCommonDisplayData( DisplayData.Builder builder, - String timestampAttribute, - String idAttribute, - ValueProvider topic) { + @Nullable String timestampAttribute, + @Nullable String idAttribute, + @Nullable ValueProvider topic) { builder .addIfNotNull( DisplayData.item("timestampAttribute", timestampAttribute) @@ -324,6 +324,13 @@ public static PubsubSubscription fromPath(String path) { subscriptionName = match.group(2); } + projectName = + checkArgumentNotNull( + projectName, "Invalid path: project name must be non-null: %s", path); + subscriptionName = + checkArgumentNotNull( + subscriptionName, "Invalid path: subscription name must be non-null: %s", path); + validateProjectName(projectName); validatePubsubName(subscriptionName); return new PubsubSubscription(PubsubSubscription.Type.NORMAL, projectName, subscriptionName); @@ -399,7 +406,7 @@ public TopicPath apply(PubsubTopic from) { /** Class representing a Cloud Pub/Sub Topic. */ public static class PubsubTopic implements Serializable { @Override - public boolean equals(Object o) { + public boolean equals(@Nullable Object o) { if (this == o) { return true; } @@ -470,6 +477,12 @@ public static PubsubTopic fromPath(String path) { topicName = match.group(2); } + projectName = + checkArgumentNotNull( + projectName, "Invalid path: project name must be non-null: %s", path); + topicName = + checkArgumentNotNull(topicName, "Invalid path: topic name must be non-null: %s", path); + validateProjectName(projectName); validatePubsubName(topicName); return new PubsubTopic(PubsubTopic.Type.NORMAL, projectName, topicName); @@ -604,7 +617,8 @@ public static Read readStrings() { */ public static Read readProtos(Class messageClass) { // TODO: Stop using ProtoCoder and instead parse the payload directly. - // We should not be relying on the fact that ProtoCoder's wire format is identical to + // We should not be relying on the fact that ProtoCoder's wire format is + // identical to // the protobuf wire format, as the wire format is not part of a coder's API. ProtoCoder coder = ProtoCoder.of(messageClass); return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build(); @@ -662,7 +676,8 @@ public static Read readProtoDynamicMessages(Descriptor descripto */ public static Read readAvros(Class clazz) { // TODO: Stop using AvroCoder and instead parse the payload directly. - // We should not be relying on the fact that AvroCoder's wire format is identical to + // We should not be relying on the fact that AvroCoder's wire format is + // identical to // the Avro wire format, as the wire format is not part of a coder's API. AvroCoder coder = AvroCoder.of(clazz); return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build(); @@ -698,6 +713,10 @@ public static Read readMessagesWithAttributesWithCoderAndParseFn( public static Read readAvroGenericRecords(org.apache.avro.Schema avroSchema) { AvroCoder coder = AvroCoder.of(avroSchema); Schema schema = AvroUtils.getSchema(GenericRecord.class, avroSchema); + if (schema == null) { + throw new IllegalArgumentException( + "Could not infer Beam schema from Avro schema: " + avroSchema); + } return Read.newBuilder(parsePayloadUsingCoder(coder)) .setCoder( SchemaCoder.of( @@ -722,6 +741,9 @@ public static Read readAvrosWithBeamSchema(Class clazz) { AvroCoder coder = AvroCoder.of(clazz); org.apache.avro.Schema avroSchema = coder.getSchema(); Schema schema = AvroUtils.getSchema(clazz, avroSchema); + if (schema == null) { + throw new IllegalArgumentException("Could not infer Beam schema for class: " + clazz); + } return Read.newBuilder(parsePayloadUsingCoder(coder)) .setCoder( SchemaCoder.of( @@ -773,7 +795,8 @@ public static Write writeStrings() { * Google Cloud Pub/Sub stream. */ public static Write writeProtos(Class messageClass) { - // TODO: Like in readProtos(), stop using ProtoCoder and instead format the payload directly. + // TODO: Like in readProtos(), stop using ProtoCoder and instead format the + // payload directly. return Write.newBuilder(formatPayloadUsingCoder(ProtoCoder.of(messageClass))) .setDynamicDestinations(false) .build(); @@ -786,7 +809,8 @@ public static Write writeProtos(Class messageClass) { public static Write writeProtos( Class messageClass, SerializableFunction, Map> attributeFn) { - // TODO: Like in readProtos(), stop using ProtoCoder and instead format the payload directly. + // TODO: Like in readProtos(), stop using ProtoCoder and instead format the + // payload directly. return Write.newBuilder(formatPayloadUsingCoder(ProtoCoder.of(messageClass), attributeFn)) .setDynamicDestinations(false) .build(); @@ -797,7 +821,8 @@ public static Write writeProtos( * Google Cloud Pub/Sub stream. */ public static Write writeAvros(Class clazz) { - // TODO: Like in readAvros(), stop using AvroCoder and instead format the payload directly. + // TODO: Like in readAvros(), stop using AvroCoder and instead format the + // payload directly. return Write.newBuilder(formatPayloadUsingCoder(AvroCoder.of(clazz))) .setDynamicDestinations(false) .build(); @@ -810,7 +835,8 @@ public static Write writeAvros(Class clazz) { public static Write writeAvros( Class clazz, SerializableFunction, Map> attributeFn) { - // TODO: Like in readAvros(), stop using AvroCoder and instead format the payload directly. + // TODO: Like in readAvros(), stop using AvroCoder and instead format the + // payload directly. return Write.newBuilder(formatPayloadUsingCoder(AvroCoder.of(clazz), attributeFn)) .setDynamicDestinations(false) .build(); @@ -1134,16 +1160,18 @@ public PCollection expand(PBegin input) { "PubSubIO cannot be configured with both a dead letter topic and a bad record router"); } + ValueProvider topicProvider = getTopicProvider(); @Nullable ValueProvider topicPath = - getTopicProvider() == null + topicProvider == null ? null - : NestedValueProvider.of(getTopicProvider(), new TopicPathTranslator()); + : NestedValueProvider.of(topicProvider, new TopicPathTranslator()); + ValueProvider subscriptionProvider = getSubscriptionProvider(); @Nullable ValueProvider subscriptionPath = - getSubscriptionProvider() == null + subscriptionProvider == null ? null - : NestedValueProvider.of(getSubscriptionProvider(), new SubscriptionPathTranslator()); + : NestedValueProvider.of(subscriptionProvider, new SubscriptionPathTranslator()); PubsubUnboundedSource source = new PubsubUnboundedSource( getClock(), @@ -1177,7 +1205,7 @@ private PCollection expandReadContinued( new SerializableFunction() { // flag that reported metrics private final SerializableFunction underlying = - Objects.requireNonNull(getParseFn()); + checkStateNotNull(getParseFn()); private transient boolean reportedMetrics = false; // public @@ -1204,8 +1232,9 @@ public T apply(PubsubMessage input) { return underlying.apply(input); } }; + ValueProvider deadLetterTopicProvider = getDeadLetterTopicProvider(); PCollection read; - if (getDeadLetterTopicProvider() == null + if (deadLetterTopicProvider == null && (getBadRecordRouter() instanceof ThrowingBadRecordRouter)) { read = preParse.apply(MapElements.into(typeDescriptor).via(parseFnWrapped)); } else { @@ -1234,7 +1263,8 @@ public T apply(PubsubMessage input) { // Write out failures to the provided dead-letter topic. result .failures() - // Since the stack trace could easily exceed Pub/Sub limits, we need to remove it from + // Since the stack trace could easily exceed Pub/Sub limits, we need to remove + // it from // the attributes. .apply( "PubsubIO.Read/Map/Remove-Stack-Trace-Attribute", @@ -1254,7 +1284,9 @@ public T apply(PubsubMessage input) { ImmutableMap attributes = ImmutableMap.builder() .put("exceptionClassName", throwable.getClass().getName()) - .put("exceptionMessage", throwable.getMessage()) + .put( + "exceptionMessage", + MoreObjects.firstNonNull(throwable.getMessage(), "")) .put("pubsubMessageId", messageId) .build(); @@ -1266,7 +1298,7 @@ public T apply(PubsubMessage input) { .via(kv -> new PubsubMessage(kv.getKey().getPayload(), kv.getValue()))) .apply( writeMessages() - .to(getDeadLetterTopicProvider().get().asPath()) + .to(checkStateNotNull(deadLetterTopicProvider).get().asPath()) .withClientFactory(getPubsubClientFactory())); } } @@ -1274,16 +1306,20 @@ public T apply(PubsubMessage input) { } @Override - public void validate(PipelineOptions options) { + public void validate(@Nullable PipelineOptions options) { if (!getValidate()) { return; } + if (options == null) { + return; + } PubsubOptions psOptions = options.as(PubsubOptions.class); // Validate the existence of the topic. - if (getTopicProvider() != null) { - PubsubTopic topic = getTopicProvider().get(); + ValueProvider topicProvider = getTopicProvider(); + if (topicProvider != null) { + PubsubTopic topic = topicProvider.get(); boolean topicExists = true; try (PubsubClient pubsubClient = getPubsubClientFactory() @@ -1405,10 +1441,10 @@ static Builder newBuilder() { @AutoValue.Builder abstract static class Builder { - abstract Builder setTopicProvider(ValueProvider topicProvider); + abstract Builder setTopicProvider(@Nullable ValueProvider topicProvider); abstract Builder setTopicFunction( - SerializableFunction, PubsubTopic> topicFunction); + @Nullable SerializableFunction, PubsubTopic> topicFunction); abstract Builder setDynamicDestinations(boolean dynamicDestinations); @@ -1546,7 +1582,7 @@ public Write withTimestampAttribute(String timestampAttribute) { * attribute with the specified name. The value of the attribute is an opaque string. * *

If the output from this sink is being read by another Beam pipeline, then {@link - * PubsubIO.Read#withIdAttribute(String)} can be used to ensure that* the other source reads + * PubsubIO.Read#withIdAttribute(String)} can be used to ensure that the other source reads * these unique identifiers from the appropriate attribute. */ public Write withIdAttribute(String idAttribute) { @@ -1582,10 +1618,11 @@ public PDone expand(PCollection input) { + "dynamic topic destinations."); } + ValueProvider topicProvider = getTopicProvider(); SerializableFunction, PubsubIO.PubsubTopic> topicFunction = getTopicFunction(); - if (topicFunction == null && getTopicProvider() != null) { - topicFunction = v -> getTopicProvider().get(); + if (topicFunction == null && topicProvider != null) { + topicFunction = v -> topicProvider.get(); } int maxMessageSize = PUBSUB_MESSAGE_MAX_TOTAL_SIZE; if (input.isBounded() == PCollection.IsBounded.BOUNDED) { @@ -1633,8 +1670,8 @@ public PDone expand(PCollection input) { return pubsubMessages.apply( new PubsubUnboundedSink( getPubsubClientFactory(), - getTopicProvider() != null - ? NestedValueProvider.of(getTopicProvider(), new TopicPathTranslator()) + topicProvider != null + ? NestedValueProvider.of(topicProvider, new TopicPathTranslator()) : null, getTimestampAttribute(), getIdAttribute(), @@ -1650,16 +1687,20 @@ public PDone expand(PCollection input) { } @Override - public void validate(PipelineOptions options) { + public void validate(@Nullable PipelineOptions options) { if (!getValidate()) { return; } + if (options == null) { + return; + } PubsubOptions psOptions = options.as(PubsubOptions.class); // Validate the existence of the topic. - if (getTopicProvider() != null) { - PubsubTopic topic = getTopicProvider().get(); + ValueProvider topicProvider = getTopicProvider(); + if (topicProvider != null) { + PubsubTopic topic = topicProvider.get(); boolean topicExists = true; try (PubsubClient pubsubClient = getPubsubClientFactory() @@ -1702,10 +1743,11 @@ private class OutgoingData { } // NOTE: A single publish request may only write to one ordering key. - // See https://cloud.google.com/pubsub/docs/publisher#using-ordering-keys for details. - private transient Map, OutgoingData> output; + // See https://cloud.google.com/pubsub/docs/publisher#using-ordering-keys for + // details. + private transient @Nullable Map, OutgoingData> output; - private transient PubsubClient pubsubClient; + private transient @Nullable PubsubClient pubsubClient; private int maxPublishBatchByteSize; private int maxPublishBatchSize; @@ -1750,16 +1792,19 @@ public void processElement(@Element PubsubMessage message, @Timestamp Instant ti final int messageSize = msg.getMessage().getData().size(); final PubsubTopic pubsubTopic; - if (getTopicProvider() != null) { - pubsubTopic = getTopicProvider().get(); + ValueProvider topicProvider = getTopicProvider(); + if (topicProvider != null) { + pubsubTopic = topicProvider.get(); } else { - pubsubTopic = PubsubTopic.fromPath(Preconditions.checkArgumentNotNull(msg.topic())); + pubsubTopic = PubsubTopic.fromPath(checkArgumentNotNull(msg.topic())); } - // Checking before adding the message stops us from violating max batch size or bytes + // Checking before adding the message stops us from violating max batch size or + // bytes String orderingKey = getPublishWithOrderingKey() ? msg.getMessage().getOrderingKey() : ""; final OutgoingData currentTopicAndOrderingKeyOutput = - output.computeIfAbsent(KV.of(pubsubTopic, orderingKey), t -> new OutgoingData()); + checkStateNotNull(output) + .computeIfAbsent(KV.of(pubsubTopic, orderingKey), t -> new OutgoingData()); // TODO(sjvanrossum): https://github.com/apache/beam/issues/31800 if (currentTopicAndOrderingKeyOutput.messages.size() >= maxPublishBatchSize || (!currentTopicAndOrderingKeyOutput.messages.isEmpty() @@ -1776,18 +1821,24 @@ public void processElement(@Element PubsubMessage message, @Timestamp Instant ti @FinishBundle public void finishBundle() throws IOException { - for (Map.Entry, OutgoingData> entry : output.entrySet()) { - publish(entry.getKey().getKey(), entry.getValue().messages); + final Map, OutgoingData> output = this.output; + if (output != null) { + for (Map.Entry, OutgoingData> entry : output.entrySet()) { + publish(entry.getKey().getKey(), entry.getValue().messages); + } + } + this.output = null; + final PubsubClient pubsubClient = this.pubsubClient; + if (pubsubClient != null) { + pubsubClient.close(); } - output = null; - pubsubClient.close(); - pubsubClient = null; + this.pubsubClient = null; } private void publish(PubsubTopic topic, List messages) throws IOException { int n = - pubsubClient.publish( - PubsubClient.topicPathFromName(topic.project, topic.topic), messages); + checkStateNotNull(pubsubClient) + .publish(PubsubClient.topicPathFromName(topic.project, topic.topic), messages); checkState(n == messages.size()); } 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 f36e94360996..d0f9635391f9 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 @@ -17,7 +17,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 static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import com.google.api.client.googleapis.json.GoogleJsonResponseException; import com.google.api.client.http.HttpRequestInitializer; @@ -51,14 +51,10 @@ import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; 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.base.Strings; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.checkerframework.checker.nullness.qual.Nullable; /** A Pubsub client using JSON transport. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubJsonClient extends PubsubClient { private static class PubsubJsonClientFactory implements PubsubClientFactory { @@ -85,7 +81,7 @@ public PubsubClient newClient( @Nullable String timestampAttribute, @Nullable String idAttribute, PubsubOptions options, - String rootUrlOverride) + @Nullable String rootUrlOverride) throws IOException { Pubsub pubsub = new Pubsub.Builder( @@ -163,8 +159,9 @@ private Map getMessageAttributes(OutgoingMessage outgoingMessage attributes.put( timestampAttribute, String.valueOf(outgoingMessage.getTimestampMsSinceEpoch())); } - if (idAttribute != null && !Strings.isNullOrEmpty(outgoingMessage.recordId())) { - attributes.put(idAttribute, outgoingMessage.recordId()); + String recordId = outgoingMessage.recordId(); + if (idAttribute != null && recordId != null && !recordId.isEmpty()) { + attributes.put(idAttribute, recordId); } return attributes; } @@ -181,45 +178,51 @@ public List pull( new PullRequest().setReturnImmediately(returnImmediately).setMaxMessages(batchSize); PullResponse response = pubsub.projects().subscriptions().pull(subscription.getPath(), request).execute(); - if (response.getReceivedMessages() == null || response.getReceivedMessages().isEmpty()) { + List receivedMessages = response.getReceivedMessages(); + if (receivedMessages == null || receivedMessages.isEmpty()) { return ImmutableList.of(); } - List incomingMessages = new ArrayList<>(response.getReceivedMessages().size()); - for (ReceivedMessage message : response.getReceivedMessages()) { + List incomingMessages = new ArrayList<>(receivedMessages.size()); + for (ReceivedMessage message : receivedMessages) { PubsubMessage pubsubMessage = message.getMessage(); - Map attributes; - if (pubsubMessage.getAttributes() != null) { - attributes = pubsubMessage.getAttributes(); - } else { + if (pubsubMessage == null) { + continue; + } + Map attributes = pubsubMessage.getAttributes(); + if (attributes == null) { attributes = new HashMap<>(); } // Payload. - byte[] elementBytes = pubsubMessage.getData() == null ? null : pubsubMessage.decodeData(); - if (elementBytes == null) { - elementBytes = new byte[0]; - } + String dataStr = pubsubMessage.getData(); + byte[] elementBytes = (dataStr == null) ? new byte[0] : pubsubMessage.decodeData(); // Timestamp. long timestampMsSinceEpoch; - if (Strings.isNullOrEmpty(timestampAttribute)) { - timestampMsSinceEpoch = parseTimestampAsMsSinceEpoch(message.getMessage().getPublishTime()); + if (timestampAttribute == null || timestampAttribute.isEmpty()) { + String publishTime = pubsubMessage.getPublishTime(); + if (publishTime == null) { + throw new IllegalStateException("Received message missing publishTime"); + } + timestampMsSinceEpoch = parseTimestampAsMsSinceEpoch(publishTime); } else { timestampMsSinceEpoch = extractTimestampAttribute(timestampAttribute, attributes); } // Ack id. String ackId = message.getAckId(); - checkState(!Strings.isNullOrEmpty(ackId)); + if (ackId == null || ackId.isEmpty()) { + throw new IllegalStateException("Received message missing ackId"); + } // Record id, if any. @Nullable String recordId = null; if (idAttribute != null) { recordId = attributes.get(idAttribute); } - if (Strings.isNullOrEmpty(recordId)) { + if (recordId == null || recordId.isEmpty()) { // Fall back to the Pubsub provided message id. - recordId = pubsubMessage.getMessageId(); + recordId = checkStateNotNull(pubsubMessage.getMessageId(), "Message ID is missing"); } com.google.pubsub.v1.PubsubMessage.Builder protoMessage = @@ -229,8 +232,9 @@ public List pull( // {@link PubsubMessage} uses `null` or empty string to represent no ordering key. // {@link com.google.pubsub.v1.PubsubMessage} does not track string field presence and uses // empty string as a default. - if (pubsubMessage.getOrderingKey() != null) { - protoMessage.setOrderingKey(pubsubMessage.getOrderingKey()); + String orderingKey = pubsubMessage.getOrderingKey(); + if (orderingKey != null) { + protoMessage.setOrderingKey(orderingKey); } incomingMessages.add( IncomingMessage.of( @@ -289,19 +293,25 @@ public void deleteTopic(TopicPath topic) throws IOException { public List listTopics(ProjectPath project) throws IOException { Topics.List request = pubsub.projects().topics().list(project.getPath()); ListTopicsResponse response = request.execute(); - if (response.getTopics() == null || response.getTopics().isEmpty()) { + List topicsList = response.getTopics(); + if (topicsList == null || topicsList.isEmpty()) { return ImmutableList.of(); } - List topics = new ArrayList<>(response.getTopics().size()); + List topics = new ArrayList<>(topicsList.size()); while (true) { - for (Topic topic : response.getTopics()) { + for (Topic topic : topicsList) { topics.add(topicPathFromPath(topic.getName())); } - if (Strings.isNullOrEmpty(response.getNextPageToken())) { + String nextPageToken = response.getNextPageToken(); + if (nextPageToken == null || nextPageToken.isEmpty()) { break; } - request.setPageToken(response.getNextPageToken()); + request.setPageToken(nextPageToken); response = request.execute(); + topicsList = response.getTopics(); + if (topicsList == null) { + break; + } } return topics; } @@ -345,21 +355,28 @@ public List listSubscriptions(ProjectPath project, TopicPath t throws IOException { Subscriptions.List request = pubsub.projects().subscriptions().list(project.getPath()); ListSubscriptionsResponse response = request.execute(); - if (response.getSubscriptions() == null || response.getSubscriptions().isEmpty()) { + List subscriptionsList = response.getSubscriptions(); + if (subscriptionsList == null || subscriptionsList.isEmpty()) { return ImmutableList.of(); } - List subscriptions = new ArrayList<>(response.getSubscriptions().size()); + List subscriptions = new ArrayList<>(subscriptionsList.size()); while (true) { - for (Subscription subscription : response.getSubscriptions()) { - if (subscription.getTopic().equals(topic.getPath())) { + for (Subscription subscription : subscriptionsList) { + String subTopic = subscription.getTopic(); + if (subTopic != null && subTopic.equals(topic.getPath())) { subscriptions.add(subscriptionPathFromPath(subscription.getName())); } } - if (Strings.isNullOrEmpty(response.getNextPageToken())) { + String nextPageToken = response.getNextPageToken(); + if (nextPageToken == null || nextPageToken.isEmpty()) { break; } - request.setPageToken(response.getNextPageToken()); + request.setPageToken(nextPageToken); response = request.execute(); + subscriptionsList = response.getSubscriptions(); + if (subscriptionsList == null) { + break; + } } return subscriptions; } @@ -391,13 +408,14 @@ public void deleteSchema(SchemaPath schemaPath) throws IOException { /** Return {@link SchemaPath} from {@link TopicPath} if exists. */ @Override - public SchemaPath getSchemaPath(TopicPath topicPath) throws IOException { + public @Nullable SchemaPath getSchemaPath(TopicPath topicPath) throws IOException { Topic topic = pubsub.projects().topics().get(topicPath.getPath()).execute(); - if (topic.getSchemaSettings() == null) { + com.google.api.services.pubsub.model.SchemaSettings schemaSettings = topic.getSchemaSettings(); + if (schemaSettings == null) { return null; } - String schemaPath = topic.getSchemaSettings().getSchema(); - if (schemaPath.equals(SchemaPath.DELETED_SCHEMA_PATH)) { + String schemaPath = schemaSettings.getSchema(); + if (schemaPath == null || schemaPath.equals(SchemaPath.DELETED_SCHEMA_PATH)) { return null; } return PubsubClient.schemaPathFromPath(schemaPath); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessage.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessage.java index 779a9b847874..7732839181e6 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessage.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessage.java @@ -17,31 +17,32 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; -import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull; - import com.google.auto.value.AutoValue; import java.util.Map; import org.checkerframework.checker.nullness.qual.Nullable; +import org.checkerframework.dataflow.qual.Pure; /** * Class representing a Pub/Sub message. Each message contains a single message payload, a map of * attached attributes, a message id and an ordering key. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubMessage { @AutoValue abstract static class Impl { + @Pure abstract @Nullable String getTopic(); + @Pure @SuppressWarnings("mutable") abstract byte[] getPayload(); + @Pure abstract @Nullable Map getAttributeMap(); + @Pure abstract @Nullable String getMessageId(); + @Pure abstract @Nullable String getOrderingKey(); static Impl create( @@ -87,27 +88,32 @@ public PubsubMessage withTopic(String topic) { impl.getOrderingKey())); } + @Pure public @Nullable String getTopic() { return impl.getTopic(); } /** Returns the main PubSub message. */ + @Pure public byte[] getPayload() { return impl.getPayload(); } /** Returns the given attribute value. If not such attribute exists, returns null. */ + @Pure public @Nullable String getAttribute(String attribute) { - checkNotNull(attribute, "attribute"); - return impl.getAttributeMap().get(attribute); + Map attributeMap = impl.getAttributeMap(); + return attributeMap == null ? null : attributeMap.get(attribute); } /** Returns the full map of attributes. This is an unmodifiable map. */ + @Pure public @Nullable Map getAttributeMap() { return impl.getAttributeMap(); } /** Returns the messageId of the message populated by Cloud Pub/Sub. */ + @Pure public @Nullable String getMessageId() { return impl.getMessageId(); } @@ -123,6 +129,7 @@ public PubsubMessage withOrderingKey(@Nullable String orderingKey) { } /** Returns the ordering key of the message. */ + @Pure public @Nullable String getOrderingKey() { return impl.getOrderingKey(); } @@ -133,7 +140,7 @@ public String toString() { } @Override - public boolean equals(Object other) { + public boolean equals(@Nullable Object other) { if (!(other instanceof PubsubMessage)) { return false; } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageToRow.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageToRow.java index d860bfba95ec..c005253110c7 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageToRow.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageToRow.java @@ -19,6 +19,8 @@ import static java.util.stream.Collectors.toList; import static org.apache.beam.sdk.io.gcp.pubsub.PubsubSchemaIOProvider.ATTRIBUTE_ARRAY_ENTRY_SCHEMA; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import com.google.auto.value.AutoValue; @@ -26,7 +28,6 @@ import java.util.List; import java.util.Map; import java.util.stream.Collectors; -import javax.annotation.Nullable; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.Schema.TypeName; @@ -41,14 +42,12 @@ import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.Instant; /** Read side converter for {@link PubsubMessage} with JSON/AVRO payload. */ @Internal @AutoValue -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) abstract class PubsubMessageToRow extends PTransform, PCollectionTuple> implements Serializable { interface SerializerProvider extends SerializableFunction {} @@ -94,7 +93,10 @@ public PCollectionTuple expand(PCollection input) { ParDo.of( useFlatSchema() ? new FlatSchemaPubsubMessageToRow( - messageSchema(), useDlq(), serializerProvider()) + messageSchema(), + useDlq(), + checkArgumentNotNull( + serializerProvider(), "Flat schema requires serializerProvider")) : new NestedSchemaPubsubMessageToRow( messageSchema(), useDlq(), serializerProvider())) .withOutputTags( @@ -133,7 +135,8 @@ protected FlatSchemaPubsubMessageToRow( * Get the value for a field from a given payload in the order they're specified in the flat * schema. */ - private Object getValueForFieldFlatSchema(Schema.Field field, Instant timestamp, Row payload) { + private @Nullable Object getValueForFieldFlatSchema( + Schema.Field field, Instant timestamp, Row payload) { String fieldName = field.getName(); if (TIMESTAMP_FIELD.equals(fieldName)) { return timestamp; @@ -147,9 +150,10 @@ public void processElement( @Element PubsubMessage element, @Timestamp Instant timestamp, MultiOutputReceiver o) { try { Row payload = payloadSerializer.deserialize(element.getPayload()); - List values = + List<@Nullable Object> values = messageSchema.getFields().stream() - .map(field -> getValueForFieldFlatSchema(field, timestamp, payload)) + .<@Nullable Object>map( + field -> getValueForFieldFlatSchema(field, timestamp, payload)) .collect(toList()); o.get(MAIN_TAG).output(Row.withSchema(messageSchema).addValues(values).build()); } catch (Exception e) { @@ -186,7 +190,8 @@ private NestedSchemaPubsubMessageToRow( } else { checkArgument( messageSchema.getField(PAYLOAD_FIELD).getType().getTypeName().equals(TypeName.ROW)); - Schema payloadSchema = messageSchema.getField(PAYLOAD_FIELD).getType().getRowSchema(); + Schema payloadSchema = + checkStateNotNull(messageSchema.getField(PAYLOAD_FIELD).getType().getRowSchema()); this.payloadSerializer = serializerProvider.apply(payloadSchema); } } @@ -198,7 +203,13 @@ private Object maybeDeserialize(byte[] payload) { return payloadSerializer.deserialize(payload); } - private Object handleAttributes(Map attributeMap) { + private @Nullable Object handleAttributes(@Nullable Map attributeMap) { + if (attributeMap == null) { + if (messageSchema.getField(ATTRIBUTES_FIELD).getType().getTypeName().isMapType()) { + return null; + } + return ImmutableList.of(); + } if (messageSchema.getField(ATTRIBUTES_FIELD).getType().getTypeName().isMapType()) { return attributeMap; } @@ -209,8 +220,11 @@ private Object handleAttributes(Map attributeMap) { } /** Get the value for a field int the order they're specified in the nested schema. */ - private Object getValueForFieldNestedSchema( - Schema.Field field, Instant timestamp, Map attributeMap, byte[] payload) { + private @Nullable Object getValueForFieldNestedSchema( + Schema.Field field, + Instant timestamp, + @Nullable Map attributeMap, + byte[] payload) { switch (field.getName()) { case TIMESTAMP_FIELD: return timestamp; @@ -232,9 +246,9 @@ private Object getValueForFieldNestedSchema( public void processElement( @Element PubsubMessage element, @Timestamp Instant timestamp, MultiOutputReceiver o) { try { - List values = + List<@Nullable Object> values = messageSchema.getFields().stream() - .map( + .<@Nullable Object>map( field -> getValueForFieldNestedSchema( field, timestamp, element.getAttributeMap(), element.getPayload())) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder.java index 968d76584bc3..d2ba0b9446f2 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; + import com.google.protobuf.Timestamp; import java.io.IOException; import java.io.InputStream; @@ -30,6 +32,7 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.extensions.protobuf.ProtoCoder; import org.apache.beam.sdk.values.TypeDescriptor; +import org.checkerframework.checker.nullness.qual.Nullable; /** * A coder for PubsubMessage including all fields of a PubSub message from server. @@ -37,22 +40,20 @@ *

Maintainers should prefer {@link PubsubMessageSchemaCoder} over this coder when adding * features to {@link PubsubIO}. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder extends CustomCoder { // A message's payload cannot be null private static final Coder PAYLOAD_CODER = ByteArrayCoder.of(); // A message's attributes can be null. - private static final Coder> ATTRIBUTES_CODER = + private static final Coder<@Nullable Map> ATTRIBUTES_CODER = NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())); // A message's messageId cannot be null private static final Coder MESSAGE_ID_CODER = StringUtf8Coder.of(); // A message's publish time, populated by server private static final Coder PUBLISH_TIME_CODER = ProtoCoder.of(Timestamp.class); // A message's ordering key can be null - private static final Coder ORDERING_KEY_CODER = NullableCoder.of(StringUtf8Coder.of()); + private static final Coder<@Nullable String> ORDERING_KEY_CODER = + NullableCoder.of(StringUtf8Coder.of()); public static Coder of(TypeDescriptor ignored) { return of(); @@ -66,7 +67,9 @@ public static PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder of() { public void encode(PubsubMessage value, OutputStream outStream) throws IOException { PAYLOAD_CODER.encode(value.getPayload(), outStream); ATTRIBUTES_CODER.encode(value.getAttributeMap(), outStream); - MESSAGE_ID_CODER.encode(value.getMessageId(), outStream); + MESSAGE_ID_CODER.encode( + checkArgumentNotNull(value.getMessageId(), "Cannot encode PubsubMessage without messageId"), + outStream); // TODO(discuss what to do with publish_time field) PUBLISH_TIME_CODER.encode(Timestamp.getDefaultInstance(), outStream); ORDERING_KEY_CODER.encode(value.getOrderingKey(), outStream); @@ -75,10 +78,10 @@ public void encode(PubsubMessage value, OutputStream outStream) throws IOExcepti @Override public PubsubMessage decode(InputStream inStream) throws IOException { byte[] payload = PAYLOAD_CODER.decode(inStream); - Map attributes = ATTRIBUTES_CODER.decode(inStream); + @Nullable Map attributes = ATTRIBUTES_CODER.decode(inStream); String messageId = MESSAGE_ID_CODER.decode(inStream); PUBLISH_TIME_CODER.decode(inStream); - String orderingKey = ORDERING_KEY_CODER.decode(inStream); + @Nullable String orderingKey = ORDERING_KEY_CODER.decode(inStream); return new PubsubMessage(payload, attributes, messageId, orderingKey); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdCoder.java index 803ff83583ce..32449d0425e8 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdCoder.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdCoder.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; + import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -28,6 +30,7 @@ import org.apache.beam.sdk.coders.NullableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.values.TypeDescriptor; +import org.checkerframework.checker.nullness.qual.Nullable; /** * A coder for PubsubMessage including attributes and the message id from the PubSub server. @@ -35,14 +38,11 @@ *

Maintainers should prefer {@link PubsubMessageSchemaCoder} over this coder when adding * features to {@link PubsubIO}. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubMessageWithAttributesAndMessageIdCoder extends CustomCoder { // A message's payload cannot be null private static final Coder PAYLOAD_CODER = ByteArrayCoder.of(); // A message's attributes can be null. - private static final Coder> ATTRIBUTES_CODER = + private static final Coder<@Nullable Map> ATTRIBUTES_CODER = NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())); // A message's messageId cannot be null private static final Coder MESSAGE_ID_CODER = StringUtf8Coder.of(); @@ -59,13 +59,15 @@ public static PubsubMessageWithAttributesAndMessageIdCoder of() { public void encode(PubsubMessage value, OutputStream outStream) throws IOException { PAYLOAD_CODER.encode(value.getPayload(), outStream); ATTRIBUTES_CODER.encode(value.getAttributeMap(), outStream); - MESSAGE_ID_CODER.encode(value.getMessageId(), outStream); + MESSAGE_ID_CODER.encode( + checkArgumentNotNull(value.getMessageId(), "Cannot encode PubsubMessage without messageId"), + outStream); } @Override public PubsubMessage decode(InputStream inStream) throws IOException { byte[] payload = PAYLOAD_CODER.decode(inStream); - Map attributes = ATTRIBUTES_CODER.decode(inStream); + @Nullable Map attributes = ATTRIBUTES_CODER.decode(inStream); String messageId = MESSAGE_ID_CODER.decode(inStream); return new PubsubMessage(payload, attributes, messageId); } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesCoder.java index 9dd3ab56a72a..38b7661fba9b 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesCoder.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesCoder.java @@ -28,6 +28,7 @@ import org.apache.beam.sdk.coders.NullableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.values.TypeDescriptor; +import org.checkerframework.checker.nullness.qual.Nullable; /** * A coder for PubsubMessage including attributes. @@ -35,14 +36,11 @@ *

Maintainers should prefer {@link PubsubMessageSchemaCoder} over this coder when adding * features to {@link PubsubIO}. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubMessageWithAttributesCoder extends CustomCoder { // A message's payload can not be null private static final Coder PAYLOAD_CODER = ByteArrayCoder.of(); // A message's attributes can be null. - private static final Coder> ATTRIBUTES_CODER = + private static final Coder<@Nullable Map> ATTRIBUTES_CODER = NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of())); public static Coder of(TypeDescriptor ignored) { @@ -58,6 +56,7 @@ public void encode(PubsubMessage value, OutputStream outStream) throws IOExcepti encode(value, outStream, Context.NESTED); } + @Override public void encode(PubsubMessage value, OutputStream outStream, Context context) throws IOException { PAYLOAD_CODER.encode(value.getPayload(), outStream); @@ -72,7 +71,7 @@ public PubsubMessage decode(InputStream inStream) throws IOException { @Override public PubsubMessage decode(InputStream inStream, Context context) throws IOException { byte[] payload = PAYLOAD_CODER.decode(inStream); - Map attributes = ATTRIBUTES_CODER.decode(inStream, context); + @Nullable Map attributes = ATTRIBUTES_CODER.decode(inStream, context); return new PubsubMessage(payload, attributes); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithMessageIdCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithMessageIdCoder.java index 7016146b3671..d150d8ac76cb 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithMessageIdCoder.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithMessageIdCoder.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; + import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -33,9 +35,6 @@ *

Maintainers should prefer {@link PubsubMessageSchemaCoder} over this coder when adding * features to {@link PubsubIO}. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubMessageWithMessageIdCoder extends CustomCoder { private static final Coder PAYLOAD_CODER = ByteArrayCoder.of(); // A message's messageId cannot be null @@ -48,7 +47,9 @@ public static PubsubMessageWithMessageIdCoder of() { @Override public void encode(PubsubMessage value, OutputStream outStream) throws IOException { PAYLOAD_CODER.encode(value.getPayload(), outStream); - MESSAGE_ID_CODER.encode(value.getMessageId(), outStream); + MESSAGE_ID_CODER.encode( + checkArgumentNotNull(value.getMessageId(), "Cannot encode PubsubMessage without messageId"), + outStream); } @Override diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java index c690115045bd..3efe18a08f32 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; + import com.google.api.client.util.Clock; import com.google.auto.service.AutoService; import java.io.Serializable; @@ -247,17 +249,23 @@ void setClock(@Nullable Clock clock) { this.clock = clock; } - @SuppressWarnings("nullness") PubsubIO.Read buildPubsubRead() { PubsubIO.Read pubsubRead = (configuration.getAttributes() == null && configuration.getAttributesMap() == null) ? PubsubIO.readMessages() : PubsubIO.readMessagesWithAttributes(); - if (!Strings.isNullOrEmpty(configuration.getTopic())) { - pubsubRead = pubsubRead.fromTopic(configuration.getTopic()); + String topic = configuration.getTopic(); + if (topic != null && !topic.isEmpty()) { + pubsubRead = pubsubRead.fromTopic(topic); } else { - pubsubRead = pubsubRead.fromSubscription(configuration.getSubscription()); + String subscription = configuration.getSubscription(); + pubsubRead = + pubsubRead.fromSubscription( + checkArgumentNotNull( + subscription, "Either topic or subscription must be specified")); } + final PubsubTestClientFactory clientFactory = this.clientFactory; + final Clock clock = this.clock; if (clientFactory != null && clock != null) { pubsubRead = pubsubRead.withClientFactory(clientFactory); pubsubRead = clientFactory.setClock(pubsubRead, clock); @@ -265,11 +273,13 @@ PubsubIO.Read buildPubsubRead() { throw new IllegalArgumentException( "Both PubsubTestClientFactory and Clock need to be specified for testing, but only one is provided"); } - if (!Strings.isNullOrEmpty(configuration.getIdAttribute())) { - pubsubRead = pubsubRead.withIdAttribute(configuration.getIdAttribute()); + String idAttribute = configuration.getIdAttribute(); + if (idAttribute != null && !idAttribute.isEmpty()) { + pubsubRead = pubsubRead.withIdAttribute(idAttribute); } - if (!Strings.isNullOrEmpty(configuration.getTimestampAttribute())) { - pubsubRead = pubsubRead.withTimestampAttribute(configuration.getTimestampAttribute()); + String timestampAttribute = configuration.getTimestampAttribute(); + if (timestampAttribute != null && !timestampAttribute.isEmpty()) { + pubsubRead = pubsubRead.withTimestampAttribute(timestampAttribute); } return pubsubRead; } @@ -277,11 +287,9 @@ PubsubIO.Read buildPubsubRead() { @Override public PCollectionRowTuple expand(PCollectionRowTuple input) { PubsubIO.Read pubsubRead = buildPubsubRead(); - @SuppressWarnings("nullness") - String errorOutput = - configuration.getErrorHandling() == null - ? null - : configuration.getErrorHandling().getOutput(); + PubsubReadSchemaTransformConfiguration.ErrorHandling errorHandling = + configuration.getErrorHandling(); + String errorOutput = errorHandling == null ? null : errorHandling.getOutput(); PCollectionTuple outputTuple = input diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubRowToMessage.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubRowToMessage.java index da3ee89de0d8..0085662fd3b6 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubRowToMessage.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubRowToMessage.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import com.google.auto.value.AutoValue; @@ -24,10 +26,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Optional; import java.util.stream.Collectors; -import javax.annotation.Nullable; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.Schema.Field; @@ -43,6 +43,7 @@ import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; +import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.DateTime; import org.joda.time.Instant; import org.joda.time.ReadableDateTime; @@ -51,9 +52,6 @@ /** Write side {@link Row} to {@link PubsubMessage} converter. */ @Internal @AutoValue -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) abstract class PubsubRowToMessage extends PTransform, PCollectionTuple> { static Builder builder() { @@ -88,18 +86,15 @@ static Builder builder() { * The {@link PayloadSerializer} to {@link PayloadSerializer#serialize(Row)} the payload or user * fields row. */ - @Nullable - abstract PayloadSerializer getPayloadSerializer(); + abstract @Nullable PayloadSerializer getPayloadSerializer(); /** The name of the attribute to apply to the {@link PubsubMessage}. */ - @Nullable - abstract String getTargetTimestampAttributeName(); + abstract @Nullable String getTargetTimestampAttributeName(); /** * Use for testing, simplify assertions of generated timestamp when input lacks a timestamp field. */ - @Nullable - abstract Instant getMockInstant(); + abstract @Nullable Instant getMockInstant(); /** Generates {@link Schema} of the {@link #ERROR} {@link PCollection}. */ static Schema errorSchema(Schema inputSchema) { @@ -345,9 +340,9 @@ static FieldMatcher of(String name, FieldType fieldType) { private final String name; - @Nullable private final TypeName typeName; + private final @Nullable TypeName typeName; - @Nullable private final FieldType fieldType; + private final @Nullable FieldType fieldType; private FieldMatcher(String name, @Nullable TypeName typeName, @Nullable FieldType fieldType) { this.name = name; @@ -379,7 +374,10 @@ boolean match(Schema schema) { if (typeName != null) { return field.getType().getTypeName().equals(typeName); } - return fieldType.equals(field.getType()); + if (fieldType != null) { + return fieldType.equals(field.getType()); + } + return true; } } @@ -404,10 +402,10 @@ static class PubsubRowToMessageDoFn extends DoFn { private final String payloadKeyName; private final Schema errorSchema; - @Nullable private final String targetTimestampKeyName; - @Nullable private final PayloadSerializer payloadSerializer; + private final @Nullable String targetTimestampKeyName; + private final @Nullable PayloadSerializer payloadSerializer; - @Nullable private final Instant mockInstant; + private final @Nullable Instant mockInstant; PubsubRowToMessageDoFn( String attributesKeyName, @@ -449,7 +447,7 @@ public void process(@Element Row row, MultiOutputReceiver receiver) { Row error = Row.withSchema(errorSchema) .withFieldValue(ERROR_DATA_FIELD_NAME, row) - .withFieldValue(ERROR_MESSAGE_FIELD.getName(), message) + .withFieldValue(ERROR_MESSAGE_FIELD.getName(), message != null ? message : "") .withFieldValue(ERROR_STACK_TRACE_FIELD.getName(), stackTrace) .build(); @@ -465,7 +463,8 @@ Map attributesWithoutTimestamp(Row row) { if (!row.getSchema().hasField(attributesKeyName)) { return new HashMap<>(); } - return row.getMap(attributesKeyName); + Map map = row.getMap(attributesKeyName); + return map == null ? new HashMap<>() : map; } /** @@ -482,7 +481,10 @@ String timestampAsString(Row row) { */ ReadableDateTime timestamp(Row row) { if (row.getSchema().hasField(sourceTimestampKeyName)) { - return row.getDateTime(sourceTimestampKeyName); + ReadableDateTime dt = row.getDateTime(sourceTimestampKeyName); + if (dt != null) { + return dt; + } } Instant instant = Instant.now(); if (mockInstant != null) { @@ -503,9 +505,10 @@ ReadableDateTime timestamp(Row row) { byte[] payload(Row row) { if (SchemaReflection.of(row.getSchema()) .matchesAll(FieldMatcher.of(payloadKeyName, PAYLOAD_BYTES_TYPE_NAME))) { - return row.getBytes(payloadKeyName); + return checkArgumentNotNull( + row.getBytes(payloadKeyName), "Payload field '%s' cannot be null", payloadKeyName); } - return Objects.requireNonNull(payloadSerializer).serialize(serializableRow(row)); + return checkStateNotNull(payloadSerializer).serialize(serializableRow(row)); } /** @@ -527,7 +530,8 @@ Row serializableRow(Row row) { } if (schemaReflection.matchesAll(FieldMatcher.of(payloadKeyName, PAYLOAD_ROW_TYPE_NAME))) { - return row.getRow(payloadKeyName); + return checkArgumentNotNull( + row.getRow(payloadKeyName), "Payload field '%s' cannot be null", payloadKeyName); } Schema withUserFieldsOnly = removeFields(row.getSchema(), attributesKeyName, sourceTimestampKeyName); @@ -553,8 +557,7 @@ static Builder builder() { abstract Field getTimestampField(); - @Nullable - abstract Field getPayloadField(); + abstract @Nullable Field getPayloadField(); /** * Builds a {@link Schema} from {@link #getAttributesField()} and {@link #getTimestampField()} @@ -565,8 +568,9 @@ Schema buildSchema(Field... additionalFields) { Schema.Builder builder = Schema.builder().addField(getAttributesField()).addField(getTimestampField()); - if (getPayloadField() != null) { - builder = builder.addField(getPayloadField()); + Field payloadField = getPayloadField(); + if (payloadField != null) { + builder = builder.addField(payloadField); } for (Field field : additionalFields) { diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java index d39cee3ddc06..bcecdcc00280 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java @@ -24,6 +24,8 @@ import static org.apache.beam.sdk.io.gcp.pubsub.PubsubMessageToRow.PAYLOAD_FIELD; import static org.apache.beam.sdk.io.gcp.pubsub.PubsubMessageToRow.TIMESTAMP_FIELD; import static org.apache.beam.sdk.schemas.Schema.TypeName.ROW; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import com.google.auto.service.AutoService; import com.google.auto.value.AutoValue; @@ -97,9 +99,6 @@ */ @Internal @AutoService(SchemaIOProvider.class) -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubSchemaIOProvider implements SchemaIOProvider { public static final FieldType ATTRIBUTE_MAP_FIELD_TYPE = Schema.FieldType.map(FieldType.STRING.withNullable(false), FieldType.STRING); @@ -137,11 +136,11 @@ public Schema configurationSchema() { * resides there, and some IO-specific configuration object. */ @Override - public PubsubSchemaIO from(String location, Row configuration, Schema dataSchema) { + public PubsubSchemaIO from(String location, Row configuration, @Nullable Schema dataSchema) { validateConfigurationSchema(configuration); - validateDlq(configuration.getValue("deadLetterQueue")); + validateDlq(configuration.getString("deadLetterQueue")); validateDataSchema(dataSchema); - return new PubsubSchemaIO(location, configuration, dataSchema); + return new PubsubSchemaIO(location, configuration, checkArgumentNotNull(dataSchema)); } @Override @@ -154,7 +153,7 @@ public PCollection.IsBounded isBounded() { return PCollection.IsBounded.UNBOUNDED; } - private void validateDataSchema(Schema schema) { + private void validateDataSchema(@Nullable Schema schema) { if (schema == null) { throw new InvalidSchemaException( "Unsupported schema specified for Pubsub source in CREATE TABLE." @@ -168,7 +167,7 @@ private void validateDataSchema(Schema schema) { } } - private void validateDlq(String deadLetterQueue) { + private void validateDlq(@Nullable String deadLetterQueue) { if (deadLetterQueue != null && deadLetterQueue.isEmpty()) { throw new InvalidConfigurationException("Dead letter queue topic name is not specified"); } @@ -252,7 +251,9 @@ public POutput expand(PCollection input) { MapElements.into(TypeDescriptor.of(PubsubMessage.class)) .via( row -> - new PubsubMessage(serializer.serialize(row), ImmutableMap.of()))); + new PubsubMessage( + checkStateNotNull(serializer).serialize(row), + ImmutableMap.of()))); } else { transformed = filtered.apply( @@ -268,33 +269,33 @@ private PubsubIO.Read readMessagesWithAttributes() { PubsubIO.Read read = PubsubIO.readMessagesWithAttributes().fromTopic(location); return config.useTimestampAttribute() - ? read.withTimestampAttribute(config.getTimestampAttributeKey()) + ? read.withTimestampAttribute(checkStateNotNull(config.getTimestampAttributeKey())) : read; } private PubsubIO.Write createPubsubMessageWrite() { PubsubIO.Write write = PubsubIO.writeMessages().to(location); if (config.useTimestampAttribute()) { - write = write.withTimestampAttribute(config.getTimestampAttributeKey()); + write = write.withTimestampAttribute(checkStateNotNull(config.getTimestampAttributeKey())); } return write; } private PubsubIO.Write writeMessagesToDlq() { PubsubIO.Write write = - PubsubIO.writeMessages().to(config.getDeadLetterQueue()); + PubsubIO.writeMessages().to(checkStateNotNull(config.getDeadLetterQueue())); return config.useTimestampAttribute() - ? write.withTimestampAttribute(config.getTimestampAttributeKey()) + ? write.withTimestampAttribute(checkStateNotNull(config.getTimestampAttributeKey())) : write; } - private boolean hasValidAttributesField(Schema schema) { + private static boolean hasValidAttributesField(Schema schema) { return fieldPresent(schema, ATTRIBUTES_FIELD, ATTRIBUTE_MAP_FIELD_TYPE) || fieldPresent(schema, ATTRIBUTES_FIELD, ATTRIBUTE_ARRAY_FIELD_TYPE); } - private boolean hasValidPayloadField(Schema schema) { + private static boolean hasValidPayloadField(Schema schema) { if (!schema.hasField(PAYLOAD_FIELD)) { return false; } @@ -304,7 +305,7 @@ private boolean hasValidPayloadField(Schema schema) { return schema.getField(PAYLOAD_FIELD).getType().getTypeName().equals(ROW); } - private boolean shouldUseNestedSchema(Schema schema) { + private static boolean shouldUseNestedSchema(Schema schema) { return hasValidPayloadField(schema) && hasValidAttributesField(schema); } @@ -350,16 +351,20 @@ boolean useTimestampAttribute() { } PayloadSerializer serializer(Schema schema) { - String format = getFormat() == null ? "json" : getFormat(); + String configFormat = getFormat(); + String format = configFormat == null ? "json" : configFormat; ImmutableMap.Builder params = ImmutableMap.builder(); - if (getThriftClass() != null) { - params.put("thriftClass", getThriftClass()); + String thriftClass = getThriftClass(); + if (thriftClass != null) { + params.put("thriftClass", thriftClass); } - if (getThriftProtocolFactoryClass() != null) { - params.put("thriftProtocolFactoryClass", getThriftProtocolFactoryClass()); + String thriftProtocolFactoryClass = getThriftProtocolFactoryClass(); + if (thriftProtocolFactoryClass != null) { + params.put("thriftProtocolFactoryClass", thriftProtocolFactoryClass); } - if (getProtoClass() != null) { - params.put("protoClass", getProtoClass()); + String protoClass = getProtoClass(); + if (protoClass != null) { + params.put("protoClass", protoClass); } return PayloadSerializers.getSerializer(format, schema, params.build()); } 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 22fcaae20cad..ee320bb80889 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 @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.google.api.client.util.Clock; @@ -40,9 +41,6 @@ * testing {@link #publish}, {@link #pull}, {@link #acknowledge} and {@link #modifyAckDeadline} * methods. Relies on statics to mimic the Pubsub service, though we try to hide that. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubTestClient extends PubsubClient implements Serializable { /** * Mimic the state of the simulated Pubsub 'service'. @@ -362,26 +360,28 @@ private static void setSchemaState( /** Handles verifying {@code STATE} at end of publish test. */ private static void performFinalPublishStateChecks() { checkState(STATE.isActive, "No test still in flight"); + Set remainingExpected = + checkStateNotNull(STATE.remainingExpectedOutgoingMessages); checkState( - STATE.remainingExpectedOutgoingMessages.isEmpty(), + remainingExpected.isEmpty(), "Still waiting for %s messages to be published", - STATE.remainingExpectedOutgoingMessages.size()); + remainingExpected.size()); } /** Handles verifying {@code STATE} at end of pull test. */ private static void performFinalPullStateChecks() { + List remainingPending = + checkStateNotNull(STATE.remainingPendingIncomingMessages); + Map pendingAck = checkStateNotNull(STATE.pendingAckIncomingMessages); + Map ackDeadline = checkStateNotNull(STATE.ackDeadline); checkState( - STATE.remainingPendingIncomingMessages.isEmpty(), + remainingPending.isEmpty(), "Still waiting for %s messages to be pulled", - STATE.remainingPendingIncomingMessages.size()); + remainingPending.size()); checkState( - STATE.pendingAckIncomingMessages.isEmpty(), - "Still waiting for %s messages to be ACKed", - STATE.pendingAckIncomingMessages.size()); + pendingAck.isEmpty(), "Still waiting for %s messages to be ACKed", pendingAck.size()); checkState( - STATE.ackDeadline.isEmpty(), - "Still waiting for %s messages to be ACKed", - STATE.ackDeadline.size()); + ackDeadline.isEmpty(), "Still waiting for %s messages to be ACKed", ackDeadline.size()); } public static PubsubTestClientFactory createFactoryForCreateSubscription() { @@ -445,13 +445,17 @@ private boolean inPublishMode() { public void advance() { synchronized (STATE) { checkState(inPullMode(), "Can only advance in pull mode"); + Clock clock = checkStateNotNull(STATE.clock); + Map ackDeadline = checkStateNotNull(STATE.ackDeadline); + Map pendingAck = checkStateNotNull(STATE.pendingAckIncomingMessages); + List remainingPending = + checkStateNotNull(STATE.remainingPendingIncomingMessages); // Any messages who's ACKs timed out are available for re-pulling. - Iterator> deadlineItr = STATE.ackDeadline.entrySet().iterator(); + Iterator> deadlineItr = ackDeadline.entrySet().iterator(); while (deadlineItr.hasNext()) { Map.Entry entry = deadlineItr.next(); - if (entry.getValue() <= STATE.clock.currentTimeMillis()) { - STATE.remainingPendingIncomingMessages.add( - STATE.pendingAckIncomingMessages.remove(entry.getKey())); + if (entry.getValue() <= clock.currentTimeMillis()) { + remainingPending.add(checkStateNotNull(pendingAck.remove(entry.getKey()))); deadlineItr.remove(); } } @@ -465,14 +469,19 @@ public void close() {} public int publish(TopicPath topic, List outgoingMessages) throws IOException { synchronized (STATE) { checkState(inPublishMode(), "Can only publish in publish mode"); - boolean isDynamic = STATE.expectedTopic == null; + TopicPath expectedTopic = STATE.expectedTopic; + boolean isDynamic = expectedTopic == null; if (!isDynamic) { checkState( - topic.equals(STATE.expectedTopic), + topic.equals(expectedTopic), "Topic %s does not match expected %s", topic, - STATE.expectedTopic); + expectedTopic); } + Set remainingExpected = + checkStateNotNull(STATE.remainingExpectedOutgoingMessages); + Set remainingFailing = + checkStateNotNull(STATE.remainingFailingOutgoingMessages); @MonotonicNonNull String batchOrderingKey = null; for (OutgoingMessage outgoingMessage : outgoingMessages) { if (batchOrderingKey == null) { @@ -480,15 +489,15 @@ public int publish(TopicPath topic, List outgoingMessages) thro } checkState(outgoingMessage.getMessage().getOrderingKey().equals(batchOrderingKey)); if (isDynamic) { - checkState(outgoingMessage.topic().equals(topic.getPath())); + checkState(topic.getPath().equals(outgoingMessage.topic())); } else { checkState(outgoingMessage.topic() == null); } - if (STATE.remainingFailingOutgoingMessages.remove(outgoingMessage)) { + if (remainingFailing.remove(outgoingMessage)) { throw new RuntimeException("Simulating failure for " + outgoingMessage); } checkState( - STATE.remainingExpectedOutgoingMessages.remove(outgoingMessage), + remainingExpected.remove(outgoingMessage), "Unexpected outgoing message %s", outgoingMessage); } @@ -505,21 +514,27 @@ public List pull( throws IOException { synchronized (STATE) { checkState(inPullMode(), "Can only pull in pull mode"); - long now = STATE.clock.currentTimeMillis(); + Clock clock = checkStateNotNull(STATE.clock); + long now = clock.currentTimeMillis(); checkState( requestTimeMsSinceEpoch == now, "Simulated time %s does not match request time %s", now, requestTimeMsSinceEpoch); + SubscriptionPath expectedSubscription = checkStateNotNull(STATE.expectedSubscription); checkState( - subscription.equals(STATE.expectedSubscription), + subscription.equals(expectedSubscription), "Subscription %s does not match expected %s", subscription, - STATE.expectedSubscription); + expectedSubscription); checkState(returnImmediately, "Pull only supported if returning immediately"); List incomingMessages = new ArrayList<>(); - Iterator pendItr = STATE.remainingPendingIncomingMessages.iterator(); + List remainingPending = + checkStateNotNull(STATE.remainingPendingIncomingMessages); + Map pendingAck = checkStateNotNull(STATE.pendingAckIncomingMessages); + Map ackDeadline = checkStateNotNull(STATE.ackDeadline); + Iterator pendItr = remainingPending.iterator(); while (pendItr.hasNext()) { IncomingMessage incomingMessage = pendItr.next(); pendItr.remove(); @@ -531,9 +546,8 @@ public List pull( incomingMessage.ackId(), incomingMessage.recordId()); incomingMessages.add(incomingMessageWithRequestTime); - STATE.pendingAckIncomingMessages.put( - incomingMessageWithRequestTime.ackId(), incomingMessageWithRequestTime); - STATE.ackDeadline.put( + pendingAck.put(incomingMessageWithRequestTime.ackId(), incomingMessageWithRequestTime); + ackDeadline.put( incomingMessageWithRequestTime.ackId(), requestTimeMsSinceEpoch + STATE.ackTimeoutSec * 1000L); if (incomingMessages.size() >= batchSize) { @@ -548,19 +562,22 @@ public List pull( public void acknowledge(SubscriptionPath subscription, List ackIds) throws IOException { synchronized (STATE) { checkState(inPullMode(), "Can only acknowledge in pull mode"); + SubscriptionPath expectedSubscription = checkStateNotNull(STATE.expectedSubscription); checkState( - subscription.equals(STATE.expectedSubscription), + subscription.equals(expectedSubscription), "Subscription %s does not match expected %s", subscription, - STATE.expectedSubscription); + expectedSubscription); + Map ackDeadline = checkStateNotNull(STATE.ackDeadline); + Map pendingAck = checkStateNotNull(STATE.pendingAckIncomingMessages); for (String ackId : ackIds) { checkState( - STATE.ackDeadline.remove(ackId) != null, + ackDeadline.remove(ackId) != null, "No message with ACK id %s is waiting for an ACK", ackId); checkState( - STATE.pendingAckIncomingMessages.remove(ackId) != null, + pendingAck.remove(ackId) != null, "No message with ACK id %s is waiting for an ACK", ackId); } @@ -572,31 +589,40 @@ public void modifyAckDeadline( SubscriptionPath subscription, List ackIds, int deadlineSeconds) throws IOException { synchronized (STATE) { checkState(inPullMode(), "Can only modify ack deadline in pull mode"); + SubscriptionPath expectedSubscription = checkStateNotNull(STATE.expectedSubscription); checkState( - subscription.equals(STATE.expectedSubscription), + subscription.equals(expectedSubscription), "Subscription %s does not match expected %s", subscription, - STATE.expectedSubscription); + expectedSubscription); + Map ackDeadline = checkStateNotNull(STATE.ackDeadline); + Map pendingAck = checkStateNotNull(STATE.pendingAckIncomingMessages); + Clock clock = checkStateNotNull(STATE.clock); + List remainingPending = + checkStateNotNull(STATE.remainingPendingIncomingMessages); for (String ackId : ackIds) { if (deadlineSeconds > 0) { checkState( - STATE.ackDeadline.remove(ackId) != null, + ackDeadline.remove(ackId) != null, "No message with ACK id %s is waiting for an ACK", ackId); checkState( - STATE.pendingAckIncomingMessages.containsKey(ackId), + pendingAck.containsKey(ackId), "No message with ACK id %s is waiting for an ACK", ackId); - STATE.ackDeadline.put(ackId, STATE.clock.currentTimeMillis() + deadlineSeconds * 1000L); + ackDeadline.put(ackId, clock.currentTimeMillis() + deadlineSeconds * 1000L); } else { checkState( - STATE.ackDeadline.remove(ackId) != null, + ackDeadline.remove(ackId) != null, "No message with ACK id %s is waiting for an ACK", ackId); - IncomingMessage message = STATE.pendingAckIncomingMessages.remove(ackId); - checkState(message != null, "No message with ACK id %s is waiting for an ACK", ackId); - STATE.remainingPendingIncomingMessages.add(message); + IncomingMessage message = + checkStateNotNull( + pendingAck.remove(ackId), + "No message with ACK id %s is waiting for an ACK", + ackId); + remainingPending.add(message); } } } @@ -656,7 +682,7 @@ public int ackDeadlineSeconds(SubscriptionPath subscription) throws IOException public boolean isEOF() { synchronized (STATE) { checkState(inPullMode(), "Can only check EOF in pull mode"); - return STATE.remainingPendingIncomingMessages.isEmpty(); + return checkStateNotNull(STATE.remainingPendingIncomingMessages).isEmpty(); } } @@ -674,12 +700,12 @@ public void deleteSchema(SchemaPath schemaPath) throws IOException { } @Override - public SchemaPath getSchemaPath(TopicPath topicPath) throws IOException { + public @Nullable SchemaPath getSchemaPath(TopicPath topicPath) throws IOException { return STATE.expectedSchemaPath; } @Override public Schema getSchema(SchemaPath schemaPath) throws IOException { - return STATE.expectedSchema; + return checkStateNotNull(STATE.expectedSchema, "expectedSchema is not set"); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java index 3fe7d51aec1e..45c0b329988d 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.google.protobuf.InvalidProtocolBufferException; @@ -96,9 +98,6 @@ * dedup messages. * */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubUnboundedSink extends PTransform, PDone> { /** Default maximum number of messages per publish. */ static final int DEFAULT_PUBLISH_BATCH_SIZE = 1000; @@ -109,7 +108,6 @@ public class PubsubUnboundedSink extends PTransform, /** Default longest delay between receiving a message and pushing it to Pubsub. */ private static final Duration DEFAULT_MAX_LATENCY = Duration.standardSeconds(2); - /** Coder for conveying outgoing messages between internal stages. */ /** Coder for conveying outgoing messages between internal stages. */ private static class OutgoingMessageCoder extends AtomicCoder { private static final NullableCoder RECORD_ID_CODER = @@ -198,7 +196,7 @@ public void processElement( recordId = Hashing.murmur3_128().hashBytes(elementBytes).toString(); break; case RANDOM: - // Since these elements go through a GroupByKey, any failures while sending to + // Since these elements go through a GroupByKey, any failures while sending to // Pubsub will be retried without falling back and generating a new record id. // Thus even though we may send the same message to Pubsub twice, it is guaranteed // to have the same record id. @@ -246,11 +244,11 @@ private static class OutgoingData { private final PubsubClientFactory pubsubFactory; private final @Nullable ValueProvider topic; - private final String timestampAttribute; - private final String idAttribute; + private final @Nullable String timestampAttribute; + private final @Nullable String idAttribute; private final int publishBatchBytes; - private final String pubsubRootUrl; + private final @Nullable String pubsubRootUrl; /** Client on which to talk to Pubsub. Null until created by {@link #startBundle}. */ private transient @Nullable PubsubClient pubsubClient; @@ -262,8 +260,8 @@ private static class OutgoingData { WriterFn( PubsubClientFactory pubsubFactory, @Nullable ValueProvider topic, - String timestampAttribute, - String idAttribute, + @Nullable String timestampAttribute, + @Nullable String idAttribute, int publishBatchSize, int publishBatchBytes) { this.pubsubFactory = pubsubFactory; @@ -277,10 +275,10 @@ private static class OutgoingData { WriterFn( PubsubClientFactory pubsubFactory, @Nullable ValueProvider topic, - String timestampAttribute, - String idAttribute, + @Nullable String timestampAttribute, + @Nullable String idAttribute, int publishBatchBytes, - String pubsubRootUrl) { + @Nullable String pubsubRootUrl) { this.pubsubFactory = pubsubFactory; this.topic = topic; this.timestampAttribute = timestampAttribute; @@ -290,11 +288,13 @@ private static class OutgoingData { } @VisibleForTesting + @Nullable String getIdAttribute() { return idAttribute; } @VisibleForTesting + @Nullable ValueProvider getTopic() { return topic; } @@ -308,11 +308,9 @@ private void publishBatch(List messages, int bytes) throws IOEx } else { // This is the dynamic topic destinations case. Since we first group by topic, we can assume // that all messages in the batch have the same topic. - topicPath = - PubsubClient.topicPathFromPath( - org.apache.beam.sdk.util.Preconditions.checkStateNotNull(messages.get(0).topic())); + topicPath = PubsubClient.topicPathFromPath(checkStateNotNull(messages.get(0).topic())); } - int n = pubsubClient.publish(topicPath, messages); + int n = checkStateNotNull(pubsubClient).publish(topicPath, messages); checkState( n == messages.size(), "Attempted to publish %s messages but %s were successful", @@ -380,8 +378,11 @@ public void processElement(ProcessContext c) throws Exception { @FinishBundle public void finishBundle() throws Exception { - pubsubClient.close(); - pubsubClient = null; + final PubsubClient pubsubClient = this.pubsubClient; + if (pubsubClient != null) { + pubsubClient.close(); + } + this.pubsubClient = null; } @Override @@ -448,21 +449,21 @@ public void populateDisplayData(DisplayData.Builder builder) { */ private final RecordIdMethod recordIdMethod; - private final String pubsubRootUrl; + private final @Nullable String pubsubRootUrl; @VisibleForTesting PubsubUnboundedSink( PubsubClientFactory pubsubFactory, @Nullable ValueProvider topic, - String timestampAttribute, - String idAttribute, + @Nullable String timestampAttribute, + @Nullable String idAttribute, int numShards, boolean publishBatchWithOrderingKey, int publishBatchSize, int publishBatchBytes, Duration maxLatency, RecordIdMethod recordIdMethod, - String pubsubRootUrl) { + @Nullable String pubsubRootUrl) { this.pubsubFactory = pubsubFactory; this.topic = topic; this.timestampAttribute = timestampAttribute; @@ -544,14 +545,14 @@ public PubsubUnboundedSink( public PubsubUnboundedSink( PubsubClientFactory pubsubFactory, - ValueProvider topic, - String timestampAttribute, - String idAttribute, + @Nullable ValueProvider topic, + @Nullable String timestampAttribute, + @Nullable String idAttribute, int numShards, boolean publishBatchWithOrderingKey, int publishBatchSize, int publishBatchBytes, - String pubsubRootUrl) { + @Nullable String pubsubRootUrl) { this( pubsubFactory, topic, @@ -604,7 +605,12 @@ public PDone expand(PCollection input) { return input .apply( "WithDynamicTopics", - WithKeys.of(PubsubMessage::getTopic).withKeyType(TypeDescriptors.strings())) + WithKeys.of( + (PubsubMessage msg) -> + checkArgumentNotNull( + msg.getTopic(), + "Topic must be set on PubsubMessage when writing to dynamic topics.")) + .withKeyType(TypeDescriptors.strings())) .apply( MapValues.into(new TypeDescriptor() {}) .via(new PubsubMessages.ParsePayloadAsPubsubMessageProto())) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java index bc91e9f381d4..a6f3917b0d59 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; @@ -114,9 +115,6 @@ * UnboundedSource.UnboundedReader} instances to execute concurrently and thus hide latency. * */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class PubsubUnboundedSource extends PTransform> { private static final Logger LOG = LoggerFactory.getLogger(PubsubUnboundedSource.class); @@ -265,12 +263,18 @@ public PubsubCheckpoint( */ @Override public void finalizeCheckpoint() throws IOException { - checkState(reader != null && safeToAckIds != null, "Cannot finalize a restored checkpoint"); - // Even if the 'true' active reader has changed since the checkpoint was taken we are + final PubsubReader reader = + checkStateNotNull(this.reader, "Cannot finalize a restored checkpoint"); + final List safeToAckIds = + checkStateNotNull(this.safeToAckIds, "Cannot finalize a restored checkpoint"); + // Even if the 'true' active reader has changed since the checkpoint was taken + // we are // fine: - // - The underlying Pubsub topic will not have changed, so the following ACKs will still + // - The underlying Pubsub topic will not have changed, so the following ACKs + // will still // go to the right place. - // - We'll delete the ACK ids from the readers in-flight state, but that only effects + // - We'll delete the ACK ids from the readers in-flight state, but that only + // effects // flow control and stats, neither of which are relevant anymore. try { int n = safeToAckIds.size(); @@ -291,8 +295,8 @@ public void finalizeCheckpoint() throws IOException { int remainingInFlight = reader.numInFlightCheckpoints.decrementAndGet(); checkState(remainingInFlight >= 0, "Miscounted in-flight checkpoints"); reader.maybeCloseClient(); - reader = null; - safeToAckIds = null; + this.reader = null; + this.safeToAckIds = null; } } @@ -330,7 +334,7 @@ public void nackAll(PubsubReader reader) throws IOException { /** The coder for our checkpoints. */ private static class PubsubCheckpointCoder extends AtomicCoder { - private static final Coder SUBSCRIPTION_PATH_CODER = + private static final Coder<@Nullable String> SUBSCRIPTION_PATH_CODER = NullableCoder.of(StringUtf8Coder.of()); private static final Coder> LIST_CODER = ListCoder.of(StringUtf8Coder.of()); @@ -348,7 +352,7 @@ public void encode(PubsubCheckpoint value, OutputStream outStream) throws IOExce @Override public PubsubCheckpoint decode(InputStream inStream) throws IOException { - String path = SUBSCRIPTION_PATH_CODER.decode(inStream); + @Nullable String path = SUBSCRIPTION_PATH_CODER.decode(inStream); List notYetReadIds = LIST_CODER.decode(inStream); return new PubsubCheckpoint(path, null, null, notYetReadIds); } @@ -370,13 +374,13 @@ static class PubsubReader extends UnboundedSource.UnboundedReader { @VisibleForTesting final SubscriptionPath subscription; /** Client on which to talk to Pubsub. Contains a null value if the client has been closed. */ - private AtomicReference pubsubClient; + private final AtomicReference<@Nullable PubsubClient> pubsubClient; /** * The closed state of this {@link PubsubReader}. If true, the reader has not yet been closed, * and it will have a non-null value within {@link #pubsubClient}. */ - private AtomicBoolean active = new AtomicBoolean(true); + private final AtomicBoolean active = new AtomicBoolean(true); /** * Ack timeout, in ms, as set on subscription when we first start reading. Not updated @@ -385,7 +389,7 @@ static class PubsubReader extends UnboundedSource.UnboundedReader { private int ackTimeoutMs; /** ACK ids of messages we have delivered downstream but not yet ACKed. */ - private Set safeToAckIds; + private final Set safeToAckIds; /** * Messages we have received from Pubsub and not yet delivered downstream. We preserve their @@ -427,13 +431,13 @@ public InFlightState(long requestTimeMsSinceEpoch, long ackDeadlineMsSinceEpoch) * Bucketed map from received time (as system time, ms since epoch) to message timestamps * (mssince epoch) of all received but not-yet read messages. Used to estimate watermark. */ - private BucketingFunction minUnreadTimestampMsSinceEpoch; + private final BucketingFunction minUnreadTimestampMsSinceEpoch; /** * Minimum of timestamps (ms since epoch) of all recently read messages. Used to estimate * watermark. */ - private MovingFunction minReadTimestampMsSinceEpoch; + private final MovingFunction minReadTimestampMsSinceEpoch; /** * System time (ms since epoch) we last received a message from Pubsub, or -1 if not yet @@ -454,60 +458,60 @@ public InFlightState(long requestTimeMsSinceEpoch, long ackDeadlineMsSinceEpoch) private long numReceived; /** Stats only: Number of messages which have recently been received. */ - private MovingFunction numReceivedRecently; + private final MovingFunction numReceivedRecently; /** Stats only: Number of messages which have recently had their deadline extended. */ - private MovingFunction numExtendedDeadlines; + private final MovingFunction numExtendedDeadlines; /** * Stats only: Number of messages which have recenttly had their deadline extended even though * it may be too late to do so. */ - private MovingFunction numLateDeadlines; + private final MovingFunction numLateDeadlines; /** Stats only: Number of messages which have recently been ACKed. */ - private MovingFunction numAcked; + private final MovingFunction numAcked; /** * Stats only: Number of messages which have recently expired (ACKs were extended for too long). */ - private MovingFunction numExpired; + private final MovingFunction numExpired; /** Stats only: Number of messages which have recently been NACKed. */ - private MovingFunction numNacked; + private final MovingFunction numNacked; /** Stats only: Number of message bytes which have recently been read by downstream consumer. */ - private MovingFunction numReadBytes; + private final MovingFunction numReadBytes; /** * Stats only: Minimum of timestamp (ms since epoch) of all recently received messages. Used to - * estimate timestamp skew. Does not contribute to watermark estimator. + * estimate timestamp skew. Does not contribute to watermark watermark. */ - private MovingFunction minReceivedTimestampMsSinceEpoch; + private final MovingFunction minReceivedTimestampMsSinceEpoch; /** * Stats only: Maximum of timestamp (ms since epoch) of all recently received messages. Used to * estimate timestamp skew. */ - private MovingFunction maxReceivedTimestampMsSinceEpoch; + private final MovingFunction maxReceivedTimestampMsSinceEpoch; /** Stats only: Minimum of recent estimated watermarks (ms since epoch). */ - private MovingFunction minWatermarkMsSinceEpoch; + private final MovingFunction minWatermarkMsSinceEpoch; /** Stats ony: Maximum of recent estimated watermarks (ms since epoch). */ - private MovingFunction maxWatermarkMsSinceEpoch; + private final MovingFunction maxWatermarkMsSinceEpoch; /** * Stats only: Number of messages with timestamps strictly behind the estimated watermark at the * time they are received. These may be considered 'late' by downstream computations. */ - private MovingFunction numLateMessages; + private final MovingFunction numLateMessages; /** * Stats only: Current number of checkpoints in flight. CAUTION: Accessed by both checkpointing * and reader threads. */ - private AtomicInteger numInFlightCheckpoints; + private final AtomicInteger numInFlightCheckpoints; /** Stats only: Maximum number of checkpoints in flight at any time. */ private int maxInFlightCheckpoints; @@ -563,7 +567,11 @@ public PubsubReader(PubsubOptions options, PubsubSource outer, SubscriptionPath @VisibleForTesting PubsubClient getPubsubClient() { - return pubsubClient.get(); + return getClient(); + } + + private PubsubClient getClient() { + return checkStateNotNull(pubsubClient.get(), "Client has been closed"); } /** @@ -575,7 +583,7 @@ PubsubClient getPubsubClient() { *

CAUTION: Retains {@code ackIds}. */ void ackBatch(List ackIds) throws IOException { - pubsubClient.get().acknowledge(subscription, ackIds); + getClient().acknowledge(subscription, ackIds); ackedIds.add(ackIds); } @@ -584,7 +592,7 @@ void ackBatch(List ackIds) throws IOException { * given {@code ackIds}. Does not retain {@code ackIds}. */ public void nackBatch(long nowMsSinceEpoch, List ackIds) throws IOException { - pubsubClient.get().modifyAckDeadline(subscription, ackIds, 0); + getClient().modifyAckDeadline(subscription, ackIds, 0); numNacked.add(nowMsSinceEpoch, ackIds.size()); } @@ -594,7 +602,7 @@ public void nackBatch(long nowMsSinceEpoch, List ackIds) throws IOExcept */ private void extendBatch(long nowMsSinceEpoch, List ackIds) throws IOException { int extensionSec = (ackTimeoutMs * ACK_EXTENSION_PCT) / (100 * 1000); - pubsubClient.get().modifyAckDeadline(subscription, ackIds, extensionSec); + getClient().modifyAckDeadline(subscription, ackIds, extensionSec); numExtendedDeadlines.add(nowMsSinceEpoch, ackIds.size()); } @@ -648,7 +656,8 @@ private void extend() throws IOException { < nowMsSinceEpoch) { // Pubsub may have already considered this message to have expired. // If so it will (eventually) be made available on a future pull request. - // If this message ends up being committed then it will be considered a duplicate + // If this message ends up being committed then it will be considered a + // duplicate // when re-pulled. assumeExpired.add(entry.getKey()); continue; @@ -693,11 +702,12 @@ private void extend() throws IOException { if (!toBeExtended.isEmpty()) { // Pubsub extends acks from it's notion of current time. - // We'll try to track that on our side, but note the deadlines won't necessarily agree. + // We'll try to track that on our side, but note the deadlines won't necessarily + // agree. long newDeadlineMsSinceEpoch = nowMsSinceEpoch + (ackTimeoutMs * ACK_EXTENSION_PCT) / 100; for (String ackId : toBeExtended) { // Maintain increasing ack deadline order. - InFlightState state = inFlight.remove(ackId); + InFlightState state = checkStateNotNull(inFlight.remove(ackId)); inFlight.put( ackId, new InFlightState(state.requestTimeMsSinceEpoch, newDeadlineMsSinceEpoch)); } @@ -711,8 +721,10 @@ private void extend() throws IOException { private void pull() throws IOException { if (inFlight.size() >= MAX_IN_FLIGHT) { // Wait for checkpoint to be finalized before pulling anymore. - // There may be lag while checkpoints are persisted and the finalizeCheckpoint method - // is invoked. By limiting the in-flight messages we can ensure we don't end up consuming + // There may be lag while checkpoints are persisted and the finalizeCheckpoint + // method + // is invoked. By limiting the in-flight messages we can ensure we don't end up + // consuming // messages faster than we can checkpoint them. return; } @@ -723,7 +735,7 @@ private void pull() throws IOException { // Pull the next batch. // BLOCKs until received. Collection receivedMessages = - pubsubClient.get().pull(requestTimeMsSinceEpoch, subscription, PULL_BATCH_SIZE, true); + getClient().pull(requestTimeMsSinceEpoch, subscription, PULL_BATCH_SIZE, true); if (receivedMessages.isEmpty()) { // Nothing available yet. Try again later. return; @@ -779,7 +791,8 @@ private void stats() { String oldestAckId = Iterables.getFirst(inFlight.keySet(), null); if (oldestAckId != null) { oldestInFlight = - (nowMsSinceEpoch - inFlight.get(oldestAckId).requestTimeMsSinceEpoch) + "ms"; + (nowMsSinceEpoch - checkStateNotNull(inFlight.get(oldestAckId)).requestTimeMsSinceEpoch) + + "ms"; } LOG.debug( @@ -828,7 +841,7 @@ private void stats() { @Override public boolean start() throws IOException { // Determine the ack timeout. - ackTimeoutMs = pubsubClient.get().ackDeadlineSeconds(subscription) * 1000; + ackTimeoutMs = getClient().ackDeadlineSeconds(subscription) * 1000; return advance(); } @@ -843,7 +856,8 @@ public boolean advance() throws IOException { stats(); if (current != null) { - // Current is consumed. It can no longer contribute to holding back the watermark. + // Current is consumed. It can no longer contribute to holding back the + // watermark. minUnreadTimestampMsSinceEpoch.remove(current.requestTimeMsSinceEpoch()); current = null; } @@ -853,24 +867,28 @@ public boolean advance() throws IOException { // Extend all pressing deadlines. // Will BLOCK until done. - // If the system is pulling messages only to let them sit in a downsteam queue then + // If the system is pulling messages only to let them sit in a downsteam queue + // then // this will have the effect of slowing down the pull rate. - // However, if the system is genuinely taking longer to process each message then + // However, if the system is genuinely taking longer to process each message + // then // the work to extend ACKs would be better done in the background. extend(); if (notYetRead.isEmpty()) { // Pull another batch. - // Will BLOCK until fetch returns, but will not block until a message is available. + // Will BLOCK until fetch returns, but will not block until a message is + // available. pull(); } // Take one message from queue. - current = notYetRead.poll(); + final PubsubClient.@Nullable IncomingMessage current = notYetRead.poll(); if (current == null) { // Try again later. return false; } + this.current = current; notYetReadBytes -= current.message().getData().size(); checkState(notYetReadBytes >= 0); long nowMsSinceEpoch = now(); @@ -888,6 +906,7 @@ public boolean advance() throws IOException { @Override public byte[] getCurrent() throws NoSuchElementException { + final PubsubClient.@Nullable IncomingMessage current = this.current; if (current == null) { throw new NoSuchElementException(); } @@ -901,6 +920,7 @@ public byte[] getCurrent() throws NoSuchElementException { @Override public Instant getCurrentTimestamp() throws NoSuchElementException { + final PubsubClient.@Nullable IncomingMessage current = this.current; if (current == null) { throw new NoSuchElementException(); } @@ -909,6 +929,7 @@ public Instant getCurrentTimestamp() throws NoSuchElementException { @Override public byte[] getCurrentRecordId() throws NoSuchElementException { + final PubsubClient.@Nullable IncomingMessage current = this.current; if (current == null) { throw new NoSuchElementException(); } @@ -934,7 +955,8 @@ public void close() throws IOException { */ private void maybeCloseClient() throws IOException { if (!active.get() && numInFlightCheckpoints.get() == 0) { - // The reader has been closed and it has no more outstanding checkpoints. The client + // The reader has been closed and it has no more outstanding checkpoints. The + // client // must be closed so it doesn't leak PubsubClient client = pubsubClient.getAndSet(null); if (client != null) { @@ -950,18 +972,23 @@ public PubsubSource getCurrentSource() { @Override public Instant getWatermark() { - if (pubsubClient.get().isEOF() && notYetRead.isEmpty()) { + if (getClient().isEOF() && notYetRead.isEmpty()) { // For testing only: Advance the watermark to the end of time to signal // the test is complete. return BoundedWindow.TIMESTAMP_MAX_VALUE; } - // NOTE: We'll allow the watermark to go backwards. The underlying runner is responsible - // for aggregating all reported watermarks and ensuring the aggregate is latched. - // If we attempt to latch locally then it is possible a temporary starvation of one reader - // could cause its estimated watermark to fast forward to current system time. Then when + // NOTE: We'll allow the watermark to go backwards. The underlying runner is + // responsible + // for aggregating all reported watermarks and ensuring the aggregate is + // latched. + // If we attempt to latch locally then it is possible a temporary starvation of + // one reader + // could cause its estimated watermark to fast forward to current system time. + // Then when // the reader resumes its watermark would be unable to resume tracking. - // By letting the underlying runner latch we avoid any problems due to localized starvation. + // By letting the underlying runner latch we avoid any problems due to localized + // starvation. long nowMsSinceEpoch = now(); long readMin = minReadTimestampMsSinceEpoch.get(nowMsSinceEpoch); long unreadMin = minUnreadTimestampMsSinceEpoch.get(); @@ -969,17 +996,21 @@ public Instant getWatermark() { && unreadMin == Long.MAX_VALUE && lastReceivedMsSinceEpoch >= 0 && nowMsSinceEpoch > lastReceivedMsSinceEpoch + SAMPLE_PERIOD.getMillis()) { - // We don't currently have any unread messages pending, we have not had any messages - // read for a while, and we have not received any new messages from Pubsub for a while. + // We don't currently have any unread messages pending, we have not had any + // messages + // read for a while, and we have not received any new messages from Pubsub for a + // while. // Advance watermark to current time. // TODO: Estimate a timestamp lag. lastWatermarkMsSinceEpoch = nowMsSinceEpoch; } else if (minReadTimestampMsSinceEpoch.isSignificant() || minUnreadTimestampMsSinceEpoch.isSignificant()) { - // Take minimum of the timestamps in all unread messages and recently read messages. + // Take minimum of the timestamps in all unread messages and recently read + // messages. lastWatermarkMsSinceEpoch = Math.min(readMin, unreadMin); } - // else: We're not confident enough to estimate a new watermark. Stick with the old one. + // else: We're not confident enough to estimate a new watermark. Stick with the + // old one. minWatermarkMsSinceEpoch.add(nowMsSinceEpoch, lastWatermarkMsSinceEpoch); maxWatermarkMsSinceEpoch.add(nowMsSinceEpoch, lastWatermarkMsSinceEpoch); return new Instant(lastWatermarkMsSinceEpoch); @@ -991,15 +1022,18 @@ public PubsubCheckpoint getCheckpointMark() { maxInFlightCheckpoints = Math.max(maxInFlightCheckpoints, cur); // It's possible for a checkpoint to be taken but never finalized. // So we simply copy whatever safeToAckIds we currently have. - // It's possible a later checkpoint will be taken before an earlier one is finalized, - // in which case we'll double ACK messages to Pubsub. However Pubsub is fine with that. + // It's possible a later checkpoint will be taken before an earlier one is + // finalized, + // in which case we'll double ACK messages to Pubsub. However Pubsub is fine + // with that. List snapshotSafeToAckIds = Lists.newArrayList(safeToAckIds); List snapshotNotYetReadIds = new ArrayList<>(notYetRead.size()); for (PubsubClient.IncomingMessage incomingMessage : notYetRead) { snapshotNotYetReadIds.add(incomingMessage.ackId()); } if (outer.subscriptionPath == null) { - // need to include the subscription in case we resume, as it's not stored in the source. + // need to include the subscription in case we resume, as it's not stored in the + // source. return new PubsubCheckpoint( subscription.getPath(), this, snapshotSafeToAckIds, snapshotNotYetReadIds); } @@ -1020,14 +1054,14 @@ public long getSplitBacklogBytes() { static class PubsubSource extends UnboundedSource { public final PubsubUnboundedSource outer; // The subscription to read from. - @VisibleForTesting final ValueProvider subscriptionPath; + @VisibleForTesting final @Nullable ValueProvider subscriptionPath; public PubsubSource(PubsubUnboundedSource outer) { this(outer, outer.getSubscriptionProvider()); } private PubsubSource( - PubsubUnboundedSource outer, ValueProvider subscriptionPath) { + PubsubUnboundedSource outer, @Nullable ValueProvider subscriptionPath) { this.outer = outer; this.subscriptionPath = subscriptionPath; } @@ -1055,16 +1089,22 @@ public PubsubReader createReader( PipelineOptions options, @Nullable PubsubCheckpoint checkpoint) { PubsubReader reader; SubscriptionPath subscription; - if (subscriptionPath == null || subscriptionPath.get() == null) { + final @Nullable ValueProvider subscriptionPath = this.subscriptionPath; + final @Nullable SubscriptionPath subscriptionPathVal = + subscriptionPath == null ? null : subscriptionPath.get(); + if (subscriptionPathVal == null) { if (checkpoint == null) { // This reader has never been started and there was no call to #split; // create a single random subscription, which will be kept in the checkpoint. subscription = outer.createRandomSubscription(options); } else { - subscription = checkpoint.getSubscription(); + subscription = + checkStateNotNull( + checkpoint.getSubscription(), + "Checkpoint must have a subscription path when subscriptionPath is not set."); } } else { - subscription = subscriptionPath.get(); + subscription = subscriptionPathVal; } try { reader = new PubsubReader(options.as(PubsubOptions.class), this, subscription); @@ -1088,7 +1128,7 @@ public PubsubReader createReader( } @Override - public @Nullable Coder getCheckpointMarkCoder() { + public Coder getCheckpointMarkCoder() { return CHECKPOINT_CODER; } @@ -1104,7 +1144,8 @@ public void validate() { @Override public boolean requiresDeduping() { - // We cannot prevent re-offering already read messages after a restore from checkpoint. + // We cannot prevent re-offering already read messages after a restore from + // checkpoint. return true; } } @@ -1205,7 +1246,7 @@ public void populateDisplayData(DisplayData.Builder builder) { @VisibleForTesting PubsubUnboundedSource( - Clock clock, + @Nullable Clock clock, PubsubClientFactory pubsubFactory, @Nullable ValueProvider project, @Nullable ValueProvider topic, @@ -1254,7 +1295,7 @@ public PubsubUnboundedSource( /** Construct an unbounded source to consume from the Pubsub {@code subscription}. */ public PubsubUnboundedSource( - Clock clock, + @Nullable Clock clock, PubsubClientFactory pubsubFactory, @Nullable ValueProvider project, @Nullable ValueProvider topic, @@ -1400,18 +1441,21 @@ private boolean usesStatsFn(PipelineOptions options) { } private SubscriptionPath createRandomSubscription(PipelineOptions options) { - TopicPath topicPath = topic.get(); + TopicPath topicPath = + checkStateNotNull( + checkStateNotNull(topic, "Topic must be set to create a random subscription").get(), + "Topic value must be set to create a random subscription"); ProjectPath projectPath; if (project != null) { - projectPath = project.get(); + projectPath = checkStateNotNull(project.get(), "Project value must be set"); } else { String projectId = options.as(GcpOptions.class).getProject(); checkState( projectId != null, "Cannot create subscription to topic %s because pipeline option 'project' not specified", topicPath); - projectPath = PubsubClient.projectPathFromId(options.as(GcpOptions.class).getProject()); + projectPath = PubsubClient.projectPathFromId(projectId); } try { diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java index 2abd6f5fa95d..1626cf50b7c7 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java @@ -17,6 +17,8 @@ */ package org.apache.beam.sdk.io.gcp.pubsub; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; + import com.google.auto.service.AutoService; import java.io.Serializable; import java.nio.charset.StandardCharsets; @@ -41,7 +43,6 @@ import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.sdk.values.TupleTagList; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets; @@ -106,7 +107,10 @@ public void processElement(@Element Row row, MultiOutputReceiver receiver) throw for (int ix = 0; ix < fields.size(); ix++) { String name = fields.get(ix).getName(); if (attributes != null && attributes.contains(name)) { - messageAttributes.put(name, row.getValue(ix)); + String val = row.getString(ix); + if (val != null) { + messageAttributes.put(name, val); + } } else if (name.equals(attributesMap)) { Map attrs = row.getMap(ix); if (attrs != null) { @@ -135,11 +139,11 @@ public void processElement(@Element Row row, MultiOutputReceiver receiver) throw @Override public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) { - if (!VALID_DATA_FORMATS.contains(configuration.getFormat().toUpperCase())) { + String format = checkArgumentNotNull(configuration.getFormat(), "Format must be specified"); + if (!VALID_DATA_FORMATS.contains(format.toUpperCase())) { throw new IllegalArgumentException( String.format( - "Format %s not supported. Only supported formats are %s", - configuration.getFormat(), VALID_FORMATS_STR)); + "Format %s not supported. Only supported formats are %s", format, VALID_FORMATS_STR)); } return new PubsubWriteSchemaTransform(configuration); } @@ -152,14 +156,10 @@ private static class PubsubWriteSchemaTransform extends SchemaTransform implemen } @Override - @SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) - }) public PCollectionRowTuple expand(PCollectionRowTuple input) { - String errorOutput = - configuration.getErrorHandling() == null - ? null - : configuration.getErrorHandling().getOutput(); + PubsubWriteSchemaTransformConfiguration.ErrorHandling errorHandling = + configuration.getErrorHandling(); + String errorOutput = errorHandling == null ? null : errorHandling.getOutput(); final Schema errorSchema = Schema.builder() @@ -167,16 +167,19 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { .addNullableRowField("row", input.get("input").getSchema()) .build(); - String format = configuration.getFormat(); + String format = checkArgumentNotNull(configuration.getFormat(), "Format must be specified"); Schema beamSchema = input.get("input").getSchema(); Schema payloadSchema; - if (configuration.getAttributes() == null && configuration.getAttributesMap() == null) { + List attributes = configuration.getAttributes(); + String attributesMap = configuration.getAttributesMap(); + if (attributes == null && attributesMap == null) { payloadSchema = beamSchema; } else { Schema.Builder payloadSchemaBuilder = Schema.builder(); for (Schema.Field f : beamSchema.getFields()) { - if (!configuration.getAttributes().contains(f.getName()) - && !f.getName().equals(configuration.getAttributesMap())) { + boolean isAttribute = attributes != null && attributes.contains(f.getName()); + boolean isAttributesMap = f.getName().equals(attributesMap); + if (!isAttribute && !isAttributesMap) { payloadSchemaBuilder.addField(f); } } @@ -190,9 +193,12 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { "Raw output only supported for single-field schemas, got %s", payloadSchema)); } if (payloadSchema.getField(0).getType().equals(Schema.FieldType.BYTES)) { - fn = row -> row.getBytes(0); + fn = row -> checkArgumentNotNull(row.getBytes(0), "Payload bytes value cannot be null"); } else if (payloadSchema.getField(0).getType().equals(Schema.FieldType.STRING)) { - fn = row -> row.getString(0).getBytes(StandardCharsets.UTF_8); + fn = + row -> + checkArgumentNotNull(row.getString(0), "Payload string value cannot be null") + .getBytes(StandardCharsets.UTF_8); } else { throw new IllegalArgumentException( String.format( @@ -217,20 +223,23 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { ParDo.of( new ErrorFn( fn, - configuration.getAttributes(), - configuration.getAttributesMap(), + attributes, + attributesMap, payloadSchema, errorSchema, errorOutput != null)) .withOutputTags(OUTPUT_TAG, TupleTagList.of(ERROR_TAG))); PubsubIO.Write writeTransform = - PubsubIO.writeMessages().to(configuration.getTopic()); - if (!Strings.isNullOrEmpty(configuration.getIdAttribute())) { - writeTransform = writeTransform.withIdAttribute(configuration.getIdAttribute()); + PubsubIO.writeMessages() + .to(checkArgumentNotNull(configuration.getTopic(), "Topic must be specified")); + String idAttribute = configuration.getIdAttribute(); + if (idAttribute != null && !idAttribute.isEmpty()) { + writeTransform = writeTransform.withIdAttribute(idAttribute); } - if (!Strings.isNullOrEmpty(configuration.getTimestampAttribute())) { - writeTransform = writeTransform.withIdAttribute(configuration.getTimestampAttribute()); + String timestampAttribute = configuration.getTimestampAttribute(); + if (timestampAttribute != null && !timestampAttribute.isEmpty()) { + writeTransform = writeTransform.withTimestampAttribute(timestampAttribute); } outputTuple.get(OUTPUT_TAG).apply(writeTransform); outputTuple.get(ERROR_TAG).setRowSchema(errorSchema); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsub.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsub.java index a55de012d383..902c9ec44459 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsub.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsub.java @@ -39,6 +39,7 @@ import io.grpc.ManagedChannelBuilder; import java.io.IOException; import java.util.List; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.LinkedBlockingDeque; @@ -71,9 +72,6 @@ * *

Deletes topic and subscription on shutdown. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class TestPubsub implements TestRule { private static final DateTimeFormatter DATETIME_FORMAT = DateTimeFormat.forPattern("YYYY-MM-dd-HH-mm-ss-SSS"); @@ -148,25 +146,29 @@ private void initializePubsub(Description description) throws IOException { } else { channel = ManagedChannelBuilder.forTarget(pubsubEndpoint).useTransportSecurity().build(); } - channelProvider = FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)); - topicAdmin = + TransportChannelProvider chanProvider = + FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)); + channelProvider = chanProvider; + TopicAdminClient topAdmin = TopicAdminClient.create( TopicAdminSettings.newBuilder() .setCredentialsProvider(pipelineOptions::getGcpCredential) - .setTransportChannelProvider(channelProvider) + .setTransportChannelProvider(chanProvider) .setEndpoint(pubsubEndpoint) .build()); - subscriptionAdmin = + topicAdmin = topAdmin; + SubscriptionAdminClient subAdmin = SubscriptionAdminClient.create( SubscriptionAdminSettings.newBuilder() .setCredentialsProvider(pipelineOptions::getGcpCredential) - .setTransportChannelProvider(channelProvider) + .setTransportChannelProvider(chanProvider) .setEndpoint(pubsubEndpoint) .build()); + subscriptionAdmin = subAdmin; TopicPath eventsTopicPathTmp = PubsubClient.topicPathFromName( pipelineOptions.getProject(), createTopicName(description, EVENTS_TOPIC_NAME)); - topicAdmin.createTopic(eventsTopicPathTmp.getPath()); + topAdmin.createTopic(eventsTopicPathTmp.getPath()); // Set this after successful creation; it signals that the topic needs teardown eventsTopicPath = eventsTopicPathTmp; @@ -178,7 +180,7 @@ private void initializePubsub(Description description) throws IOException { String.format( "projects/%s/subscriptions/%s", pipelineOptions.getProject(), subscriptionName)); - subscriptionAdmin.createSubscription( + subAdmin.createSubscription( subscriptionPathTmp.getPath(), topicPath().getPath(), PushConfig.getDefaultInstance(), @@ -188,25 +190,30 @@ private void initializePubsub(Description description) throws IOException { } private void tearDown() { - if (subscriptionAdmin == null || topicAdmin == null || channel == null) { + SubscriptionAdminClient subAdmin = subscriptionAdmin; + TopicAdminClient topAdmin = topicAdmin; + ManagedChannel chan = channel; + if (subAdmin == null || topAdmin == null || chan == null) { return; } try { - if (subscriptionPath != null) { - subscriptionAdmin.deleteSubscription(subscriptionPath.getPath()); + SubscriptionPath subPath = subscriptionPath; + if (subPath != null) { + subAdmin.deleteSubscription(subPath.getPath()); } - if (eventsTopicPath != null) { + TopicPath topPath = eventsTopicPath; + if (topPath != null) { for (String subscriptionPath : - topicAdmin.listTopicSubscriptions(eventsTopicPath.getPath()).iterateAll()) { - subscriptionAdmin.deleteSubscription(subscriptionPath); + topAdmin.listTopicSubscriptions(topPath.getPath()).iterateAll()) { + subAdmin.deleteSubscription(subscriptionPath); } - topicAdmin.deleteTopic(eventsTopicPath.getPath()); + topAdmin.deleteTopic(topPath.getPath()); } } finally { - subscriptionAdmin.close(); - topicAdmin.close(); - channel.shutdown(); + subAdmin.close(); + topAdmin.close(); + chan.shutdown(); subscriptionAdmin = null; topicAdmin = null; @@ -251,31 +258,37 @@ static String createTopicName(Description description, String name) throws IOExc /** Topic path where events will be published to. */ public TopicPath topicPath() { - return eventsTopicPath; + return Preconditions.checkNotNull(eventsTopicPath, "eventsTopicPath is not initialized"); } /** Subscription path used to listen for messages on {@link #topicPath()}. */ public SubscriptionPath subscriptionPath() { - return subscriptionPath; + return Preconditions.checkNotNull(subscriptionPath, "subscriptionPath is not initialized"); } private List listSubscriptions(TopicPath topicPath) { - Preconditions.checkNotNull(topicAdmin); + TopicAdminClient topAdmin = + Preconditions.checkNotNull(topicAdmin, "topicAdmin is not initialized"); + SubscriptionPath subPath = + Preconditions.checkNotNull(subscriptionPath, "subscriptionPath is not initialized"); // Exclude subscriptionPath, the subscription that we created - return Streams.stream(topicAdmin.listTopicSubscriptions(topicPath.getPath()).iterateAll()) - .filter((path) -> !path.equals(subscriptionPath.getPath())) + return Streams.stream(topAdmin.listTopicSubscriptions(topicPath.getPath()).iterateAll()) + .filter((path) -> !path.equals(subPath.getPath())) .collect(Collectors.toList()); } /** Publish messages to {@link #topicPath()}. */ public void publish(List messages) { - Preconditions.checkNotNull(eventsTopicPath); + TopicPath topPath = + Preconditions.checkNotNull(eventsTopicPath, "eventsTopicPath is not initialized"); + TransportChannelProvider chanProvider = + Preconditions.checkNotNull(channelProvider, "channelProvider is not initialized"); Publisher eventPublisher; try { eventPublisher = - Publisher.newBuilder(eventsTopicPath.getPath()) + Publisher.newBuilder(topPath.getPath()) .setCredentialsProvider(pipelineOptions::getGcpCredential) - .setChannelProvider(channelProvider) + .setChannelProvider(chanProvider) .setEndpoint(pubsubEndpoint) .build(); } catch (IOException e) { @@ -288,8 +301,11 @@ public void publish(List messages) { (message) -> { com.google.pubsub.v1.PubsubMessage.Builder builder = com.google.pubsub.v1.PubsubMessage.newBuilder() - .setData(ByteString.copyFrom(message.getPayload())) - .putAllAttributes(message.getAttributeMap()); + .setData(ByteString.copyFrom(message.getPayload())); + Map attributeMap = message.getAttributeMap(); + if (attributeMap != null) { + builder.putAllAttributes(attributeMap); + } return eventPublisher.publish(builder.build()); }) .collect(Collectors.toList()); @@ -311,7 +327,10 @@ public void publish(List messages) { */ public List waitForNMessages(int n, Duration timeoutDuration) throws IOException, InterruptedException { - Preconditions.checkNotNull(subscriptionPath); + SubscriptionPath subPath = + Preconditions.checkNotNull(subscriptionPath, "subscriptionPath is not initialized"); + TransportChannelProvider chanProvider = + Preconditions.checkNotNull(channelProvider, "channelProvider is not initialized"); BlockingQueue receivedMessages = new LinkedBlockingDeque<>(n); @@ -326,9 +345,9 @@ public List waitForNMessages(int n, Duration timeoutDuration) }; Subscriber subscriber = - Subscriber.newBuilder(subscriptionPath.getPath(), receiver) + Subscriber.newBuilder(subPath.getPath(), receiver) .setCredentialsProvider(pipelineOptions::getGcpCredential) - .setChannelProvider(channelProvider) + .setChannelProvider(chanProvider) .setEndpoint(pubsubEndpoint) .build(); subscriber.startAsync(); @@ -369,10 +388,13 @@ public List waitForNMessages(int n, Duration timeoutDuration) * hasProperty("payload", equalTo("hello".getBytes(StandardCharsets.US_ASCII))), * hasProperty("payload", equalTo("world".getBytes(StandardCharsets.US_ASCII)))) * .waitForUpTo(Duration.standardSeconds(20)); - * + * } * + * @param matchers Matchers to assert on the received messages. */ - public PollingAssertion assertThatTopicEventuallyReceives(Matcher... matchers) { + @SafeVarargs + public final PollingAssertion assertThatTopicEventuallyReceives( + Matcher... matchers) { return timeoutDuration -> assertThat( waitForNMessages(matchers.length, timeoutDuration), containsInAnyOrder(matchers)); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java index f11ba9555a80..02b27886f233 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java @@ -18,6 +18,7 @@ package org.apache.beam.sdk.io.gcp.pubsub; import static org.apache.beam.sdk.io.gcp.pubsub.TestPubsub.createTopicName; +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.google.cloud.pubsub.v1.AckReplyConsumer; @@ -74,7 +75,6 @@ *

Uses a random temporary Pubsub topic for synchronization. */ @SuppressWarnings({ - "nullness", // TODO(https://github.com/apache/beam/issues/20497) // TODO(https://github.com/apache/beam/issues/21230): Remove when new version of // errorprone is released (2.11.0) "unused" @@ -137,18 +137,20 @@ public void evaluate() throws Throwable { } private void initializePubsub(Description description) throws IOException { - topicAdmin = + TopicAdminClient topAdmin = TopicAdminClient.create( TopicAdminSettings.newBuilder() .setCredentialsProvider(pipelineOptions::getGcpCredential) .setEndpoint(pubsubEndpoint) .build()); - subscriptionAdmin = + topicAdmin = topAdmin; + SubscriptionAdminClient subAdmin = SubscriptionAdminClient.create( SubscriptionAdminSettings.newBuilder() .setCredentialsProvider(pipelineOptions::getGcpCredential) .setEndpoint(pubsubEndpoint) .build()); + subscriptionAdmin = subAdmin; // Example topic name: // integ-test-TestClassName-testMethodName-2018-12-11-23-32-333--result @@ -161,26 +163,26 @@ private void initializePubsub(Description description) throws IOException { SubscriptionPath resultSubscriptionPathTmp = PubsubClient.subscriptionPathFromName( pipelineOptions.getProject(), - "result-subscription-" + String.valueOf(ThreadLocalRandom.current().nextLong())); + "result-subscription-" + ThreadLocalRandom.current().nextLong()); SubscriptionPath startSubscriptionPathTmp = PubsubClient.subscriptionPathFromName( pipelineOptions.getProject(), - "start-subscription-" + String.valueOf(ThreadLocalRandom.current().nextLong())); + "start-subscription-" + ThreadLocalRandom.current().nextLong()); // Set variables after successful creation; this signals that they need teardown - topicAdmin.createTopic(resultTopicPathTmp.getPath()); + topAdmin.createTopic(resultTopicPathTmp.getPath()); resultTopicPath = resultTopicPathTmp; - topicAdmin.createTopic(startTopicPathTmp.getPath()); + topAdmin.createTopic(startTopicPathTmp.getPath()); startTopicPath = startTopicPathTmp; - subscriptionAdmin.createSubscription( + subAdmin.createSubscription( resultSubscriptionPathTmp.getPath(), - resultTopicPath.getPath(), + resultTopicPathTmp.getPath(), PushConfig.getDefaultInstance(), 600); resultSubscriptionPath = resultSubscriptionPathTmp; - subscriptionAdmin.createSubscription( + subAdmin.createSubscription( startSubscriptionPathTmp.getPath(), - startTopicPath.getPath(), + startTopicPathTmp.getPath(), PushConfig.getDefaultInstance(), 600); startSubscriptionPath = startSubscriptionPathTmp; @@ -195,25 +197,31 @@ private static void logIfThrows(ThrowingRunnable runnable) { } private void tearDown() { - if (subscriptionAdmin == null || topicAdmin == null) { + SubscriptionAdminClient subAdmin = subscriptionAdmin; + TopicAdminClient topAdmin = topicAdmin; + if (subAdmin == null || topAdmin == null) { return; } - if (resultSubscriptionPath != null) { - logIfThrows(() -> subscriptionAdmin.deleteSubscription(resultSubscriptionPath.getPath())); + SubscriptionPath resSubPath = resultSubscriptionPath; + if (resSubPath != null) { + logIfThrows(() -> subAdmin.deleteSubscription(resSubPath.getPath())); } - if (startSubscriptionPath != null) { - logIfThrows(() -> subscriptionAdmin.deleteSubscription(startSubscriptionPath.getPath())); + SubscriptionPath startSubPath = startSubscriptionPath; + if (startSubPath != null) { + logIfThrows(() -> subAdmin.deleteSubscription(startSubPath.getPath())); } - if (resultTopicPath != null) { - logIfThrows(() -> topicAdmin.deleteTopic(resultTopicPath.getPath())); + TopicPath resTopPath = resultTopicPath; + if (resTopPath != null) { + logIfThrows(() -> topAdmin.deleteTopic(resTopPath.getPath())); } - if (startTopicPath != null) { - logIfThrows(() -> topicAdmin.deleteTopic(startTopicPath.getPath())); + TopicPath startTopPathVal = startTopicPath; + if (startTopPathVal != null) { + logIfThrows(() -> topAdmin.deleteTopic(startTopPathVal.getPath())); } - logIfThrows(subscriptionAdmin::close); - logIfThrows(topicAdmin::close); + logIfThrows(subAdmin::close); + logIfThrows(topAdmin::close); subscriptionAdmin = null; topicAdmin = null; @@ -227,7 +235,7 @@ private void tearDown() { /** Outputs a message that the pipeline has started. */ public PTransform signalStart() { - return new PublishStart(startTopicPath); + return new PublishStart(checkStateNotNull(startTopicPath, "startTopicPath is not initialized")); } /** @@ -250,7 +258,11 @@ public PTransform, POutput> signalSuccessWhen( SerializableFunction formatter, SerializableFunction, Boolean> successPredicate) { - return new PublishSuccessWhen<>(coder, formatter, successPredicate, resultTopicPath); + return new PublishSuccessWhen<>( + coder, + formatter, + successPredicate, + checkStateNotNull(resultTopicPath, "resultTopicPath is not initialized")); } /** @@ -260,7 +272,7 @@ public PTransform, POutput> signalSuccessWhen( public PTransform, POutput> signalSuccessWhen( Coder coder, SerializableFunction, Boolean> successPredicate) { - return signalSuccessWhen(coder, T::toString, successPredicate); + return signalSuccessWhen(coder, String::valueOf, successPredicate); } /** @@ -270,10 +282,12 @@ public PTransform, POutput> signalSuccessWhen( * the start signal being published, which occurs immediately upon pipeline startup. */ public Supplier waitForStart(Duration duration) { + SubscriptionPath subPath = + checkStateNotNull(startSubscriptionPath, "startSubscriptionPath is not initialized"); return Suppliers.memoize( () -> { try { - String result = pollForResultForDuration(startSubscriptionPath, duration); + String result = pollForResultForDuration(subPath, duration); checkState(START_SIGNAL_MESSAGE.equals(result)); return null; } catch (IOException e) { @@ -284,7 +298,9 @@ public Supplier waitForStart(Duration duration) { /** Wait for a success signal for {@code duration}. */ public void waitForSuccess(Duration duration) throws IOException { - String result = pollForResultForDuration(resultSubscriptionPath, duration); + SubscriptionPath subPath = + checkStateNotNull(resultSubscriptionPath, "resultSubscriptionPath is not initialized"); + String result = pollForResultForDuration(subPath, duration); if (!RESULT_SUCCESS_MESSAGE.equals(result)) { throw new AssertionError(result); } @@ -293,7 +309,7 @@ public void waitForSuccess(Duration duration) throws IOException { private String pollForResultForDuration( SubscriptionPath signalSubscriptionPath, Duration timeoutDuration) throws IOException { - AtomicReference result = new AtomicReference<>(null); + AtomicReference<@Nullable String> result = new AtomicReference<>(null); MessageReceiver receiver = (com.google.pubsub.v1.PubsubMessage message, AckReplyConsumer replyConsumer) -> { @@ -330,13 +346,14 @@ private String pollForResultForDuration( subscriber.stopAsync(); subscriber.awaitTerminated(); - if (result.get() == null) { + String resultVal = result.get(); + if (resultVal == null) { throw new AssertionError( String.format( "Did not receive signal on %s in %ss", signalSubscriptionPath, timeoutDuration.getStandardSeconds())); } - return result.get(); + return resultVal; } /** {@link PTransform} that signals once when the pipeline has started. */ @@ -397,7 +414,7 @@ public POutput expand(PCollection input) { * "FAILURE". */ static class StatefulPredicateCheck extends DoFn, String> { - private SerializableFunction, Boolean> successPredicate; + private final SerializableFunction, Boolean> successPredicate; // keep all events seen so far in the state cell private static final String SEEN_EVENTS = "seenEvents";