Skip to content

Commit

Permalink
NIFI-12933 Rearranged properties on GCP processors
Browse files Browse the repository at this point in the history 
Also used current API methods for relationships/properties collections

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

This closes apache#8545.
  • Loading branch information
@turcsanyip @pvillard31
turcsanyip authored and pvillard31 committed Mar 25, 2024
1 parent 635eb9e commit 8eb013a
Show file tree
Hide file tree
Showing 13 changed files with 146 additions and 157 deletions.
9 changes: 0 additions & 9 deletions ...ifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/AbstractGCPProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,15 +85,6 @@ protected CloudService getCloudService() {
return cloudService;
}

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return List.of(PROJECT_ID,
GCP_CREDENTIALS_PROVIDER_SERVICE,
RETRY_COUNT,
PROXY_CONFIGURATION_SERVICE
);
}


@Override
public void migrateProperties(final PropertyConfiguration config) {
Expand Down
17 changes: 2 additions & 15 deletions ...sors/src/main/java/org/apache/nifi/processors/gcp/bigquery/AbstractBigQueryProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,8 @@
import org.apache.nifi.util.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand All @@ -63,8 +61,7 @@ public abstract class AbstractBigQueryProcessor extends AbstractGCPProcessor<Big
.description("FlowFiles are routed to this relationship if the Google BigQuery operation fails.")
.build();

public static final Set<Relationship> relationships = Collections.unmodifiableSet(
new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE)));
public static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS, REL_FAILURE);

public static final PropertyDescriptor DATASET = new PropertyDescriptor.Builder()
.name(BigQueryAttributes.DATASET_ATTR)
Expand Down Expand Up @@ -98,17 +95,7 @@ public abstract class AbstractBigQueryProcessor extends AbstractGCPProcessor<Big

@Override
public Set<Relationship> getRelationships() {
return relationships;
}

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> descriptors = new ArrayList<>();
descriptors.addAll(super.getSupportedPropertyDescriptors());
descriptors.add(DATASET);
descriptors.add(TABLE_NAME);
descriptors.add(IGNORE_UNKNOWN);
return Collections.unmodifiableList(descriptors);
return RELATIONSHIPS;
}

@Override
Expand Down
12 changes: 3 additions & 9 deletions ...ifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/bigquery/PutBigQuery.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,6 @@
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.gcp.ProxyAwareTransportFactory;
import org.apache.nifi.processors.gcp.bigquery.proto.ProtoUtils;
import org.apache.nifi.proxy.ProxyConfiguration;
import org.apache.nifi.serialization.RecordReader;
Expand All @@ -80,18 +79,13 @@
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Phaser;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;

import static java.util.stream.Collectors.collectingAndThen;
import static java.util.stream.Collectors.toList;

