NIFI-15210 Fixed verify method of GCS processors and other GCP fixes

turcsanyip · pvillard31 · commit 77e51cb7fa61 · 2025-11-12T21:22:29.000+01:00
Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com> This closes #10519.
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/ConsumeGCPubSub.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/ConsumeGCPubSub.java
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/PublishGCPubSub.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/pubsub/PublishGCPubSub.java
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/AbstractGCSProcessor.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/AbstractGCSProcessor.java
@@ -43,6 +43,8 @@
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.BUCKET_DESC;
+
 /**
  * Base class for creating processors which connect to Google Cloud Storage.
  *
@@ -68,6 +70,14 @@ public Set<Relationship> getRelationships() {
         return RELATIONSHIPS;
     }
 
+    public static final PropertyDescriptor BUCKET = new PropertyDescriptor.Builder()
+            .name("Bucket")
+            .description(BUCKET_DESC)
+            .required(true)
+            .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+            .build();
+
     // https://cloud.google.com/storage/docs/request-endpoints#storage-set-client-endpoint-java
     public static final PropertyDescriptor STORAGE_API_URL = new PropertyDescriptor.Builder()
             .name("Storage API URL")
@@ -137,7 +147,7 @@ protected void customValidate(ValidationContext validationContext, Collection<Va
     protected abstract List<String> getRequiredPermissions();
 
     protected String getBucketName(final ProcessContext context, final Map<String, String> attributes) {
-        return context.getProperty("gcs-bucket").evaluateAttributeExpressions(attributes).getValue();
+        return context.getProperty(BUCKET).evaluateAttributeExpressions(attributes).getValue();
     }
 
     @Override
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObject.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObject.java
@@ -37,7 +37,6 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.nifi.processors.gcp.storage.StorageAttributes.BUCKET_ATTR;
-import static org.apache.nifi.processors.gcp.storage.StorageAttributes.BUCKET_DESC;
 import static org.apache.nifi.processors.gcp.storage.StorageAttributes.KEY_DESC;
 
 
@@ -49,12 +48,9 @@
 @InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
 public class DeleteGCSObject extends AbstractGCSProcessor {
     public static final PropertyDescriptor BUCKET = new PropertyDescriptor.Builder()
-            .name("Bucket")
-            .description(BUCKET_DESC)
-            .required(true)
+            .fromPropertyDescriptor(AbstractGCSProcessor.BUCKET)
             .defaultValue("${" + BUCKET_ATTR + "}")
             .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
-            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
             .build();
 
     public static final PropertyDescriptor KEY = new PropertyDescriptor.Builder()
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/FetchGCSObject.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/FetchGCSObject.java
@@ -169,12 +169,9 @@
 )
 public class FetchGCSObject extends AbstractGCSProcessor {
     public static final PropertyDescriptor BUCKET = new PropertyDescriptor.Builder()
-            .name("Bucket")
-            .description(BUCKET_DESC)
-            .required(true)
+            .fromPropertyDescriptor(AbstractGCSProcessor.BUCKET)
             .defaultValue("${" + BUCKET_ATTR + "}")
             .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
-            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
             .build();
 
     public static final PropertyDescriptor KEY = new PropertyDescriptor.Builder()
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/ListGCSBucket.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/ListGCSBucket.java
@@ -221,14 +221,6 @@ public class ListGCSBucket extends AbstractGCSProcessor {
         .required(true)
         .build();
 
-    public static final PropertyDescriptor BUCKET = new PropertyDescriptor.Builder()
-            .name("Bucket")
-            .description(BUCKET_DESC)
-            .required(true)
-            .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
-            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
-            .build();
-
     public static final PropertyDescriptor PREFIX = new PropertyDescriptor.Builder()
             .name("Prefix")
             .description("The prefix used to filter the object list. In most cases, it should end with a forward slash ('/').")
@@ -409,11 +401,6 @@ protected List<String> getRequiredPermissions() {
         return Collections.singletonList("storage.objects.list");
     }
 
-    @Override
-    protected String getBucketName(final ProcessContext context, final Map<String, String> attributes) {
-        return context.getProperty(BUCKET).evaluateAttributeExpressions().getValue();
-    }
-
     @Override
     public List<ConfigVerificationResult> verify(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes) {
         final List<ConfigVerificationResult> results = new ArrayList<>(super.verify(context, verificationLogger, attributes));
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/PutGCSObject.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/main/java/org/apache/nifi/processors/gcp/storage/PutGCSObject.java
@@ -148,12 +148,9 @@
 })
 public class PutGCSObject extends AbstractGCSProcessor {
     public static final PropertyDescriptor BUCKET = new PropertyDescriptor.Builder()
-        .name("Bucket")
-        .description(BUCKET_DESC)
-        .required(true)
+        .fromPropertyDescriptor(AbstractGCSProcessor.BUCKET)
         .defaultValue("${" + BUCKET_ATTR + "}")
         .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
-        .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .build();
 
     public static final PropertyDescriptor KEY = new PropertyDescriptor.Builder()
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/drive/AbstractGoogleDriveIT.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/drive/AbstractGoogleDriveIT.java
@@ -27,6 +27,7 @@
 import com.google.api.services.drive.model.File;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.gcp.credentials.factory.AuthenticationStrategy;
 import org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors;
 import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService;
 import org.apache.nifi.processors.gcp.util.GoogleUtils;
@@ -106,6 +107,7 @@ protected TestRunner createTestRunner() throws Exception {
         GCPCredentialsControllerService gcpCredentialsControllerService = new GCPCredentialsControllerService();
         testRunner.addControllerService("gcp_credentials_provider_service", gcpCredentialsControllerService);
 
+        testRunner.setProperty(gcpCredentialsControllerService, CredentialPropertyDescriptors.AUTHENTICATION_STRATEGY, AuthenticationStrategy.SERVICE_ACCOUNT_JSON_FILE);
         testRunner.setProperty(gcpCredentialsControllerService, CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE, CREDENTIAL_JSON_FILE_PATH);
         testRunner.enableControllerService(gcpCredentialsControllerService);
         testRunner.setProperty(GoogleUtils.GCP_CREDENTIALS_PROVIDER_SERVICE, "gcp_credentials_provider_service");
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/pubsub/AbstractGCPubSubIT.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/pubsub/AbstractGCPubSubIT.java
@@ -1,47 +1,48 @@
-/*
- * 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;
-
-import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService;
-import org.apache.nifi.reporting.InitializationException;
-import org.apache.nifi.util.TestRunner;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class AbstractGCPubSubIT {
-
-    protected static final String PROJECT_ID = "my-gcm-client";
-    protected static final String CONTROLLER_SERVICE = "GCPCredentialsService";
-    protected static TestRunner runner;
-
-    protected TestRunner setCredentialsCS(TestRunner runner) throws InitializationException {
-        final String serviceAccountJsonFilePath = "path/to/credentials/json";
-        final Map<String, String> propertiesMap = new HashMap<>();
-        final GCPCredentialsControllerService credentialsControllerService = new GCPCredentialsControllerService();
-
-        propertiesMap.put("application-default-credentials", "false");
-        propertiesMap.put("compute-engine-credentials", "false");
-        propertiesMap.put("service-account-json-file", serviceAccountJsonFilePath);
-
-        runner.addControllerService(CONTROLLER_SERVICE, credentialsControllerService, propertiesMap);
-        runner.enableControllerService(credentialsControllerService);
-        runner.assertValid(credentialsControllerService);
-
-        return runner;
-    }
-}
+/*
+ * 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;
+
+import org.apache.nifi.processors.gcp.credentials.factory.AuthenticationStrategy;
+import org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors;
+import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class AbstractGCPubSubIT {
+
+    protected static final String PROJECT_ID = "my-gcm-client";
+    protected static final String CONTROLLER_SERVICE = "GCPCredentialsService";
+    protected static TestRunner runner;
+
+    protected TestRunner setCredentialsCS(TestRunner runner) throws InitializationException {
+        final String serviceAccountJsonFilePath = "path/to/credentials/json";
+        final Map<String, String> propertiesMap = new HashMap<>();
+        final GCPCredentialsControllerService credentialsControllerService = new GCPCredentialsControllerService();
+
+        propertiesMap.put(CredentialPropertyDescriptors.AUTHENTICATION_STRATEGY.getName(), AuthenticationStrategy.SERVICE_ACCOUNT_JSON_FILE.getValue());
+        propertiesMap.put(CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE.getName(), serviceAccountJsonFilePath);
+
+        runner.addControllerService(CONTROLLER_SERVICE, credentialsControllerService, propertiesMap);
+        runner.enableControllerService(credentialsControllerService);
+        runner.assertValid(credentialsControllerService);
+
+        return runner;
+    }
+}
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSIT.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSIT.java
@@ -24,6 +24,8 @@
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.gcp.credentials.factory.AuthenticationStrategy;
+import org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors;
 import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService;
 import org.apache.nifi.util.TestRunner;
 import org.apache.nifi.util.TestRunners;
@@ -94,6 +96,7 @@ protected static TestRunner buildNewRunner(Processor processor) throws Exception
         final GCPCredentialsControllerService credentialsControllerService = new GCPCredentialsControllerService();
         final TestRunner runner = TestRunners.newTestRunner(processor);
         runner.addControllerService("gcpCredentialsControllerService", credentialsControllerService);
+        runner.setProperty(credentialsControllerService, CredentialPropertyDescriptors.AUTHENTICATION_STRATEGY, AuthenticationStrategy.APPLICATION_DEFAULT);
         runner.enableControllerService(credentialsControllerService);
 
         runner.setProperty(AbstractGCSProcessor.GCP_CREDENTIALS_PROVIDER_SERVICE, "gcpCredentialsControllerService");
diff --git a/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectIT.java b/nifi-extension-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectIT.java
@@ -87,9 +87,9 @@ public void testSimpleFetchEncrypted() throws Exception {
         final Map<String, String> attributes = new HashMap<>();
         attributes.put("filename", KEY);
         final List<ConfigVerificationResult> results = processor.verify(runner.getProcessContext(), runner.getLogger(), attributes);
-        assertEquals(2, results.size());
+        assertEquals(3, results.size());
         assertEquals(ConfigVerificationResult.Outcome.SUCCESSFUL, results.get(1).getOutcome());
-        assertTrue(results.get(1).getExplanation().matches("Successfully fetched \\[delete-me\\] from Bucket \\[gcloud-test-bucket-temp-.*\\], totaling 3 bytes"));
+        assertTrue(results.get(2).getExplanation().matches("Successfully fetched \\[delete-me\\] from Bucket \\[gcloud-test-bucket-temp-.*\\], totaling 3 bytes"));
 
         runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS, 1);
         final List<MockFlowFile> ffs = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS);
