gresockj commented on a change in pull request #5460: URL: https://github.com/apache/nifi/pull/5460#discussion_r740259834
########## File path: nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/lite/PublishGCPubSubLite.java ########## @@ -0,0 +1,330 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.gcp.pubsub.lite; + +import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; +import com.google.api.gax.batching.BatchingSettings; +import com.google.api.gax.core.FixedCredentialsProvider; +import com.google.api.gax.rpc.ApiException; +import com.google.cloud.pubsublite.TopicPath; +import com.google.cloud.pubsublite.cloudpubsub.Publisher; +import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; +import com.google.pubsub.v1.PubsubMessage; + +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.components.ConfigVerificationResult.Outcome; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.DataUnit; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.VerifiableProcessor; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.gcp.pubsub.AbstractGCPubSubProcessor; +import org.threeten.bp.Duration; + +import java.io.ByteArrayOutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.MESSAGE_ID_ATTRIBUTE; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.MESSAGE_ID_DESCRIPTION; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.TOPIC_NAME_ATTRIBUTE; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.TOPIC_NAME_DESCRIPTION; + +@SeeAlso({ConsumeGCPubSubLite.class}) +@InputRequirement(Requirement.INPUT_REQUIRED) +@Tags({"google", "google-cloud", "gcp", "message", "pubsub", "publish", "lite"}) +@CapabilityDescription("Publishes the content of the incoming flowfile to the configured Google Cloud PubSub Lite topic. The processor supports dynamic properties." + + " If any dynamic properties are present, they will be sent along with the message in the form of 'attributes'. In its current state, this processor will " + + "only work if running on a Google Cloud Compute Engine instance and if using the GCP Credentials Controller Service with 'Use Application Default " + + "Credentials' or 'Use Compute Engine Credentials'.") +@DynamicProperty(name = "Attribute name", value = "Value to be set to the attribute", + description = "Attributes to be set for the outgoing Google Cloud PubSub Lite message", expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) +@WritesAttributes({ + @WritesAttribute(attribute = MESSAGE_ID_ATTRIBUTE, description = MESSAGE_ID_DESCRIPTION), + @WritesAttribute(attribute = TOPIC_NAME_ATTRIBUTE, description = TOPIC_NAME_DESCRIPTION) +}) +@SystemResourceConsideration(resource = SystemResource.MEMORY, description = "The entirety of the FlowFile's content " + + "will be read into memory to be sent as a PubSub message.") +public class PublishGCPubSubLite extends AbstractGCPubSubProcessor implements VerifiableProcessor { + + public static final PropertyDescriptor TOPIC_NAME = new PropertyDescriptor.Builder() + .name("gcp-pubsub-topic") + .displayName("Topic Name") + .description("Name of the Google Cloud PubSub Topic. Example: projects/8476107443/locations/europe-west1-d/topics/my-lite-topic") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + public static final PropertyDescriptor ORDERING_KEY = new PropertyDescriptor + .Builder().name("gcp-ordering_key") + .displayName("Ordering Key") + .description("Messages with the same ordering key will always get published to the same partition. When this property is not " + + "set, messages can get published to different partitions if more than one partition exists for the topic.") + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + + public static final PropertyDescriptor BATCH_BYTES = new PropertyDescriptor + .Builder().name("gcp-batch-bytes") + .displayName("Batch Bytes Threshold") + .description("Publish request gets triggered based on this Batch Bytes Threshold property and" + + " the " + BATCH_SIZE.getDisplayName() + " property, whichever condition is met first.") + .required(true) + .defaultValue("3 MB") + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .addValidator(StandardValidators.DATA_SIZE_VALIDATOR) + .build(); + + private Publisher publisher = null; + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return ImmutableList.of(TOPIC_NAME, + GCP_CREDENTIALS_PROVIDER_SERVICE, + ORDERING_KEY, + BATCH_SIZE, + BATCH_BYTES); + } + + @Override + protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) { + return new PropertyDescriptor.Builder() + .required(false) + .name(propertyDescriptorName) + .displayName(propertyDescriptorName) + .addValidator(StandardValidators.ATTRIBUTE_KEY_PROPERTY_NAME_VALIDATOR) + .dynamic(true) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + } + + @Override + public Set<Relationship> getRelationships() { + return Collections.unmodifiableSet( + new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE)) + ); + } + + @Override + protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) { + final Collection<ValidationResult> results = new ArrayList<ValidationResult>(1); + final String topic = validationContext.getProperty(TOPIC_NAME).evaluateAttributeExpressions().getValue(); + + try { + TopicPath.parse(topic); + } catch (final ApiException e) { + results.add(new ValidationResult.Builder() + .subject(TOPIC_NAME.getName()) + .input(topic) + .valid(false) + .explanation("The Topic does not have a valid format.") + .build()); + } + + return results; + } + + @Override + @OnScheduled + public void onScheduled(final ProcessContext context) { + try { + if (publisher == null) { + publisher = getPublisher(context); + } + } catch (final Exception e) { + getLogger().error("Failed to create Google Cloud PubSub Lite Publisher", e); + throw new ProcessException(e); + } + try { + publisher.startAsync().awaitRunning(); + } catch (final Exception e) { + getLogger().error("Failed to create Google Cloud PubSub Lite Publisher", publisher.failureCause()); + throw new ProcessException(e); + } + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + final int flowFileCount = context.getProperty(BATCH_SIZE).asInteger(); + final List<FlowFile> flowFiles = session.get(flowFileCount); + + if (flowFiles.isEmpty()) { + context.yield(); + return; + } + + if (publisher == null) { + getLogger().error("Google Cloud PubSub Lite Publisher was not properly created. Yielding the processor..."); + context.yield(); + return; + } + + if(!publisher.isRunning()) { + getLogger().error("Google Cloud PubSub Lite Publisher is not running. Yielding the processor...", publisher.failureCause()); + throw new ProcessException(publisher.failureCause()); + } + + final long startNanos = System.nanoTime(); + final List<FlowFile> successfulFlowFiles = new ArrayList<>(); + final String topicName = context.getProperty(TOPIC_NAME).evaluateAttributeExpressions().getValue(); + final List<ApiFuture<String>> futures = new ArrayList<>(); + + try { + for (FlowFile flowFile : flowFiles) { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + session.exportTo(flowFile, baos); + final ByteString flowFileContent = ByteString.copyFrom(baos.toByteArray()); + final String orderingKey = context.getProperty(ORDERING_KEY).evaluateAttributeExpressions(flowFile).getValue(); + + PubsubMessage message= null; + if (orderingKey != null) { + message = PubsubMessage.newBuilder().setData(flowFileContent) + .setOrderingKey(orderingKey) + .setPublishTime(Timestamp.newBuilder().build()) + .putAllAttributes(getDynamicAttributesMap(context, flowFile)) + .build(); + } else { + message = PubsubMessage.newBuilder().setData(flowFileContent) + .setPublishTime(Timestamp.newBuilder().build()) + .putAllAttributes(getDynamicAttributesMap(context, flowFile)) + .build(); + } Review comment: This can be streamlined a little by declaring the builder as a final local variable, and then adding the orderingKey if it's non-null. ########## File path: nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/lite/PublishGCPubSubLite.java ########## @@ -0,0 +1,330 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.gcp.pubsub.lite; + +import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; +import com.google.api.gax.batching.BatchingSettings; +import com.google.api.gax.core.FixedCredentialsProvider; +import com.google.api.gax.rpc.ApiException; +import com.google.cloud.pubsublite.TopicPath; +import com.google.cloud.pubsublite.cloudpubsub.Publisher; +import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; +import com.google.pubsub.v1.PubsubMessage; + +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.components.ConfigVerificationResult.Outcome; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.DataUnit; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.VerifiableProcessor; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.gcp.pubsub.AbstractGCPubSubProcessor; +import org.threeten.bp.Duration; + +import java.io.ByteArrayOutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.MESSAGE_ID_ATTRIBUTE; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.MESSAGE_ID_DESCRIPTION; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.TOPIC_NAME_ATTRIBUTE; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.TOPIC_NAME_DESCRIPTION; + +@SeeAlso({ConsumeGCPubSubLite.class}) +@InputRequirement(Requirement.INPUT_REQUIRED) +@Tags({"google", "google-cloud", "gcp", "message", "pubsub", "publish", "lite"}) +@CapabilityDescription("Publishes the content of the incoming flowfile to the configured Google Cloud PubSub Lite topic. The processor supports dynamic properties." + + " If any dynamic properties are present, they will be sent along with the message in the form of 'attributes'. In its current state, this processor will " + + "only work if running on a Google Cloud Compute Engine instance and if using the GCP Credentials Controller Service with 'Use Application Default " + + "Credentials' or 'Use Compute Engine Credentials'.") +@DynamicProperty(name = "Attribute name", value = "Value to be set to the attribute", + description = "Attributes to be set for the outgoing Google Cloud PubSub Lite message", expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) +@WritesAttributes({ + @WritesAttribute(attribute = MESSAGE_ID_ATTRIBUTE, description = MESSAGE_ID_DESCRIPTION), + @WritesAttribute(attribute = TOPIC_NAME_ATTRIBUTE, description = TOPIC_NAME_DESCRIPTION) +}) +@SystemResourceConsideration(resource = SystemResource.MEMORY, description = "The entirety of the FlowFile's content " + + "will be read into memory to be sent as a PubSub message.") +public class PublishGCPubSubLite extends AbstractGCPubSubProcessor implements VerifiableProcessor { + + public static final PropertyDescriptor TOPIC_NAME = new PropertyDescriptor.Builder() + .name("gcp-pubsub-topic") + .displayName("Topic Name") + .description("Name of the Google Cloud PubSub Topic. Example: projects/8476107443/locations/europe-west1-d/topics/my-lite-topic") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + public static final PropertyDescriptor ORDERING_KEY = new PropertyDescriptor + .Builder().name("gcp-ordering_key") Review comment: Perhaps `gcp-ordering-key` instead of `gcp-ordering_key` ########## File path: nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/lite/PublishGCPubSubLite.java ########## @@ -0,0 +1,330 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.processors.gcp.pubsub.lite; + +import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; +import com.google.api.gax.batching.BatchingSettings; +import com.google.api.gax.core.FixedCredentialsProvider; +import com.google.api.gax.rpc.ApiException; +import com.google.cloud.pubsublite.TopicPath; +import com.google.cloud.pubsublite.cloudpubsub.Publisher; +import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; +import com.google.pubsub.v1.PubsubMessage; + +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.SystemResource; +import org.apache.nifi.annotation.behavior.SystemResourceConsideration; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.components.ConfigVerificationResult.Outcome; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.DataUnit; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.VerifiableProcessor; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.gcp.pubsub.AbstractGCPubSubProcessor; +import org.threeten.bp.Duration; + +import java.io.ByteArrayOutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.MESSAGE_ID_ATTRIBUTE; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.MESSAGE_ID_DESCRIPTION; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.TOPIC_NAME_ATTRIBUTE; +import static org.apache.nifi.processors.gcp.pubsub.PubSubAttributes.TOPIC_NAME_DESCRIPTION; + +@SeeAlso({ConsumeGCPubSubLite.class}) +@InputRequirement(Requirement.INPUT_REQUIRED) +@Tags({"google", "google-cloud", "gcp", "message", "pubsub", "publish", "lite"}) +@CapabilityDescription("Publishes the content of the incoming flowfile to the configured Google Cloud PubSub Lite topic. The processor supports dynamic properties." + + " If any dynamic properties are present, they will be sent along with the message in the form of 'attributes'. In its current state, this processor will " + + "only work if running on a Google Cloud Compute Engine instance and if using the GCP Credentials Controller Service with 'Use Application Default " + + "Credentials' or 'Use Compute Engine Credentials'.") +@DynamicProperty(name = "Attribute name", value = "Value to be set to the attribute", + description = "Attributes to be set for the outgoing Google Cloud PubSub Lite message", expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) +@WritesAttributes({ + @WritesAttribute(attribute = MESSAGE_ID_ATTRIBUTE, description = MESSAGE_ID_DESCRIPTION), + @WritesAttribute(attribute = TOPIC_NAME_ATTRIBUTE, description = TOPIC_NAME_DESCRIPTION) +}) +@SystemResourceConsideration(resource = SystemResource.MEMORY, description = "The entirety of the FlowFile's content " + + "will be read into memory to be sent as a PubSub message.") +public class PublishGCPubSubLite extends AbstractGCPubSubProcessor implements VerifiableProcessor { + + public static final PropertyDescriptor TOPIC_NAME = new PropertyDescriptor.Builder() + .name("gcp-pubsub-topic") + .displayName("Topic Name") + .description("Name of the Google Cloud PubSub Topic. Example: projects/8476107443/locations/europe-west1-d/topics/my-lite-topic") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .build(); + + public static final PropertyDescriptor ORDERING_KEY = new PropertyDescriptor + .Builder().name("gcp-ordering_key") + .displayName("Ordering Key") + .description("Messages with the same ordering key will always get published to the same partition. When this property is not " + + "set, messages can get published to different partitions if more than one partition exists for the topic.") + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + + public static final PropertyDescriptor BATCH_BYTES = new PropertyDescriptor + .Builder().name("gcp-batch-bytes") + .displayName("Batch Bytes Threshold") + .description("Publish request gets triggered based on this Batch Bytes Threshold property and" + + " the " + BATCH_SIZE.getDisplayName() + " property, whichever condition is met first.") + .required(true) + .defaultValue("3 MB") + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .addValidator(StandardValidators.DATA_SIZE_VALIDATOR) + .build(); + + private Publisher publisher = null; + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return ImmutableList.of(TOPIC_NAME, + GCP_CREDENTIALS_PROVIDER_SERVICE, + ORDERING_KEY, + BATCH_SIZE, + BATCH_BYTES); + } + + @Override + protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) { + return new PropertyDescriptor.Builder() + .required(false) + .name(propertyDescriptorName) + .displayName(propertyDescriptorName) + .addValidator(StandardValidators.ATTRIBUTE_KEY_PROPERTY_NAME_VALIDATOR) + .dynamic(true) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + } + + @Override + public Set<Relationship> getRelationships() { + return Collections.unmodifiableSet( + new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE)) + ); + } + + @Override + protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) { + final Collection<ValidationResult> results = new ArrayList<ValidationResult>(1); + final String topic = validationContext.getProperty(TOPIC_NAME).evaluateAttributeExpressions().getValue(); + + try { + TopicPath.parse(topic); + } catch (final ApiException e) { + results.add(new ValidationResult.Builder() + .subject(TOPIC_NAME.getName()) + .input(topic) + .valid(false) + .explanation("The Topic does not have a valid format.") + .build()); + } + + return results; + } + + @Override + @OnScheduled + public void onScheduled(final ProcessContext context) { + try { + if (publisher == null) { + publisher = getPublisher(context); + } + } catch (final Exception e) { + getLogger().error("Failed to create Google Cloud PubSub Lite Publisher", e); + throw new ProcessException(e); + } + try { + publisher.startAsync().awaitRunning(); + } catch (final Exception e) { + getLogger().error("Failed to create Google Cloud PubSub Lite Publisher", publisher.failureCause()); + throw new ProcessException(e); + } + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + final int flowFileCount = context.getProperty(BATCH_SIZE).asInteger(); + final List<FlowFile> flowFiles = session.get(flowFileCount); + + if (flowFiles.isEmpty()) { + context.yield(); + return; + } + + if (publisher == null) { + getLogger().error("Google Cloud PubSub Lite Publisher was not properly created. Yielding the processor..."); + context.yield(); + return; + } + + if(!publisher.isRunning()) { + getLogger().error("Google Cloud PubSub Lite Publisher is not running. Yielding the processor...", publisher.failureCause()); + throw new ProcessException(publisher.failureCause()); + } + + final long startNanos = System.nanoTime(); + final List<FlowFile> successfulFlowFiles = new ArrayList<>(); + final String topicName = context.getProperty(TOPIC_NAME).evaluateAttributeExpressions().getValue(); + final List<ApiFuture<String>> futures = new ArrayList<>(); + + try { + for (FlowFile flowFile : flowFiles) { + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); + session.exportTo(flowFile, baos); + final ByteString flowFileContent = ByteString.copyFrom(baos.toByteArray()); + final String orderingKey = context.getProperty(ORDERING_KEY).evaluateAttributeExpressions(flowFile).getValue(); + + PubsubMessage message= null; Review comment: Can be final, also missing a space -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