@TriggerSerially
@Tags({"google", "google cloud", "bq", "bigquery"})
Expand Down Expand Up @@ -179,7 +173,7 @@ public class PutBigQuery extends AbstractBigQueryProcessor {
.defaultValue("false")
.build();

private static final List<PropertyDescriptor> DESCRIPTORS = Stream.of(
private static final List<PropertyDescriptor> DESCRIPTORS = List.of(
GCP_CREDENTIALS_PROVIDER_SERVICE,
PROJECT_ID,
BIGQUERY_API_ENDPOINT,
Expand All @@ -190,8 +184,8 @@ public class PutBigQuery extends AbstractBigQueryProcessor {
APPEND_RECORD_COUNT,
RETRY_COUNT,
SKIP_INVALID_ROWS,
ProxyConfiguration.createProxyConfigPropertyDescriptor(false, ProxyAwareTransportFactory.PROXY_SPECS)
).collect(collectingAndThen(toList(), Collections::unmodifiableList));
PROXY_CONFIGURATION_SERVICE
);

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
Expand Down
10 changes: 0 additions & 10 deletions ...c/main/java/org/apache/nifi/processors/gcp/pubsub/AbstractGCPubSubWithProxyProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,21 +24,11 @@
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.net.SocketAddress;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.proxy.ProxyConfiguration;

public abstract class AbstractGCPubSubWithProxyProcessor extends AbstractGCPubSubProcessor {
@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return List.of(
PROJECT_ID,
PROXY_CONFIGURATION_SERVICE,
GCP_CREDENTIALS_PROVIDER_SERVICE
);
}

protected TransportChannelProvider getTransportChannelProvider(ProcessContext context) {
final ProxyConfiguration proxyConfiguration = ProxyConfiguration.getConfiguration(context);
Expand Down
35 changes: 21 additions & 14 deletions ...i-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/ConsumeGCPubSub.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,11 +95,32 @@ public class ConsumeGCPubSub extends AbstractGCPubSubWithProxyProcessor {
.expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
.build();

private static final List<PropertyDescriptor> DESCRIPTORS = List.of(
GCP_CREDENTIALS_PROVIDER_SERVICE,
PROJECT_ID,
SUBSCRIPTION,
BATCH_SIZE_THRESHOLD,
API_ENDPOINT,
PROXY_CONFIGURATION_SERVICE
);

public static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS);

private SubscriberStub subscriber = null;
private PullRequest pullRequest;

private final AtomicReference<Exception> storedException = new AtomicReference<>();

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return DESCRIPTORS;
}

@Override
public Set<Relationship> getRelationships() {
return RELATIONSHIPS;
}

@OnScheduled
public void onScheduled(ProcessContext context) {
final Integer batchSize = context.getProperty(BATCH_SIZE_THRESHOLD).asInteger();
Expand Down Expand Up @@ -188,20 +209,6 @@ public void onStopped() {
}
}

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> descriptors = new ArrayList<>(super.getSupportedPropertyDescriptors());
descriptors.add(SUBSCRIPTION);
descriptors.add(BATCH_SIZE_THRESHOLD);
descriptors.add(API_ENDPOINT);
return Collections.unmodifiableList(descriptors);
}

@Override
public Set<Relationship> getRelationships() {
return Collections.singleton(REL_SUCCESS);
}

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
if (subscriber == null) {
Expand Down
37 changes: 20 additions & 17 deletions ...i-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/PublishGCPubSub.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,10 +75,8 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -173,22 +171,29 @@ public class PublishGCPubSub extends AbstractGCPubSubWithProxyProcessor {
.description("FlowFiles are routed to this relationship if the Google Cloud Pub/Sub operation fails but attempting the operation again may succeed.")
.build();

private static final List<PropertyDescriptor> DESCRIPTORS = List.of(
GCP_CREDENTIALS_PROVIDER_SERVICE,
PROJECT_ID,
TOPIC_NAME,
MESSAGE_DERIVATION_STRATEGY,
RECORD_READER,
RECORD_WRITER,
MAX_BATCH_SIZE,
MAX_MESSAGE_SIZE,
BATCH_SIZE_THRESHOLD,
BATCH_BYTES_THRESHOLD,
BATCH_DELAY_THRESHOLD,
API_ENDPOINT,
PROXY_CONFIGURATION_SERVICE
);

public static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS, REL_FAILURE, REL_RETRY);

protected Publisher publisher = null;

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> descriptors = new ArrayList<>(super.getSupportedPropertyDescriptors());
descriptors.add(MAX_BATCH_SIZE);
descriptors.add(MESSAGE_DERIVATION_STRATEGY);
descriptors.add(RECORD_READER);
descriptors.add(RECORD_WRITER);
descriptors.add(MAX_MESSAGE_SIZE);
descriptors.add(TOPIC_NAME);
descriptors.add(BATCH_SIZE_THRESHOLD);
descriptors.add(BATCH_BYTES_THRESHOLD);
descriptors.add(BATCH_DELAY_THRESHOLD);
descriptors.add(API_ENDPOINT);
return Collections.unmodifiableList(descriptors);
return DESCRIPTORS;
}

@Override
Expand All @@ -205,9 +210,7 @@ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String

@Override
public Set<Relationship> getRelationships() {
return Collections.unmodifiableSet(
new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE, REL_RETRY))
);
return RELATIONSHIPS;
}

@Override
Expand Down
34 changes: 19 additions & 15 deletions ...cessors/src/main/java/org/apache/nifi/processors/gcp/pubsub/lite/ConsumeGCPubSubLite.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,9 +52,7 @@
import org.apache.nifi.processors.gcp.pubsub.AbstractGCPubSubProcessor;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -116,9 +114,28 @@ public class ConsumeGCPubSubLite extends AbstractGCPubSubProcessor implements Ve
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();

private static final List<PropertyDescriptor> DESCRIPTORS = List.of(
GCP_CREDENTIALS_PROVIDER_SERVICE,
SUBSCRIPTION,
BYTES_OUTSTANDING,
MESSAGES_OUTSTANDING
);

public static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS);

private Subscriber subscriber = null;
private BlockingQueue<Message> messages = new LinkedBlockingQueue<>();

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return DESCRIPTORS;
}

@Override
public Set<Relationship> getRelationships() {
return RELATIONSHIPS;
}

@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
final Collection<ValidationResult> results = new ArrayList<ValidationResult>(1);
Expand Down Expand Up @@ -168,19 +185,6 @@ public void onStopped() {
}
}

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return Collections.unmodifiableList(Arrays.asList(SUBSCRIPTION,
GCP_CREDENTIALS_PROVIDER_SERVICE,
BYTES_OUTSTANDING,
MESSAGES_OUTSTANDING));
}

@Override
public Set<Relationship> getRelationships() {
return Collections.singleton(REL_SUCCESS);
}

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
if (subscriber == null) {
Expand Down
25 changes: 13 additions & 12 deletions ...cessors/src/main/java/org/apache/nifi/processors/gcp/pubsub/lite/PublishGCPubSubLite.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,11 +61,8 @@

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;
Expand Down Expand Up @@ -112,16 +109,22 @@ public class PublishGCPubSubLite extends AbstractGCPubSubProcessor implements Ve
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.build();

private static final List<PropertyDescriptor> DESCRIPTORS = List.of(
GCP_CREDENTIALS_PROVIDER_SERVICE,
TOPIC_NAME,
ORDERING_KEY,
BATCH_SIZE_THRESHOLD,
BATCH_BYTES_THRESHOLD,
BATCH_DELAY_THRESHOLD
);

public static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS, REL_FAILURE);

private Publisher publisher = null;

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return Collections.unmodifiableList(Arrays.asList(TOPIC_NAME,
GCP_CREDENTIALS_PROVIDER_SERVICE,
ORDERING_KEY,
BATCH_SIZE_THRESHOLD,
BATCH_BYTES_THRESHOLD,
BATCH_DELAY_THRESHOLD));
return DESCRIPTORS;
}

@Override
Expand All @@ -138,9 +141,7 @@ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String

@Override
public Set<Relationship> getRelationships() {
return Collections.unmodifiableSet(
new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE))
);
return RELATIONSHIPS;
}

@Override
Expand Down
15 changes: 2 additions & 13 deletions ...processors/src/main/java/org/apache/nifi/processors/gcp/storage/AbstractGCSProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,7 @@

import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand All @@ -62,12 +59,11 @@ public abstract class AbstractGCSProcessor extends AbstractGCPProcessor<Storage,
.description("FlowFiles are routed to this relationship if the Google Cloud Storage operation fails.")
.build();

public static final Set<Relationship> relationships = Collections.unmodifiableSet(
new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE)));
public static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS, REL_FAILURE);

@Override
public Set<Relationship> getRelationships() {
return relationships;
return RELATIONSHIPS;
}

// https://cloud.google.com/storage/docs/request-endpoints#storage-set-client-endpoint-java
Expand All @@ -81,13 +77,6 @@ public Set<Relationship> getRelationships() {
.required(false)
.build();

@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> propertyDescriptors = new ArrayList<>(super.getSupportedPropertyDescriptors());
propertyDescriptors.add(STORAGE_API_URL);
return Collections.unmodifiableList(propertyDescriptors);
}

@Override
public List<ConfigVerificationResult> verify(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes) {
final List<ConfigVerificationResult> results = new ArrayList<>(verifyCloudService(context, verificationLogger, attributes));
Expand Down
Loading

0 comments on commit 8eb013a

Please sign in to comment.