diff --git a/PROCESSORS.md b/PROCESSORS.md
index 88a0742..cbd1ac2 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -16,6 +16,7 @@
 - [DefragmentText](#defragmenttext)
 - [DeleteAzureBlobStorage](#deleteazureblobstorage)
 - [DeleteAzureDataLakeStorage](#deleteazuredatalakestorage)
+- [DeleteGCSObject](#deletegcsobject)
 - [DeleteS3Object](#deletes3object)
 - [ExecuteProcess](#executeprocess)
 - [ExecutePythonProcessor](#executepythonprocessor)
@@ -24,6 +25,7 @@
 - [ExtractText](#extracttext)
 - [FetchAzureBlobStorage](#fetchazureblobstorage)
 - [FetchAzureDataLakeStorage](#fetchazuredatalakestorage)
+- [FetchGCSObject](#fetchgcsobject)
 - [FetchFile](#fetchfile)
 - [FetchOPCProcessor](#fetchopcprocessor)
 - [FetchS3Object](#fetchs3object)
@@ -38,6 +40,7 @@
 - [InvokeHTTP](#invokehttp)
 - [ListAzureBlobStorage](#listazureblobstorage)
 - [ListAzureDataLakeStorage](#listazuredatalakestorage)
+- [ListGCSBucket](#listgcsbucket)
 - [ListenHTTP](#listenhttp)
 - [ListenSyslog](#listensyslog)
 - [ListFile](#listfile)
@@ -393,6 +396,42 @@
 |failure|If file deletion from Azure Data Lake storage fails the flowfile is transferred to this relationship|
 |success|If file deletion from Azure Data Lake storage succeeds the flowfile is transferred to this relationship|
 
+## DeleteGCSObject
+
+### Description
+
+Deletes an object from a Google Cloud Bucket.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name                                 | Default Value | Allowable Values                                                                  | Description                                                                                                                                     |
+|--------------------------------------|---------------|-----------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Bucket**                           | ${gcs.bucket} |                                                                                   | Bucket of the object.<br>**Supports Expression Language: true**                                                                                 |
+| **Key**                              | ${filename}   |                                                                                   | Name of the object.<br>**Supports Expression Language: true**                                                                                   |
+| **Number of retries**                | 6             | integers                                                                          | How many retry attempts should be made before routing to the failure relationship.                                                              |
+| **GCP Credentials Provider Service** |               | [GCPCredentialsControllerService](CONTROLLERS.md#GCPCredentialsControllerService) | The Controller Service used to obtain Google Cloud Platform credentials.                                                                        |
+| Server Side Encryption Key           |               |                                                                                   | An AES256 Encryption Key (encoded in base64) for server-side encryption of the object.<br>**Supports Expression Language: true**                |
+| Object Generation                    |               |                                                                                   | The generation of the Object to download. If left empty, then it will download the latest generation.<br>**Supports Expression Language: true** |
+| Endpoint Override URL                |               |                                                                                   | Overrides the default Google Cloud Storage endpoints                                                                                            |
+
+
+### Relationships
+
+| Name    | Description                                                                                  |
+|---------|----------------------------------------------------------------------------------------------|
+| success | FlowFiles are routed to this relationship after a successful Google Cloud Storage operation. |
+| failure | FlowFiles are routed to this relationship if the Google Cloud Storage operation fails.       |
+
+
+### Output Attributes
+
+| Attribute            | Relationship | Description                                             |
+|----------------------|--------------|---------------------------------------------------------|
+| _gcs.status.message_ | failure      | The status message received from google cloud.          |
+| _gcs.error.reason_   | failure      | The description of the error occurred during operation. |
+| _gcs.error.domain_   | failure      | The domain of the error occurred during operation.      |
 
 ## DeleteS3Object
 
@@ -603,6 +642,43 @@
 |success|Files that have been successfully fetched from Azure storage are transferred to this relationship|
 
 
+## FetchGCSObject
+
+### Description
+
+Fetches a file from a Google Cloud Bucket. Designed to be used in tandem with ListGCSBucket.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name                                 | Default Value | Allowable Values                                                                  | Description                                                                                                                                     |
+|--------------------------------------|---------------|-----------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Bucket**                           | ${gcs.bucket} |                                                                                   | Bucket of the object.<br>**Supports Expression Language: true**                                                                                 |
+| **Key**                              | ${filename}   |                                                                                   | Name of the object.<br>**Supports Expression Language: true**                                                                                   |
+| **Number of retries**                | 6             | integers                                                                          | How many retry attempts should be made before routing to the failure relationship.                                                              |
+| **GCP Credentials Provider Service** |               | [GCPCredentialsControllerService](CONTROLLERS.md#GCPCredentialsControllerService) | The Controller Service used to obtain Google Cloud Platform credentials.                                                                        |
+| Server Side Encryption Key           |               |                                                                                   | An AES256 Encryption Key (encoded in base64) for server-side encryption of the object.<br>**Supports Expression Language: true**                |
+| Object Generation                    |               |                                                                                   | The generation of the Object to download. If left empty, then it will download the latest generation.<br>**Supports Expression Language: true** |
+| Endpoint Override URL                |               |                                                                                   | Overrides the default Google Cloud Storage endpoints                                                                                            |
+
+
+### Relationships
+
+| Name    | Description                                                                                  |
+|---------|----------------------------------------------------------------------------------------------|
+| success | FlowFiles are routed to this relationship after a successful Google Cloud Storage operation. |
+| failure | FlowFiles are routed to this relationship if the Google Cloud Storage operation fails.       |
+
+
+### Output Attributes
+
+| Attribute            | Relationship | Description                                             |
+|----------------------|--------------|---------------------------------------------------------|
+| _gcs.status.message_ | failure      | The status message received from google cloud.          |
+| _gcs.error.reason_   | failure      | The description of the error occurred during operation. |
+| _gcs.error.domain_   | failure      | The domain of the error occurred during operation.      |
+
 ## FetchFile
 
 ### Description
@@ -1003,6 +1079,59 @@
 |success|All FlowFiles that are received are routed to success|
 
 
+## ListGCSBucket
+
+### Description
+
+Retrieves a listing of objects from an GCS bucket.
+For each object that is listed, creates a FlowFile that represents the object so that it can be fetched or deleted in conjunction with FetchGCSObject/DeleteGCSObject.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name                                 | Default Value | Allowable Values                                                                  | Description                                                                        |
+|--------------------------------------|---------------|-----------------------------------------------------------------------------------|------------------------------------------------------------------------------------|
+| **Bucket**                           |               |                                                                                   | Bucket of the object.<br>**Supports Expression Language: true**                    |
+| **Number of retries**                | 6             | integers                                                                          | How many retry attempts should be made before routing to the failure relationship. |
+| **GCP Credentials Provider Service** |               | [GCPCredentialsControllerService](CONTROLLERS.md#GCPCredentialsControllerService) | The Controller Service used to obtain Google Cloud Platform credentials.           |
+| Endpoint Override URL                |               |                                                                                   | Overrides the default Google Cloud Storage endpoints                               |
+| List all versions                    | false         | false<br>true                                                                     | Set this option to `true` to get all the previous versions separately.             |
+
+
+### Relationships
+
+| Name    | Description                                                                                   |
+|---------|-----------------------------------------------------------------------------------------------|
+| success | FlowFiles are routed to this relationship after a successful Google Cloud Storage operation.  |
+
+### Output Attributes
+
+| Attribute                  | Relationship | Description                                                        |
+|----------------------------|--------------|--------------------------------------------------------------------|
+| _gcs.bucket_               | success      | Bucket of the object.                                              |
+| _gcs.key, filename_        | success      | Name of the object.                                                |
+| _gcs.size_                 | success      | Size of the object.                                                |
+| _gcs.crc32c_               | success      | The CRC32C checksum of object's data, encoded in base64            |
+| _gcs.md5_                  | success      | The MD5 hash of the object's data encoded in base64.               |
+| _gcs.owner.entity_         | success      | The owner entity, in the form "user-emailAddress".                 |
+| _gcs.owner.entity.id_      | success      | The ID for the entity.                                             |
+| _gcs.content.encoding_     | success      | The content encoding of the object.                                |
+| _gcs.content.language_     | success      | The content language of the object.                                |
+| _gcs.content.disposition_  | success      | The data content disposition of the object.                        |
+| _gcs.media.link_           | success      | The media download link to the object.                             |
+| _gcs.self.link_            | success      | The link to this object.                                           |
+| _gcs.etag_                 | success      | The HTTP 1.1 Entity tag for the object.                            |
+| _gcs.generated.id_         | success      | The service-generated ID for the object                            |
+| _gcs.generation_           | success      | The content generation of this object. Used for object versioning. |
+| _gcs.metageneration_       | success      | The metageneration of the object.                                  |
+| _gcs.create.time_          | success      | Unix timestamp of the object's creation in milliseconds            |
+| _gcs.update.time_          | success      | Unix timestamp of the object's last modification in milliseconds   |
+| _gcs.delete.time_          | success      | Unix timestamp of the object's deletion in milliseconds            |
+| _gcs.encryption.algorithm_ | success      | The algorithm used to encrypt the object.                          |
+| _gcs.encryption.sha256_    | success      | The SHA256 hash of the key used to encrypt the object              |
+
+
 ## ListenHTTP
 
 ### Description
@@ -1614,7 +1743,7 @@
 |--------------------------------------|---------------|-----------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | **Bucket**                           | ${gcs.bucket} |                                                                                                           | Bucket of the object.<br>**Supports Expression Language: true**                                                                                                                                                                                           |
 | **Key**                              | ${filename}   |                                                                                                           | Name of the object.<br>**Supports Expression Language: true**                                                                                                                                                                                             |
-| **Number Of retries**                | 6             | integers                                                                                                  | How many retry attempts should be made before routing to the failure relationship.                                                                                                                                                                        |
+| **Number of retries**                | 6             | integers                                                                                                  | How many retry attempts should be made before routing to the failure relationship.                                                                                                                                                                        |
 | **GCP Credentials Provider Service** |               | [GCPCredentialsControllerService](CONTROLLERS.md#GCPCredentialsControllerService)                         | The Controller Service used to obtain Google Cloud Platform credentials.                                                                                                                                                                                  |
 | Object ACL                           |               | authenticatedRead<br>bucketOwnerFullControl<br>bucketOwnerRead<br>private<br>projectPrivate<br>publicRead | Access Control to be attached to the object uploaded. Not providing this will revert to bucket defaults. For more information please visit [Google Cloud Access control lists](https://cloud.google.com/storage/docs/access-control/lists#predefined-acl) |
 | Server Side Encryption Key           |               |                                                                                                           | An AES256 Encryption Key (encoded in base64) for server-side encryption of the object.<br>**Supports Expression Language: true**                                                                                                                          |
@@ -1634,6 +1763,7 @@
 
 | Attribute                  | Relationship | Description                                                        |
 |----------------------------|--------------|--------------------------------------------------------------------|
+| _gcs.status.message_       | failure      | The status message received from google cloud.                     |
 | _gcs.error.reason_         | failure      | The description of the error occurred during upload.               |
 | _gcs.error.domain_         | failure      | The domain of the error occurred during upload.                    |
 | _gcs.bucket_               | success      | Bucket of the object.                                              |
@@ -1652,9 +1782,9 @@
 | _gcs.generated.id_         | success      | The service-generated ID for the object                            |
 | _gcs.generation_           | success      | The content generation of this object. Used for object versioning. |
 | _gcs.metageneration_       | success      | The metageneration of the object.                                  |
-| _gcs.create.time_          | success      | The creation time of the object (milliseconds)                     |
-| _gcs.update.time_          | success      | The last modification time of the object (milliseconds)            |
-| _gcs.delete.time_          | success      | The deletion time of the object (milliseconds)                     |
+| _gcs.create.time_          | success      | Unix timestamp of the object's creation in milliseconds            |
+| _gcs.update.time_          | success      | Unix timestamp of the object's last modification in milliseconds   |
+| _gcs.delete.time_          | success      | Unix timestamp of the object's deletion in milliseconds            |
 | _gcs.encryption.algorithm_ | success      | The algorithm used to encrypt the object.                          |
 | _gcs.encryption.sha256_    | success      | The SHA256 hash of the key used to encrypt the object              |
 
diff --git a/README.md b/README.md
index 26af00c..2ffb244 100644
--- a/README.md
+++ b/README.md
@@ -80,7 +80,7 @@
 | CivetWeb                         | [ListenHTTP](PROCESSORS.md#listenhttp)                                                                                                                                                                                                                                                                                                                                                                                                                                                             | -DDISABLE_CIVET=ON     |
 | CURL                             | [InvokeHTTP](PROCESSORS.md#invokehttp)                                                                                                                                                                                                                                                                                                                                                                                                                                                             | -DDISABLE_CURL=ON      |
 | GPS                              | GetGPS                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | -DENABLE_GPS=ON        |
-| Google Cloud Platform            | [GcpCredentialsControllerService](CONTROLLERS.md#GcpCredentialsControllerService)<br>[PutGCSObject](PROCESSORS.md#putgcsobject)                                                                                                                                                                                                                                                                                                                                                                    | -DENABLE_GCP=ON        |
+| Google Cloud Platform            | [DeleteGCSObject](PROCESSORS.md#deletegcsobject)<br>[FetchGCSObject](PROCESSORS.md#fetchgcsobject)<br>[GCPCredentialsControllerService](CONTROLLERS.md#GCPCredentialsControllerService)<br>[ListGCSBucket](PROCESSORS.md#listgcsbucket)<br>[PutGCSObject](PROCESSORS.md#putgcsobject)                                                                                                                                                                                                              | -DENABLE_GCP=ON        |
 | Kafka                            | [PublishKafka](PROCESSORS.md#publishkafka)                                                                                                                                                                                                                                                                                                                                                                                                                                                         | -DENABLE_LIBRDKAFKA=ON |
 | Kubernetes                       | [KubernetesControllerService](CONTROLLERS.md#kubernetesControllerService)                                                                                                                                                                                                                                                                                                                                                                                                                          | -DENABLE_KUBERNETES=ON |
 | JNI                              | **NiFi Processors**                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | -DENABLE_JNI=ON        |
diff --git a/docker/test/integration/features/google_cloud_storage.feature b/docker/test/integration/features/google_cloud_storage.feature
index 18af9b1..ad2b6f1 100644
--- a/docker/test/integration/features/google_cloud_storage.feature
+++ b/docker/test/integration/features/google_cloud_storage.feature
@@ -17,3 +17,31 @@
 
     Then a flowfile with the content "hello_gcs" is placed in the monitored directory in less than 45 seconds
     And an object with the content "hello_gcs" is present in the Google Cloud storage
+
+  Scenario: A MiNiFi instance can fetch the listed objects from Google Cloud storage bucket
+    Given a Google Cloud storage server is set up and a single object with contents "preloaded data" is present
+    And a ListGCSBucket processor
+    And a FetchGCSObject processor
+    And the ListGCSBucket and the FetchGCSObject processors are set up with a GCPCredentialsControllerService to communicate with the Google Cloud storage server
+    And a PutFile processor with the "Directory" property set to "/tmp/output"
+    And the "success" relationship of the ListGCSBucket processor is connected to the FetchGCSObject
+    And the "success" relationship of the FetchGCSObject processor is connected to the PutFile
+
+    When all instances start up
+
+    Then a flowfile with the content "preloaded data" is placed in the monitored directory in less than 10 seconds
+
+
+  Scenario: A MiNiFi instance can delete the listed objects from Google Cloud storage bucket
+    Given a Google Cloud storage server is set up with some test data
+    And a ListGCSBucket processor
+    And a DeleteGCSObject processor
+    And the ListGCSBucket and the DeleteGCSObject processors are set up with a GCPCredentialsControllerService to communicate with the Google Cloud storage server
+    And a PutFile processor with the "Directory" property set to "/tmp/output"
+    And the "success" relationship of the ListGCSBucket processor is connected to the DeleteGCSObject
+    And the "success" relationship of the DeleteGCSObject processor is connected to the PutFile
+
+    When all instances start up
+
+    Then the test bucket of Google Cloud Storage is empty
+    And at least one empty flowfile is placed in the monitored directory in less than 10 seconds
diff --git a/docker/test/integration/minifi/processors/DeleteGCSObject.py b/docker/test/integration/minifi/processors/DeleteGCSObject.py
new file mode 100644
index 0000000..f816c6f
--- /dev/null
+++ b/docker/test/integration/minifi/processors/DeleteGCSObject.py
@@ -0,0 +1,14 @@
+from ..core.Processor import Processor
+
+
+class DeleteGCSObject(Processor):
+    def __init__(
+            self):
+        super(DeleteGCSObject, self).__init__(
+            'DeleteGCSObject',
+            properties={
+                'Bucket': 'test-bucket',
+                'Endpoint Override URL': 'fake-gcs-server:4443',
+                'Number of retries': 2
+            },
+            auto_terminate=["success", "failure"])
diff --git a/docker/test/integration/minifi/processors/FetchGCSObject.py b/docker/test/integration/minifi/processors/FetchGCSObject.py
new file mode 100644
index 0000000..557e565
--- /dev/null
+++ b/docker/test/integration/minifi/processors/FetchGCSObject.py
@@ -0,0 +1,14 @@
+from ..core.Processor import Processor
+
+
+class FetchGCSObject(Processor):
+    def __init__(
+            self):
+        super(FetchGCSObject, self).__init__(
+            'FetchGCSObject',
+            properties={
+                'Bucket': 'test-bucket',
+                'Endpoint Override URL': 'fake-gcs-server:4443',
+                'Number of retries': 2
+            },
+            auto_terminate=["success", "failure"])
diff --git a/docker/test/integration/minifi/processors/ListGCSBucket.py b/docker/test/integration/minifi/processors/ListGCSBucket.py
new file mode 100644
index 0000000..5c66254
--- /dev/null
+++ b/docker/test/integration/minifi/processors/ListGCSBucket.py
@@ -0,0 +1,14 @@
+from ..core.Processor import Processor
+
+
+class ListGCSBucket(Processor):
+    def __init__(
+            self):
+        super(ListGCSBucket, self).__init__(
+            'ListGCSBucket',
+            properties={
+                'Bucket': 'test-bucket',
+                'Endpoint Override URL': 'fake-gcs-server:4443',
+                'Number of retries': 2
+            },
+            auto_terminate=["success"])
diff --git a/docker/test/integration/steps/steps.py b/docker/test/integration/steps/steps.py
index f588c8c..8769ed6 100644
--- a/docker/test/integration/steps/steps.py
+++ b/docker/test/integration/steps/steps.py
@@ -403,8 +403,9 @@
 
 
 # google cloud storage setup
-@given("a Google Cloud storage server is set up with some test data")
 @given("a Google Cloud storage server is set up")
+@given("a Google Cloud storage server is set up with some test data")
+@given('a Google Cloud storage server is set up and a single object with contents "preloaded data" is present')
 def step_impl(context):
     context.test.acquire_container("fake-gcs-server", "fake-gcs-server")
 
@@ -445,6 +446,17 @@
     p1.set_property("GCP Credentials Provider Service", gcp_controller_service.name)
 
 
+@given(u'the {processor_one} and the {processor_two} processors are set up with a GCPCredentialsControllerService to communicate with the Google Cloud storage server')
+def step_impl(context, processor_one, processor_two):
+    gcp_controller_service = GCPCredentialsControllerService(credentials_location="Use Anonymous credentials")
+    p1 = context.test.get_node_by_name(processor_one)
+    p2 = context.test.get_node_by_name(processor_two)
+    p1.controller_services.append(gcp_controller_service)
+    p1.set_property("GCP Credentials Provider Service", gcp_controller_service.name)
+    p2.controller_services.append(gcp_controller_service)
+    p2.set_property("GCP Credentials Provider Service", gcp_controller_service.name)
+
+
 @given("the kafka broker is started")
 def step_impl(context):
     context.test.start_kafka_broker()
diff --git a/extensions/gcp/GCPAttributes.h b/extensions/gcp/GCPAttributes.h
index 92548fd..38af0d3 100644
--- a/extensions/gcp/GCPAttributes.h
+++ b/extensions/gcp/GCPAttributes.h
@@ -24,6 +24,7 @@
 
 constexpr const char* GCS_ERROR_REASON = "gcs.error.reason";
 constexpr const char* GCS_ERROR_DOMAIN = "gcs.error.domain";
+constexpr const char* GCS_STATUS_MESSAGE = "gcs.status.message";
 constexpr const char* GCS_BUCKET_ATTR = "gcs.bucket";
 constexpr const char* GCS_OBJECT_NAME_ATTR = "gcs.key";
 constexpr const char* GCS_SIZE_ATTR = "gcs.size";
diff --git a/extensions/gcp/processors/DeleteGCSObject.cpp b/extensions/gcp/processors/DeleteGCSObject.cpp
new file mode 100644
index 0000000..e95178d
--- /dev/null
+++ b/extensions/gcp/processors/DeleteGCSObject.cpp
@@ -0,0 +1,121 @@
+/**
+ * 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.
+ */
+
+#include "DeleteGCSObject.h"
+
+#include "core/Resource.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "core/FlowFile.h"
+#include "../GCPAttributes.h"
+
+namespace gcs = ::google::cloud::storage;
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+const core::Property DeleteGCSObject::Bucket(
+    core::PropertyBuilder::createProperty("Bucket")
+        ->withDescription("Bucket of the object.")
+        ->withDefaultValue("${gcs.bucket}")
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property DeleteGCSObject::Key(
+    core::PropertyBuilder::createProperty("Key")
+        ->withDescription("Name of the object.")
+        ->withDefaultValue("${filename}")
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property DeleteGCSObject::ObjectGeneration(
+    core::PropertyBuilder::createProperty("Object Generation")
+        ->withDescription("The generation of the Object to download. If left empty, then it will download the latest generation.")
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property DeleteGCSObject::EncryptionKey(
+    core::PropertyBuilder::createProperty("Server Side Encryption Key")
+        ->withDescription("The AES256 Encryption Key (encoded in base64) for server-side decryption of the object.")
+        ->isRequired(false)
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Relationship DeleteGCSObject::Success("success", "FlowFiles are routed to this relationship after a successful Google Cloud Storage operation.");
+const core::Relationship DeleteGCSObject::Failure("failure", "FlowFiles are routed to this relationship if the Google Cloud Storage operation fails.");
+
+void DeleteGCSObject::initialize() {
+  setSupportedProperties({GCPCredentials,
+                          Bucket,
+                          Key,
+                          ObjectGeneration,
+                          NumberOfRetries,
+                          EncryptionKey,
+                          EndpointOverrideURL});
+  setSupportedRelationships({Success, Failure});
+}
+
+void DeleteGCSObject::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) {
+  gsl_Expects(context && session && gcp_credentials_);
+
+  auto flow_file = session->get();
+  if (!flow_file) {
+    context->yield();
+    return;
+  }
+
+  auto bucket = context->getProperty(Bucket, flow_file);
+  if (!bucket || bucket->empty()) {
+    logger_->log_error("Missing bucket name");
+    session->transfer(flow_file, Failure);
+    return;
+  }
+  auto object_name = context->getProperty(Key, flow_file);
+  if (!object_name || object_name->empty()) {
+    logger_->log_error("Missing object name");
+    session->transfer(flow_file, Failure);
+    return;
+  }
+
+  gcs::Generation generation;
+
+  if (auto gen_str = context->getProperty(ObjectGeneration, flow_file); gen_str && !gen_str->empty()) {
+    try {
+      uint64_t gen;
+      utils::internal::ValueParser(*gen_str).parse(gen).parseEnd();
+      generation = gcs::Generation(gen);
+    } catch (const utils::internal::ValueException&) {
+      logger_->log_error("Invalid generation: %s", *gen_str);
+      session->transfer(flow_file, Failure);
+      return;
+    }
+  }
+
+  auto status = getClient().DeleteObject(*bucket, *object_name, generation, gcs::IfGenerationNotMatch(0));
+
+  if (!status.ok()) {
+    flow_file->setAttribute(GCS_STATUS_MESSAGE, status.message());
+    flow_file->setAttribute(GCS_ERROR_REASON, status.error_info().reason());
+    flow_file->setAttribute(GCS_ERROR_DOMAIN, status.error_info().domain());
+    logger_->log_error("Failed to delete %s object from %s bucket on Google Cloud Storage %s %s", *object_name, *bucket, status.message(), status.error_info().reason());
+    session->transfer(flow_file, Failure);
+    return;
+  }
+
+  session->transfer(flow_file, Success);
+}
+
+REGISTER_RESOURCE(DeleteGCSObject, "Deletes an object from a Google Cloud Bucket.");
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/DeleteGCSObject.h b/extensions/gcp/processors/DeleteGCSObject.h
new file mode 100644
index 0000000..22644c9
--- /dev/null
+++ b/extensions/gcp/processors/DeleteGCSObject.h
@@ -0,0 +1,55 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <string>
+#include <memory>
+
+#include "GCSProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+
+class DeleteGCSObject : public GCSProcessor {
+ public:
+  explicit DeleteGCSObject(const std::string& name, const utils::Identifier& uuid = {})
+      : GCSProcessor(name, uuid, core::logging::LoggerFactory<DeleteGCSObject>::getLogger()) {
+  }
+  ~DeleteGCSObject() override = default;
+
+  EXTENSIONAPI static const core::Property Bucket;
+  EXTENSIONAPI static const core::Property Key;
+  EXTENSIONAPI static const core::Property EncryptionKey;
+  EXTENSIONAPI static const core::Property ObjectGeneration;
+
+  EXTENSIONAPI static const core::Relationship Success;
+  EXTENSIONAPI static const core::Relationship Failure;
+
+  void initialize() override;
+  void onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) override;
+
+  core::annotation::Input getInputRequirement() const override {
+    return core::annotation::Input::INPUT_REQUIRED;
+  }
+
+  bool isSingleThreaded() const override {
+    return true;
+  }
+};
+
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/FetchGCSObject.cpp b/extensions/gcp/processors/FetchGCSObject.cpp
new file mode 100644
index 0000000..58e36df
--- /dev/null
+++ b/extensions/gcp/processors/FetchGCSObject.cpp
@@ -0,0 +1,197 @@
+/**
+ * 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.
+ */
+
+#include "FetchGCSObject.h"
+
+#include <utility>
+
+#include "core/Resource.h"
+#include "core/FlowFile.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "../GCPAttributes.h"
+
+namespace gcs = ::google::cloud::storage;
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+const core::Property FetchGCSObject::Bucket(
+    core::PropertyBuilder::createProperty("Bucket")
+        ->withDescription("Bucket of the object.")
+        ->withDefaultValue("${gcs.bucket}")
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property FetchGCSObject::Key(
+    core::PropertyBuilder::createProperty("Key")
+        ->withDescription("Name of the object.")
+        ->withDefaultValue("${filename}")
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property FetchGCSObject::ObjectGeneration(
+    core::PropertyBuilder::createProperty("Object Generation")
+        ->withDescription("The generation of the Object to download. If left empty, then it will download the latest generation.")
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property FetchGCSObject::EncryptionKey(
+    core::PropertyBuilder::createProperty("Server Side Encryption Key")
+        ->withDescription("The AES256 Encryption Key (encoded in base64) for server-side decryption of the object.")
+        ->isRequired(false)
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Relationship FetchGCSObject::Success("success", "FlowFiles are routed to this relationship after a successful Google Cloud Storage operation.");
+const core::Relationship FetchGCSObject::Failure("failure", "FlowFiles are routed to this relationship if the Google Cloud Storage operation fails.");
+
+
+namespace {
+class FetchFromGCSCallback {
+ public:
+  FetchFromGCSCallback(gcs::Client& client, std::string bucket, std::string key)
+      : bucket_(std::move(bucket)),
+        key_(std::move(key)),
+        client_(client) {
+  }
+
+  int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+    auto reader = client_.ReadObject(bucket_, key_, encryption_key_, generation_, gcs::IfGenerationNotMatch(0));
+    auto set_members = gsl::finally([&]{
+      status_ = reader.status();
+      result_generation_ = reader.generation();
+      meta_generation_ = reader.metageneration();
+      storage_class_ = reader.storage_class();
+    });
+    if (!reader)
+      return 0;
+    std::string contents{std::istreambuf_iterator<char>{reader}, {}};
+    auto write_ret = stream->write(gsl::make_span(contents).as_span<std::byte>());
+    reader.Close();
+    return write_ret;
+  }
+
+  [[nodiscard]] auto getStatus() const noexcept { return status_; }
+  [[nodiscard]] auto getGeneration() const noexcept { return result_generation_; }
+  [[nodiscard]] auto getMetaGeneration() const noexcept { return meta_generation_; }
+  [[nodiscard]] auto getStorageClass() const noexcept { return storage_class_; }
+
+
+  void setEncryptionKey(const gcs::EncryptionKey& encryption_key) {
+    encryption_key_ = encryption_key;
+  }
+
+  void setGeneration(const gcs::Generation generation) {
+    generation_ = generation;
+  }
+
+ private:
+  std::string bucket_;
+  std::string key_;
+  gcs::Client& client_;
+
+  gcs::EncryptionKey encryption_key_;
+  gcs::Generation generation_;
+
+  google::cloud::Status status_;
+  std::optional<std::int64_t> result_generation_;
+  std::optional<std::int64_t> meta_generation_;
+  std::optional<std::string> storage_class_;
+};
+}  // namespace
+
+
+void FetchGCSObject::initialize() {
+  setSupportedProperties({GCPCredentials,
+                          Bucket,
+                          Key,
+                          ObjectGeneration,
+                          NumberOfRetries,
+                          EncryptionKey,
+                          EndpointOverrideURL});
+  setSupportedRelationships({Success, Failure});
+}
+
+void FetchGCSObject::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& session_factory) {
+  GCSProcessor::onSchedule(context, session_factory);
+  gsl_Expects(context);
+  if (auto encryption_key = context->getProperty(EncryptionKey)) {
+    try {
+      encryption_key_ = gcs::EncryptionKey::FromBase64Key(*encryption_key);
+    } catch (const google::cloud::RuntimeStatusError&) {
+      throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Could not decode the base64-encoded encryption key from property " + EncryptionKey.getName());    }
+  }
+}
+
+void FetchGCSObject::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) {
+  gsl_Expects(context && session && gcp_credentials_);
+
+  auto flow_file = session->get();
+  if (!flow_file) {
+    context->yield();
+    return;
+  }
+
+  auto bucket = context->getProperty(Bucket, flow_file);
+  if (!bucket || bucket->empty()) {
+    logger_->log_error("Missing bucket name");
+    session->transfer(flow_file, Failure);
+    return;
+  }
+  auto object_name = context->getProperty(Key, flow_file);
+  if (!object_name || object_name->empty()) {
+    logger_->log_error("Missing object name");
+    session->transfer(flow_file, Failure);
+    return;
+  }
+
+  gcs::Client client = getClient();
+  FetchFromGCSCallback callback(client, *bucket, *object_name);
+  callback.setEncryptionKey(encryption_key_);
+
+  if (auto gen_str = context->getProperty(ObjectGeneration, flow_file); gen_str && !gen_str->empty()) {
+    try {
+      uint64_t gen;
+      utils::internal::ValueParser(*gen_str).parse(gen).parseEnd();
+      callback.setGeneration(gcs::Generation(gen));
+    } catch (const utils::internal::ValueException&) {
+      logger_->log_error("Invalid generation: %s", *gen_str);
+      session->transfer(flow_file, Failure);
+      return;
+    }
+  }
+
+  session->write(flow_file, std::ref(callback));
+  if (!callback.getStatus().ok()) {
+    flow_file->setAttribute(GCS_STATUS_MESSAGE, callback.getStatus().message());
+    flow_file->setAttribute(GCS_ERROR_REASON, callback.getStatus().error_info().reason());
+    flow_file->setAttribute(GCS_ERROR_DOMAIN, callback.getStatus().error_info().domain());
+    logger_->log_error("Failed to fetch from Google Cloud Storage %s %s", callback.getStatus().message(), callback.getStatus().error_info().reason());
+    session->transfer(flow_file, Failure);
+    return;
+  }
+
+  if (auto generation = callback.getGeneration())
+    flow_file->setAttribute(GCS_GENERATION, std::to_string(*generation));
+  if (auto meta_generation = callback.getMetaGeneration())
+    flow_file->setAttribute(GCS_META_GENERATION, std::to_string(*meta_generation));
+  if (auto storage_class = callback.getStorageClass())
+    flow_file->setAttribute(GCS_STORAGE_CLASS, *storage_class);
+  session->transfer(flow_file, Success);
+}
+
+REGISTER_RESOURCE(FetchGCSObject, "Fetches a file from a Google Cloud Bucket. Designed to be used in tandem with ListGCSBucket.");
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/FetchGCSObject.h b/extensions/gcp/processors/FetchGCSObject.h
new file mode 100644
index 0000000..d71e747
--- /dev/null
+++ b/extensions/gcp/processors/FetchGCSObject.h
@@ -0,0 +1,60 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <string>
+#include <memory>
+
+#include "GCSProcessor.h"
+#include "google/cloud/storage/well_known_headers.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+
+class FetchGCSObject : public GCSProcessor {
+ public:
+  explicit FetchGCSObject(const std::string& name, const utils::Identifier& uuid = {})
+      : GCSProcessor(name, uuid, core::logging::LoggerFactory<FetchGCSObject>::getLogger()) {
+  }
+  ~FetchGCSObject() override = default;
+
+  EXTENSIONAPI static const core::Property Bucket;
+  EXTENSIONAPI static const core::Property Key;
+  EXTENSIONAPI static const core::Property EncryptionKey;
+  EXTENSIONAPI static const core::Property ObjectGeneration;
+
+  EXTENSIONAPI static const core::Relationship Success;
+  EXTENSIONAPI static const core::Relationship Failure;
+
+  void initialize() override;
+  void onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) override;
+  void onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) override;
+
+  core::annotation::Input getInputRequirement() const override {
+    return core::annotation::Input::INPUT_REQUIRED;
+  }
+
+  bool isSingleThreaded() const override {
+    return true;
+  }
+
+ private:
+  google::cloud::storage::EncryptionKey encryption_key_;
+};
+
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/GCSProcessor.cpp b/extensions/gcp/processors/GCSProcessor.cpp
new file mode 100644
index 0000000..445f295
--- /dev/null
+++ b/extensions/gcp/processors/GCSProcessor.cpp
@@ -0,0 +1,86 @@
+/**
+ * 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.
+ */
+
+#include "GCSProcessor.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "../controllerservices/GCPCredentialsControllerService.h"
+
+namespace gcs = ::google::cloud::storage;
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+
+const core::Property GCSProcessor::GCPCredentials(
+    core::PropertyBuilder::createProperty("GCP Credentials Provider Service")
+        ->withDescription("The Controller Service used to obtain Google Cloud Platform credentials.")
+        ->isRequired(true)
+        ->asType<GCPCredentialsControllerService>()
+        ->build());
+
+const core::Property GCSProcessor::NumberOfRetries(
+    core::PropertyBuilder::createProperty("Number of retries")
+        ->withDescription("How many retry attempts should be made before routing to the failure relationship.")
+        ->withDefaultValue<uint64_t>(6)
+        ->isRequired(true)
+        ->supportsExpressionLanguage(false)
+        ->build());
+
+const core::Property GCSProcessor::EndpointOverrideURL(
+    core::PropertyBuilder::createProperty("Endpoint Override URL")
+        ->withDescription("Overrides the default Google Cloud Storage endpoints")
+        ->isRequired(false)
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+namespace {
+std::shared_ptr<google::cloud::storage::oauth2::Credentials> getCredentials(core::ProcessContext& context) {
+  std::string service_name;
+  if (context.getProperty(GCSProcessor::GCPCredentials.getName(), service_name) && !IsNullOrEmpty(service_name)) {
+    auto gcp_credentials_controller_service = std::dynamic_pointer_cast<const GCPCredentialsControllerService>(context.getControllerService(service_name));
+    if (!gcp_credentials_controller_service)
+      return nullptr;
+    return gcp_credentials_controller_service->getCredentials();
+  }
+  return nullptr;
+}
+}  // namespace
+
+
+void GCSProcessor::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>&) {
+  gsl_Expects(context);
+  if (auto number_of_retries = context->getProperty<uint64_t>(NumberOfRetries)) {
+    retry_policy_ = std::make_shared<google::cloud::storage::LimitedErrorCountRetryPolicy>(*number_of_retries);
+  }
+
+  gcp_credentials_ = getCredentials(*context);
+  if (!gcp_credentials_) {
+    throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing GCP Credentials");
+  }
+
+  endpoint_url_ = context->getProperty(EndpointOverrideURL);
+  if (endpoint_url_)
+    logger_->log_debug("Endpoint overwritten: %s", *endpoint_url_);
+}
+
+gcs::Client GCSProcessor::getClient() const {
+  auto options = gcs::ClientOptions(gcp_credentials_);
+  if (endpoint_url_)
+    options.set_endpoint(*endpoint_url_);
+  return gcs::Client(options, *retry_policy_);
+}
+
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/GCSProcessor.h b/extensions/gcp/processors/GCSProcessor.h
new file mode 100644
index 0000000..f673a3c
--- /dev/null
+++ b/extensions/gcp/processors/GCSProcessor.h
@@ -0,0 +1,55 @@
+/**
+ * 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.
+ */
+
+
+#pragma once
+
+#include <string>
+#include <memory>
+#include <utility>
+#include <optional>
+
+#include "core/logging/Logger.h"
+#include "core/Processor.h"
+#include "google/cloud/storage/oauth2/credentials.h"
+#include "google/cloud/storage/client.h"
+#include "google/cloud/storage/retry_policy.h"
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+class GCSProcessor : public core::Processor {
+ public:
+  GCSProcessor(const std::string& name, const minifi::utils::Identifier& uuid, std::shared_ptr<core::logging::Logger> logger)
+      : core::Processor(name, uuid),
+        logger_(std::move(logger)) {
+  }
+
+  EXTENSIONAPI static const core::Property GCPCredentials;
+  EXTENSIONAPI static const core::Property NumberOfRetries;
+  EXTENSIONAPI static const core::Property EndpointOverrideURL;
+
+  void onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) override;
+
+ protected:
+  virtual google::cloud::storage::Client getClient() const;
+
+  std::optional<std::string> endpoint_url_;
+  std::shared_ptr<google::cloud::storage::oauth2::Credentials> gcp_credentials_;
+  google::cloud::storage::RetryPolicyOption::Type retry_policy_ = std::make_shared<google::cloud::storage::LimitedErrorCountRetryPolicy>(6);
+  std::shared_ptr<core::logging::Logger> logger_;
+};
+
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/ListGCSBucket.cpp b/extensions/gcp/processors/ListGCSBucket.cpp
new file mode 100644
index 0000000..be6dc22
--- /dev/null
+++ b/extensions/gcp/processors/ListGCSBucket.cpp
@@ -0,0 +1,81 @@
+/**
+ * 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.
+ */
+
+#include "ListGCSBucket.h"
+
+#include "core/Resource.h"
+#include "core/FlowFile.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "../GCPAttributes.h"
+
+namespace gcs = ::google::cloud::storage;
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+const core::Property ListGCSBucket::Bucket(
+    core::PropertyBuilder::createProperty("Bucket")
+        ->withDescription("Bucket of the object.")
+        ->isRequired(true)
+        ->supportsExpressionLanguage(true)
+        ->build());
+
+const core::Property ListGCSBucket::ListAllVersions(
+    core::PropertyBuilder::createProperty("List all versions")
+        ->withDescription("Set this option to `true` to get all the previous versions separately.")
+        ->withDefaultValue<bool>(false)
+        ->build());
+
+const core::Relationship ListGCSBucket::Success("success", "FlowFiles are routed to this relationship after a successful Google Cloud Storage operation.");
+
+void ListGCSBucket::initialize() {
+  setSupportedProperties({GCPCredentials,
+                          Bucket,
+                          NumberOfRetries,
+                          EndpointOverrideURL,
+                          ListAllVersions});
+  setSupportedRelationships({Success});
+}
+
+
+void ListGCSBucket::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& session_factory) {
+  GCSProcessor::onSchedule(context, session_factory);
+  gsl_Expects(context);
+  context->getProperty(Bucket.getName(), bucket_);
+}
+
+void ListGCSBucket::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) {
+  gsl_Expects(context && session && gcp_credentials_);
+
+  gcs::Client client = getClient();
+  auto list_all_versions = context->getProperty<bool>(ListAllVersions);
+  gcs::Versions versions = (list_all_versions && *list_all_versions) ? gcs::Versions(true) : gcs::Versions(false);
+  auto objects_in_bucket = client.ListObjects(bucket_, versions);
+  for (const auto& object_in_bucket : objects_in_bucket) {
+    if (object_in_bucket.ok()) {
+      auto flow_file = session->create();
+      flow_file->updateAttribute(core::SpecialFlowAttribute::FILENAME, object_in_bucket->name());
+      setAttributesFromObjectMetadata(*flow_file, *object_in_bucket);
+      session->transfer(flow_file, Success);
+    } else {
+      logger_->log_error("Invalid object in bucket %s", bucket_);
+    }
+  }
+}
+
+REGISTER_RESOURCE(ListGCSBucket, "Retrieves a listing of objects from an GCS bucket. "
+                                 "For each object that is listed, creates a FlowFile that represents the object so that it can be fetched in conjunction with FetchGCSObject.");
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/ListGCSBucket.h b/extensions/gcp/processors/ListGCSBucket.h
new file mode 100644
index 0000000..bdf8555
--- /dev/null
+++ b/extensions/gcp/processors/ListGCSBucket.h
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <string>
+#include <memory>
+
+#include "GCSProcessor.h"
+#include "core/logging/LoggerConfiguration.h"
+
+namespace org::apache::nifi::minifi::extensions::gcp {
+
+class ListGCSBucket : public GCSProcessor {
+ public:
+  explicit ListGCSBucket(const std::string& name, const utils::Identifier& uuid = {})
+      : GCSProcessor(name, uuid, core::logging::LoggerFactory<ListGCSBucket>::getLogger()) {
+  }
+  ~ListGCSBucket() override = default;
+
+  EXTENSIONAPI static const core::Property Bucket;
+  EXTENSIONAPI static const core::Property ListAllVersions;
+
+  EXTENSIONAPI static const core::Relationship Success;
+
+  void initialize() override;
+  void onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) override;
+  void onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) override;
+
+  core::annotation::Input getInputRequirement() const override {
+    return core::annotation::Input::INPUT_FORBIDDEN;
+  }
+
+  bool isSingleThreaded() const override {
+    return true;
+  }
+
+ private:
+  std::string bucket_;
+};
+
+}  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/PutGCSObject.cpp b/extensions/gcp/processors/PutGCSObject.cpp
index 00a3d0c..eb5f499 100644
--- a/extensions/gcp/processors/PutGCSObject.cpp
+++ b/extensions/gcp/processors/PutGCSObject.cpp
@@ -17,27 +17,17 @@
 
 #include "PutGCSObject.h"
 
-#include <vector>
 #include <utility>
 
 #include "core/Resource.h"
 #include "core/FlowFile.h"
 #include "core/ProcessContext.h"
 #include "core/ProcessSession.h"
-#include "io/StreamPipe.h"
-#include "utils/OptionalUtils.h"
 #include "../GCPAttributes.h"
 
 namespace gcs = ::google::cloud::storage;
 
 namespace org::apache::nifi::minifi::extensions::gcp {
-const core::Property PutGCSObject::GCPCredentials(
-    core::PropertyBuilder::createProperty("GCP Credentials Provider Service")
-        ->withDescription("The Controller Service used to obtain Google Cloud Platform credentials.")
-        ->isRequired(true)
-        ->asType<GCPCredentialsControllerService>()
-        ->build());
-
 const core::Property PutGCSObject::Bucket(
     core::PropertyBuilder::createProperty("Bucket")
         ->withDescription("Bucket of the object.")
@@ -52,14 +42,6 @@
         ->supportsExpressionLanguage(true)
         ->build());
 
-const core::Property PutGCSObject::NumberOfRetries(
-    core::PropertyBuilder::createProperty("Number of retries")
-        ->withDescription("How many retry attempts should be made before routing to the failure relationship.")
-        ->withDefaultValue<uint64_t>(6)
-        ->isRequired(true)
-        ->supportsExpressionLanguage(false)
-        ->build());
-
 const core::Property PutGCSObject::ContentType(
     core::PropertyBuilder::createProperty("Content Type")
         ->withDescription("Content Type for the file, i.e. text/plain ")
@@ -102,17 +84,9 @@
         ->withDefaultValue<bool>(true)
         ->build());
 
-const core::Property PutGCSObject::EndpointOverrideURL(
-    core::PropertyBuilder::createProperty("Endpoint Override URL")
-        ->withDescription("Overrides the default Google Cloud Storage endpoints")
-        ->isRequired(false)
-        ->supportsExpressionLanguage(true)
-        ->build());
-
 const core::Relationship PutGCSObject::Success("success", "Files that have been successfully written to Google Cloud Storage are transferred to this relationship");
 const core::Relationship PutGCSObject::Failure("failure", "Files that could not be written to Google Cloud Storage for some reason are transferred to this relationship");
 
-
 namespace {
 class UploadToGCSCallback {
  public:
@@ -183,16 +157,6 @@
   google::cloud::StatusOr<gcs::ObjectMetadata> result_;
 };
 
-std::shared_ptr<google::cloud::storage::oauth2::Credentials> getCredentials(core::ProcessContext& context) {
-  std::string service_name;
-  if (context.getProperty(PutGCSObject::GCPCredentials.getName(), service_name) && !IsNullOrEmpty(service_name)) {
-    auto gcp_credentials_controller_service = std::dynamic_pointer_cast<const GCPCredentialsControllerService>(context.getControllerService(service_name));
-    if (!gcp_credentials_controller_service)
-      return nullptr;
-    return gcp_credentials_controller_service->getCredentials();
-  }
-  return nullptr;
-}
 }  // namespace
 
 
@@ -211,16 +175,10 @@
   setSupportedRelationships({Success, Failure});
 }
 
-gcs::Client PutGCSObject::getClient(const gcs::ClientOptions& options) const {
-  return gcs::Client(options, *retry_policy_);
-}
 
-
-void PutGCSObject::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>&) {
+void PutGCSObject::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& session_factory) {
+  GCSProcessor::onSchedule(context, session_factory);
   gsl_Expects(context);
-  if (auto number_of_retries = context->getProperty<uint64_t>(NumberOfRetries)) {
-    retry_policy_ = std::make_shared<google::cloud::storage::LimitedErrorCountRetryPolicy>(*number_of_retries);
-  }
   if (auto encryption_key = context->getProperty(EncryptionKey)) {
     try {
       encryption_key_ = gcs::EncryptionKey::FromBase64Key(*encryption_key);
@@ -228,10 +186,6 @@
       throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Could not decode the base64-encoded encryption key from property " + EncryptionKey.getName());
     }
   }
-  gcp_credentials_ = getCredentials(*context);
-  if (!gcp_credentials_) {
-    throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing GCP Credentials");
-  }
 }
 
 void PutGCSObject::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) {
@@ -256,13 +210,7 @@
     return;
   }
 
-  auto options = gcs::ClientOptions(gcp_credentials_);
-  if (auto endpoint_override_url = context->getProperty(EndpointOverrideURL)) {
-    options.set_endpoint(*endpoint_override_url);
-    logger_->log_debug("Endpoint override url %s", *endpoint_override_url);
-  }
-
-  gcs::Client client = getClient(options);
+  gcs::Client client = getClient();
   UploadToGCSCallback callback(client, *bucket, *object_name);
 
   if (auto crc32_checksum = context->getProperty(Crc32cChecksum, flow_file)) {
@@ -286,6 +234,7 @@
   session->read(flow_file, std::ref(callback));
   auto& result = callback.getResult();
   if (!result.ok()) {
+    flow_file->setAttribute(GCS_STATUS_MESSAGE, result.status().message());
     flow_file->setAttribute(GCS_ERROR_REASON, result.status().error_info().reason());
     flow_file->setAttribute(GCS_ERROR_DOMAIN, result.status().error_info().domain());
     logger_->log_error("Failed to upload to Google Cloud Storage %s %s", result.status().message(), result.status().error_info().reason());
diff --git a/extensions/gcp/processors/PutGCSObject.h b/extensions/gcp/processors/PutGCSObject.h
index 4539622..d9cd680 100644
--- a/extensions/gcp/processors/PutGCSObject.h
+++ b/extensions/gcp/processors/PutGCSObject.h
@@ -20,17 +20,14 @@
 #include <string>
 #include <memory>
 
-#include "core/Processor.h"
-#include "core/logging/Logger.h"
+#include "GCSProcessor.h"
 #include "core/logging/LoggerConfiguration.h"
-#include "../controllerservices/GCPCredentialsControllerService.h"
-#include "google/cloud/storage/client.h"
-#include "google/cloud/storage/retry_policy.h"
 #include "utils/Enum.h"
+#include "google/cloud/storage/well_known_headers.h"
 
 namespace org::apache::nifi::minifi::extensions::gcp {
 
-class PutGCSObject : public core::Processor {
+class PutGCSObject : public GCSProcessor {
  public:
   SMART_ENUM(PredefinedAcl,
              (AUTHENTICATED_READ, "authenticatedRead"),
@@ -42,25 +39,18 @@
              (PUBLIC_READ_WRITE, "publicReadWrite"));
 
   explicit PutGCSObject(const std::string& name, const utils::Identifier& uuid = {})
-      : core::Processor(name, uuid) {
+      : GCSProcessor(name, uuid, core::logging::LoggerFactory<PutGCSObject>::getLogger()) {
   }
-  PutGCSObject(const PutGCSObject&) = delete;
-  PutGCSObject(PutGCSObject&&) = delete;
-  PutGCSObject& operator=(const PutGCSObject&) = delete;
-  PutGCSObject& operator=(PutGCSObject&&) = delete;
   ~PutGCSObject() override = default;
 
-  EXTENSIONAPI static const core::Property GCPCredentials;
   EXTENSIONAPI static const core::Property Bucket;
   EXTENSIONAPI static const core::Property Key;
-  EXTENSIONAPI static const core::Property NumberOfRetries;
   EXTENSIONAPI static const core::Property ContentType;
   EXTENSIONAPI static const core::Property MD5Hash;
   EXTENSIONAPI static const core::Property Crc32cChecksum;
   EXTENSIONAPI static const core::Property EncryptionKey;
   EXTENSIONAPI static const core::Property ObjectACL;
   EXTENSIONAPI static const core::Property OverwriteObject;
-  EXTENSIONAPI static const core::Property EndpointOverrideURL;
 
   EXTENSIONAPI static const core::Relationship Success;
   EXTENSIONAPI static const core::Relationship Failure;
@@ -78,15 +68,7 @@
   }
 
  private:
-  virtual google::cloud::storage::Client getClient(const google::cloud::storage::ClientOptions& options) const;
-
   google::cloud::storage::EncryptionKey encryption_key_;
-
-  std::shared_ptr<google::cloud::storage::oauth2::Credentials> gcp_credentials_;
-  std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<PutGCSObject>::getLogger();
-
- protected:
-  google::cloud::storage::RetryPolicyOption::Type retry_policy_ = std::make_shared<google::cloud::storage::LimitedErrorCountRetryPolicy>(6);
 };
 
 }  // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/tests/DeleteGCSObjectTests.cpp b/extensions/gcp/tests/DeleteGCSObjectTests.cpp
new file mode 100644
index 0000000..2b0e446
--- /dev/null
+++ b/extensions/gcp/tests/DeleteGCSObjectTests.cpp
@@ -0,0 +1,137 @@
+/**
+ * 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.
+ */
+#include "../processors/DeleteGCSObject.h"
+#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "GCPAttributes.h"
+#include "core/Resource.h"
+#include "SingleProcessorTestController.h"
+#include "google/cloud/storage/testing/mock_client.h"
+#include "google/cloud/storage/internal/object_metadata_parser.h"
+#include "google/cloud/storage/testing/canonical_errors.h"
+
+namespace gcs = ::google::cloud::storage;
+namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+
+using DeleteGCSObject = org::apache::nifi::minifi::extensions::gcp::DeleteGCSObject;
+using GCPCredentialsControllerService = org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
+using DeleteObjectRequest = gcs::internal::DeleteObjectRequest;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
+using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+
+namespace {
+class DeleteGCSObjectMocked : public DeleteGCSObject {
+  using org::apache::nifi::minifi::extensions::gcp::DeleteGCSObject::DeleteGCSObject;
+ public:
+  gcs::Client getClient() const override {
+    return gcs::testing::ClientFromMock(mock_client_, *retry_policy_);
+  }
+  std::shared_ptr<gcs::testing::MockClient> mock_client_ = std::make_shared<gcs::testing::MockClient>();
+};
+REGISTER_RESOURCE(DeleteGCSObjectMocked, "DeleteGCSObjectMocked");
+}  // namespace
+
+class DeleteGCSObjectTests : public ::testing::Test {
+ public:
+  void SetUp() override {
+    gcp_credentials_node_ = test_controller_.plan->addController("GCPCredentialsControllerService", "gcp_credentials_controller_service");
+    test_controller_.plan->setProperty(gcp_credentials_node_,
+                                       GCPCredentialsControllerService::CredentialsLoc.getName(),
+                                       toString(GCPCredentialsControllerService::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
+    test_controller_.plan->setProperty(delete_gcs_object_,
+                                       DeleteGCSObject::GCPCredentials.getName(),
+                                       "gcp_credentials_controller_service");
+  }
+  std::shared_ptr<DeleteGCSObjectMocked> delete_gcs_object_ = std::make_shared<DeleteGCSObjectMocked>("DeleteGCSObjectMocked");
+  org::apache::nifi::minifi::test::SingleProcessorTestController test_controller_{delete_gcs_object_};
+  std::shared_ptr<minifi::core::controller::ControllerServiceNode>  gcp_credentials_node_;
+};
+
+TEST_F(DeleteGCSObjectTests, MissingBucket) {
+  EXPECT_CALL(*delete_gcs_object_->mock_client_, CreateResumableSession).Times(0);
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::Bucket.getName(), ""));
+  const auto& result = test_controller_.trigger("hello world");
+  EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size());
+  ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size());
+  EXPECT_EQ(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
+  EXPECT_EQ(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON));
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0]));
+}
+
+TEST_F(DeleteGCSObjectTests, ServerGivesPermaError) {
+  EXPECT_CALL(*delete_gcs_object_->mock_client_, DeleteObject)
+      .WillOnce(testing::Return(PermanentError()));
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::Bucket.getName(), "bucket-from-property"));
+  const auto& result = test_controller_.trigger("hello world");
+  EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size());
+  ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size());
+  EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
+  EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON));
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0]));
+}
+
+TEST_F(DeleteGCSObjectTests, ServerGivesTransientErrors) {
+  EXPECT_CALL(*delete_gcs_object_->mock_client_, DeleteObject)
+      .WillOnce(testing::Return(TransientError()))
+      .WillOnce(testing::Return(TransientError()));
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::NumberOfRetries.getName(), "1"));
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::Bucket.getName(), "bucket-from-property"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}});
+  EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size());
+  ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size());
+  EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
+  EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON));
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0]));
+}
+
+
+TEST_F(DeleteGCSObjectTests, HandlingSuccessfullDeletion) {
+  EXPECT_CALL(*delete_gcs_object_->mock_client_, DeleteObject)
+      .WillOnce([](DeleteObjectRequest const& request) {
+        EXPECT_EQ("bucket-from-attribute", request.bucket_name());
+        EXPECT_TRUE(request.HasOption<gcs::Generation>());
+        EXPECT_TRUE(request.GetOption<gcs::Generation>().has_value());
+        EXPECT_EQ(23, request.GetOption<gcs::Generation>().value());
+        return google::cloud::make_status_or(gcs::internal::EmptyResponse{});
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::ObjectGeneration.getName(), "${gcs.generation}"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}, {minifi_gcp::GCS_GENERATION, "23"}});
+  ASSERT_EQ(1, result.at(DeleteGCSObject::Success).size());
+  EXPECT_EQ(0, result.at(DeleteGCSObject::Failure).size());
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Success)[0]));
+}
+
+TEST_F(DeleteGCSObjectTests, EmptyGeneration) {
+  EXPECT_CALL(*delete_gcs_object_->mock_client_, DeleteObject)
+      .WillOnce([](DeleteObjectRequest const& request) {
+        EXPECT_EQ("bucket-from-attribute", request.bucket_name());
+        EXPECT_FALSE(request.HasOption<gcs::Generation>());
+        return google::cloud::make_status_or(gcs::internal::EmptyResponse{});
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::ObjectGeneration.getName(), "${gcs.generation}"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}});
+  ASSERT_EQ(1, result.at(DeleteGCSObject::Success).size());
+  EXPECT_EQ(0, result.at(DeleteGCSObject::Failure).size());
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Success)[0]));
+}
+
+TEST_F(DeleteGCSObjectTests, InvalidGeneration) {
+  EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_, DeleteGCSObject::ObjectGeneration.getName(), "${gcs.generation}"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}, {minifi_gcp::GCS_GENERATION, "23 banana"}});
+  ASSERT_EQ(0, result.at(DeleteGCSObject::Success).size());
+  EXPECT_EQ(1, result.at(DeleteGCSObject::Failure).size());
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0]));
+}
diff --git a/extensions/gcp/tests/FetchGCSObjectTests.cpp b/extensions/gcp/tests/FetchGCSObjectTests.cpp
new file mode 100644
index 0000000..696d3b4
--- /dev/null
+++ b/extensions/gcp/tests/FetchGCSObjectTests.cpp
@@ -0,0 +1,170 @@
+/**
+ * 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.
+ */
+#include "../processors/FetchGCSObject.h"
+#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "GCPAttributes.h"
+#include "core/Resource.h"
+#include "SingleProcessorTestController.h"
+#include "google/cloud/storage/testing/mock_client.h"
+#include "google/cloud/storage/internal/object_metadata_parser.h"
+#include "google/cloud/storage/testing/canonical_errors.h"
+
+namespace gcs = ::google::cloud::storage;
+namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+
+using FetchGCSObject = org::apache::nifi::minifi::extensions::gcp::FetchGCSObject;
+using GCPCredentialsControllerService = org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
+using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+
+namespace {
+class FetchGCSObjectMocked : public FetchGCSObject {
+  using org::apache::nifi::minifi::extensions::gcp::FetchGCSObject::FetchGCSObject;
+ public:
+  gcs::Client getClient() const override {
+    return gcs::testing::ClientFromMock(mock_client_, *retry_policy_);
+  }
+  std::shared_ptr<gcs::testing::MockClient> mock_client_ = std::make_shared<gcs::testing::MockClient>();
+};
+REGISTER_RESOURCE(FetchGCSObjectMocked, "FetchGCSObjectMocked");
+}  // namespace
+
+class FetchGCSObjectTests : public ::testing::Test {
+ public:
+  void SetUp() override {
+    gcp_credentials_node_ = test_controller_.plan->addController("GCPCredentialsControllerService", "gcp_credentials_controller_service");
+    test_controller_.plan->setProperty(gcp_credentials_node_,
+                                       GCPCredentialsControllerService::CredentialsLoc.getName(),
+                                       toString(GCPCredentialsControllerService::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
+    test_controller_.plan->setProperty(fetch_gcs_object_,
+                                       FetchGCSObject::GCPCredentials.getName(),
+                                       "gcp_credentials_controller_service");
+  }
+  std::shared_ptr<FetchGCSObjectMocked> fetch_gcs_object_ = std::make_shared<FetchGCSObjectMocked>("FetchGCSObjectMocked");
+  org::apache::nifi::minifi::test::SingleProcessorTestController test_controller_{fetch_gcs_object_};
+  std::shared_ptr<minifi::core::controller::ControllerServiceNode>  gcp_credentials_node_;
+};
+
+TEST_F(FetchGCSObjectTests, MissingBucket) {
+  EXPECT_CALL(*fetch_gcs_object_->mock_client_, CreateResumableSession).Times(0);
+  EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_, FetchGCSObject::Bucket.getName(), ""));
+  const auto& result = test_controller_.trigger("hello world");
+  EXPECT_EQ(0, result.at(FetchGCSObject::Success).size());
+  ASSERT_EQ(1, result.at(FetchGCSObject::Failure).size());
+  EXPECT_EQ(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
+  EXPECT_EQ(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON));
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(FetchGCSObject::Failure)[0]));
+}
+
+TEST_F(FetchGCSObjectTests, ServerError) {
+  EXPECT_CALL(*fetch_gcs_object_->mock_client_, ReadObject)
+      .WillOnce([](gcs::internal::ReadObjectRangeRequest const& request) {
+        EXPECT_EQ(request.bucket_name(), "bucket-from-property") << request;
+        auto* mock_source = new gcs::testing::MockObjectReadSource;
+        ::testing::InSequence seq;
+        EXPECT_CALL(*mock_source, IsOpen).WillRepeatedly(testing::Return(true));
+        EXPECT_CALL(*mock_source, Read)
+            .WillOnce(testing::Return(google::cloud::Status(
+                google::cloud::StatusCode::kInvalidArgument,
+                "Invalid Argument")));
+        EXPECT_CALL(*mock_source, IsOpen).WillRepeatedly(testing::Return(false));
+
+        std::unique_ptr<gcs::internal::ObjectReadSource> result(mock_source);
+
+        return google::cloud::make_status_or(std::move(result));
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_, FetchGCSObject::Bucket.getName(), "bucket-from-property"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}});
+  EXPECT_EQ(0, result.at(FetchGCSObject::Success).size());
+  ASSERT_EQ(1, result.at(FetchGCSObject::Failure).size());
+  EXPECT_NE(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
+  EXPECT_NE(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON));
+}
+
+TEST_F(FetchGCSObjectTests, HappyPath) {
+  std::string const text = "stored text";
+  std::size_t offset = 0;
+  // Simulate a Read() call in the MockObjectReadSource object created below
+  auto simulate_read = [&text, &offset](void* buf, std::size_t n) {
+    auto const l = (std::min)(n, text.size() - offset);
+    std::memcpy(buf, text.data() + offset, l);
+    offset += l;
+    return gcs::internal::ReadSourceResult{
+        l, gcs::internal::HttpResponse{200, {}, {}}};
+  };
+  EXPECT_CALL(*fetch_gcs_object_->mock_client_, ReadObject)
+      .WillOnce([&](gcs::internal::ReadObjectRangeRequest const& request) {
+        EXPECT_EQ(request.bucket_name(), "bucket-from-attribute") << request;
+        EXPECT_TRUE(request.HasOption<gcs::Generation>());
+        EXPECT_TRUE(request.GetOption<gcs::Generation>().has_value());
+        EXPECT_EQ(23, request.GetOption<gcs::Generation>().value());
+        std::unique_ptr<gcs::testing::MockObjectReadSource> mock_source(new gcs::testing::MockObjectReadSource);
+        ::testing::InSequence seq;
+        EXPECT_CALL(*mock_source, IsOpen()).WillRepeatedly(testing::Return(true));
+        EXPECT_CALL(*mock_source, Read).WillOnce(simulate_read);
+        EXPECT_CALL(*mock_source, IsOpen()).WillRepeatedly(testing::Return(false));
+
+        return google::cloud::make_status_or(
+            std::unique_ptr<gcs::internal::ObjectReadSource>(
+                std::move(mock_source)));
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_, FetchGCSObject::ObjectGeneration.getName(), "${gcs.generation}"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}, {minifi_gcp::GCS_GENERATION, "23"}});
+  ASSERT_EQ(1, result.at(FetchGCSObject::Success).size());
+  EXPECT_EQ(0, result.at(FetchGCSObject::Failure).size());
+  EXPECT_EQ("stored text", test_controller_.plan->getContent(result.at(FetchGCSObject::Success)[0]));
+}
+
+TEST_F(FetchGCSObjectTests, EmptyGeneration) {
+  std::string const text = "stored text";
+  std::size_t offset = 0;
+  // Simulate a Read() call in the MockObjectReadSource object created below
+  auto simulate_read = [&text, &offset](void* buf, std::size_t n) {
+    auto const l = (std::min)(n, text.size() - offset);
+    std::memcpy(buf, text.data() + offset, l);
+    offset += l;
+    return gcs::internal::ReadSourceResult{
+        l, gcs::internal::HttpResponse{200, {}, {}}};
+  };
+  EXPECT_CALL(*fetch_gcs_object_->mock_client_, ReadObject)
+      .WillOnce([&](gcs::internal::ReadObjectRangeRequest const& request) {
+        EXPECT_EQ(request.bucket_name(), "bucket-from-attribute") << request;
+        EXPECT_FALSE(request.HasOption<gcs::Generation>());
+        std::unique_ptr<gcs::testing::MockObjectReadSource> mock_source(new gcs::testing::MockObjectReadSource);
+        ::testing::InSequence seq;
+        EXPECT_CALL(*mock_source, IsOpen()).WillRepeatedly(testing::Return(true));
+        EXPECT_CALL(*mock_source, Read).WillOnce(simulate_read);
+        EXPECT_CALL(*mock_source, IsOpen()).WillRepeatedly(testing::Return(false));
+
+        return google::cloud::make_status_or(
+            std::unique_ptr<gcs::internal::ObjectReadSource>(
+                std::move(mock_source)));
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_, FetchGCSObject::ObjectGeneration.getName(), "${gcs.generation}"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}});
+  ASSERT_EQ(1, result.at(FetchGCSObject::Success).size());
+  EXPECT_EQ(0, result.at(FetchGCSObject::Failure).size());
+  EXPECT_EQ("stored text", test_controller_.plan->getContent(result.at(FetchGCSObject::Success)[0]));
+}
+
+TEST_F(FetchGCSObjectTests, InvalidGeneration) {
+  EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_, FetchGCSObject::ObjectGeneration.getName(), "${gcs.generation}"));
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}, {minifi_gcp::GCS_GENERATION, "23 banana"}});
+  ASSERT_EQ(0, result.at(FetchGCSObject::Success).size());
+  EXPECT_EQ(1, result.at(FetchGCSObject::Failure).size());
+  EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(FetchGCSObject::Failure)[0]));
+}
diff --git a/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp b/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
index 4eb97cd..f6e3b71 100644
--- a/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
+++ b/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
@@ -27,7 +27,6 @@
 #include "rapidjson/writer.h"
 #include "google/cloud/internal/setenv.h"
 
-
 namespace gcs = ::google::cloud::storage;
 
 using GCPCredentialsControllerService = org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
diff --git a/extensions/gcp/tests/ListGCSBucketTests.cpp b/extensions/gcp/tests/ListGCSBucketTests.cpp
new file mode 100644
index 0000000..a585194
--- /dev/null
+++ b/extensions/gcp/tests/ListGCSBucketTests.cpp
@@ -0,0 +1,147 @@
+/**
+ * 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.
+ */
+#include "../processors/ListGCSBucket.h"
+#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "core/Resource.h"
+#include "SingleProcessorTestController.h"
+#include "google/cloud/storage/testing/mock_client.h"
+#include "google/cloud/storage/internal/object_metadata_parser.h"
+#include "google/cloud/storage/testing/canonical_errors.h"
+
+namespace gcs = ::google::cloud::storage;
+namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+
+using ListGCSBucket = org::apache::nifi::minifi::extensions::gcp::ListGCSBucket;
+using ListObjectsRequest = gcs::internal::ListObjectsRequest;
+using ListObjectsResponse = gcs::internal::ListObjectsResponse;
+using GCPCredentialsControllerService = org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
+using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+
+namespace {
+class ListGCSBucketMocked : public ListGCSBucket {
+  using org::apache::nifi::minifi::extensions::gcp::ListGCSBucket::ListGCSBucket;
+ public:
+  gcs::Client getClient() const override {
+    return gcs::testing::ClientFromMock(mock_client_, *retry_policy_);
+  }
+  std::shared_ptr<gcs::testing::MockClient> mock_client_ = std::make_shared<gcs::testing::MockClient>();
+};
+REGISTER_RESOURCE(ListGCSBucketMocked, "ListGCSBucketMocked");
+
+auto CreateObject(int index, int generation = 1) {
+  std::string id = "object-" + std::to_string(index);
+  std::string name = id;
+  std::string link =
+      "https://storage.googleapis.com/storage/v1/b/test-bucket/" + id + "#1";
+  nlohmann::json metadata{
+      {"bucket", "test-bucket"},
+      {"id", id},
+      {"name", name},
+      {"selfLink", link},
+      {"generation", generation},
+      {"kind", "storage#object"},
+  };
+  return google::cloud::storage::internal::ObjectMetadataParser::FromJson(metadata).value();
+}
+}  // namespace
+
+class ListGCSBucketTests : public ::testing::Test {
+ public:
+  void SetUp() override {
+    gcp_credentials_node_ = test_controller_.plan->addController("GCPCredentialsControllerService", "gcp_credentials_controller_service");
+    test_controller_.plan->setProperty(gcp_credentials_node_,
+                                       GCPCredentialsControllerService::CredentialsLoc.getName(),
+                                       toString(GCPCredentialsControllerService::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
+    test_controller_.plan->setProperty(list_gcs_bucket_,
+                                       ListGCSBucket::GCPCredentials.getName(),
+                                       "gcp_credentials_controller_service");
+  }
+  std::shared_ptr<ListGCSBucketMocked> list_gcs_bucket_ = std::make_shared<ListGCSBucketMocked>("ListGCSBucketMocked");
+  org::apache::nifi::minifi::test::SingleProcessorTestController test_controller_{list_gcs_bucket_};
+  std::shared_ptr<minifi::core::controller::ControllerServiceNode>  gcp_credentials_node_;
+};
+
+TEST_F(ListGCSBucketTests, MissingBucket) {
+  EXPECT_CALL(*list_gcs_bucket_->mock_client_, CreateResumableSession).Times(0);
+  EXPECT_THROW(test_controller_.trigger(), utils::internal::RequiredPropertyMissingException);
+}
+
+TEST_F(ListGCSBucketTests, ServerGivesPermaError) {
+  auto return_permanent_error = [](ListObjectsRequest const&) {
+    return google::cloud::StatusOr<ListObjectsResponse>(PermanentError());
+  };
+  EXPECT_CALL(*list_gcs_bucket_->mock_client_, ListObjects)
+      .WillOnce(return_permanent_error);
+  EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_, ListGCSBucket::Bucket.getName(), "bucket-from-property"));
+  const auto& result = test_controller_.trigger();
+  EXPECT_EQ(0, result.at(ListGCSBucket::Success).size());
+}
+
+TEST_F(ListGCSBucketTests, ServerGivesTransientErrors) {
+  auto return_temp_error = [](ListObjectsRequest const&) {
+    return google::cloud::StatusOr<ListObjectsResponse>(TransientError());
+  };
+  EXPECT_CALL(*list_gcs_bucket_->mock_client_, ListObjects)
+      .WillOnce(return_temp_error)
+      .WillOnce(return_temp_error);
+  EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_, ListGCSBucket::NumberOfRetries.getName(), "1"));
+  EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_, ListGCSBucket::Bucket.getName(), "bucket-from-property"));
+  const auto& result = test_controller_.trigger();
+  EXPECT_EQ(0, result.at(ListGCSBucket::Success).size());
+}
+
+TEST_F(ListGCSBucketTests, WithoutVersions) {
+  EXPECT_CALL(*list_gcs_bucket_->mock_client_, ListObjects)
+      .WillOnce([](ListObjectsRequest const& req)
+                    -> google::cloud::StatusOr<ListObjectsResponse> {
+        EXPECT_EQ("bucket-from-property", req.bucket_name());
+        EXPECT_TRUE(req.HasOption<gcs::Versions>());
+        EXPECT_FALSE(req.GetOption<gcs::Versions>().value());
+
+        ListObjectsResponse response;
+        response.items.emplace_back(CreateObject(1, 1));
+        response.items.emplace_back(CreateObject(1, 2));
+        response.items.emplace_back(CreateObject(1, 3));
+        return response;
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_, ListGCSBucket::Bucket.getName(), "bucket-from-property"));
+  const auto& result = test_controller_.trigger();
+  EXPECT_EQ(3, result.at(ListGCSBucket::Success).size());
+}
+
+
+TEST_F(ListGCSBucketTests, WithVersions) {
+  EXPECT_CALL(*list_gcs_bucket_->mock_client_, ListObjects)
+      .WillOnce([](ListObjectsRequest const& req)
+                    -> google::cloud::StatusOr<ListObjectsResponse> {
+        EXPECT_EQ("bucket-from-property", req.bucket_name());
+        EXPECT_TRUE(req.HasOption<gcs::Versions>());
+        EXPECT_TRUE(req.GetOption<gcs::Versions>().value());
+
+        ListObjectsResponse response;
+        response.items.emplace_back(CreateObject(1));
+        response.items.emplace_back(CreateObject(2));
+        response.items.emplace_back(CreateObject(3));
+        return response;
+      });
+  EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_, ListGCSBucket::Bucket.getName(), "bucket-from-property"));
+  EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_, ListGCSBucket::ListAllVersions.getName(), "true"));
+  const auto& result = test_controller_.trigger();
+  EXPECT_EQ(3, result.at(ListGCSBucket::Success).size());
+}
+
diff --git a/extensions/gcp/tests/PutGCSObjectTests.cpp b/extensions/gcp/tests/PutGCSObjectTests.cpp
index 0c2a3ba..a5c8919 100644
--- a/extensions/gcp/tests/PutGCSObjectTests.cpp
+++ b/extensions/gcp/tests/PutGCSObjectTests.cpp
@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 #include "../processors/PutGCSObject.h"
+#include "../controllerservices/GCPCredentialsControllerService.h"
 #include "GCPAttributes.h"
 #include "core/Resource.h"
 #include "SingleProcessorTestController.h"
@@ -39,7 +40,7 @@
 class PutGCSObjectMocked : public PutGCSObject {
   using org::apache::nifi::minifi::extensions::gcp::PutGCSObject::PutGCSObject;
  public:
-  gcs::Client getClient(const gcs::ClientOptions&) const override {
+  gcs::Client getClient() const override {
     return gcs::testing::ClientFromMock(mock_client_, *retry_policy_);
   }
   std::shared_ptr<gcs::testing::MockClient> mock_client_ = std::make_shared<gcs::testing::MockClient>();
@@ -85,8 +86,7 @@
 TEST_F(PutGCSObjectTests, MissingBucket) {
   EXPECT_CALL(*put_gcs_object_->mock_client_, CreateResumableSession).Times(0);
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), ""));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   EXPECT_EQ(0, result.at(PutGCSObject::Success).size());
   ASSERT_EQ(1, result.at(PutGCSObject::Failure).size());
   EXPECT_EQ(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
@@ -107,8 +107,7 @@
         return google::cloud::make_status_or(std::unique_ptr<gcs::internal::ResumableUploadSession>(std::move(mock_upload_session)));
       });
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "${gcs.bucket}"));
-  test_controller_.enqueueFlowFile("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}});
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world", {{minifi_gcp::GCS_BUCKET_ATTR, "bucket-from-attribute"}});
   ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
   EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Success)[0]));
@@ -127,8 +126,7 @@
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::NumberOfRetries.getName(), "2"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   EXPECT_EQ(0, result.at(PutGCSObject::Success).size());
   ASSERT_EQ(1, result.at(PutGCSObject::Failure).size());
   EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
@@ -146,8 +144,7 @@
       .WillOnce(return_permanent_error);
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   EXPECT_EQ(0, result.at(PutGCSObject::Success).size());
   ASSERT_EQ(1, result.at(PutGCSObject::Failure).size());
   EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN));
@@ -171,8 +168,7 @@
       });
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   EXPECT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
 }
@@ -195,8 +191,7 @@
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Crc32cChecksum.getName(), "${crc32c}"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world", {{"crc32c", "yZRlqg=="}, {"md5", "XrY7u+Ae7tCTyyK7j1rNww=="}});
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world", {{"crc32c", "yZRlqg=="}, {"md5", "XrY7u+Ae7tCTyyK7j1rNww=="}});
   EXPECT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
 }
@@ -215,8 +210,7 @@
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::OverwriteObject.getName(), "false"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world", {{"crc32c", "yZRlqg=="}, {"md5", "XrY7u+Ae7tCTyyK7j1rNww=="}});
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world", {{"crc32c", "yZRlqg=="}, {"md5", "XrY7u+Ae7tCTyyK7j1rNww=="}});
   ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
   EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Success)[0]));
@@ -236,8 +230,7 @@
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::EncryptionKey.getName(), "ZW5jcnlwdGlvbl9rZXk="));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
   EXPECT_NE(std::nullopt, result.at(PutGCSObject::Success)[0]->getAttribute(minifi_gcp::GCS_ENCRYPTION_SHA256_ATTR));
@@ -250,8 +243,7 @@
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::EncryptionKey.getName(), "not_base64_key"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world");
-  EXPECT_THROW(test_controller_.trigger(), minifi::Exception);
+  EXPECT_THROW(test_controller_.trigger("hello world"), minifi::Exception);
 }
 
 TEST_F(PutGCSObjectTests, NoContentType) {
@@ -267,8 +259,7 @@
       });
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
   EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Success)[0]));
@@ -288,8 +279,7 @@
       });
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
-  test_controller_.enqueueFlowFile("hello world", {{"mime.type", "text/attribute"}});
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world", {{"mime.type", "text/attribute"}});
   ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
   EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Success)[0]));
@@ -310,8 +300,7 @@
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Bucket.getName(), "bucket-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::Key.getName(), "object-name-from-property"));
   EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_, PutGCSObject::ObjectACL.getName(), toString(PutGCSObject::PredefinedAcl::AUTHENTICATED_READ)));
-  test_controller_.enqueueFlowFile("hello world");
-  const auto& result = test_controller_.trigger();
+  const auto& result = test_controller_.trigger("hello world");
   ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
   EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
   EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Success)[0]));
diff --git a/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp b/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
index 8509144..e5df28f 100644
--- a/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
+++ b/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
@@ -317,8 +317,7 @@
   invokehttp->setProperty(InvokeHTTP::InvalidHTTPHeaderFieldHandlingStrategy, "fail");
   invokehttp->setProperty(InvokeHTTP::AttributesToSend, ".*");
   invokehttp->setAutoTerminatedRelationships({InvokeHTTP::RelNoRetry, InvokeHTTP::Success, InvokeHTTP::RelResponse, InvokeHTTP::RelRetry});
-  test_controller.enqueueFlowFile("data", {{"invalid header", "value"}});
-  const auto result = test_controller.trigger();
+  const auto result = test_controller.trigger("data", {{"invalid header", "value"}});
   auto file_contents = result.at(InvokeHTTP::RelFailure);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller.plan->getContent(file_contents[0]) == "data");
@@ -338,8 +337,7 @@
   invokehttp->setProperty(InvokeHTTP::InvalidHTTPHeaderFieldHandlingStrategy, "fail");
   invokehttp->setProperty(InvokeHTTP::AttributesToSend, "valid.*");
   invokehttp->setAutoTerminatedRelationships({InvokeHTTP::RelNoRetry, InvokeHTTP::Success, InvokeHTTP::RelResponse, InvokeHTTP::RelRetry});
-  test_controller.enqueueFlowFile("data", {{"invalid header", "value"}, {"valid-header", "value2"}});
-  const auto result = test_controller.trigger();
+  const auto result = test_controller.trigger("data", {{"invalid header", "value"}, {"valid-header", "value2"}});
   REQUIRE(result.at(InvokeHTTP::RelFailure).empty());
   const auto& success_contents = result.at(InvokeHTTP::Success);
   REQUIRE(success_contents.size() == 1);
@@ -360,8 +358,7 @@
   invokehttp->setProperty(InvokeHTTP::URL, TestHTTPServer::URL);
   invokehttp->setProperty(InvokeHTTP::AttributesToSend, ".*");
   invokehttp->setAutoTerminatedRelationships({InvokeHTTP::RelNoRetry, InvokeHTTP::RelFailure, InvokeHTTP::RelResponse, InvokeHTTP::RelRetry});
-  test_controller.enqueueFlowFile("data", {{"invalid header", "value"}, {"", "value2"}});
-  const auto result = test_controller.trigger();
+  const auto result = test_controller.trigger("data", {{"invalid header", "value"}, {"", "value2"}});
   auto file_contents = result.at(InvokeHTTP::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller.plan->getContent(file_contents[0]) == "data");
@@ -383,8 +380,7 @@
   invokehttp->setProperty(InvokeHTTP::InvalidHTTPHeaderFieldHandlingStrategy, "drop");
   invokehttp->setProperty(InvokeHTTP::AttributesToSend, ".*");
   invokehttp->setAutoTerminatedRelationships({InvokeHTTP::RelNoRetry, InvokeHTTP::RelFailure, InvokeHTTP::RelResponse, InvokeHTTP::RelRetry});
-  test_controller.enqueueFlowFile("data", {{"legit-header", "value1"}, {"invalid header", "value2"}});
-  const auto result = test_controller.trigger();
+  const auto result = test_controller.trigger("data", {{"legit-header", "value1"}, {"invalid header", "value2"}});
   auto file_contents = result.at(InvokeHTTP::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller.plan->getContent(file_contents[0]) == "data");
@@ -406,8 +402,7 @@
   invokehttp->setProperty(InvokeHTTP::InvalidHTTPHeaderFieldHandlingStrategy, "drop");
   invokehttp->setProperty(InvokeHTTP::AttributesToSend, "");
   invokehttp->setAutoTerminatedRelationships({InvokeHTTP::RelNoRetry, InvokeHTTP::RelFailure, InvokeHTTP::RelResponse, InvokeHTTP::RelRetry});
-  test_controller.enqueueFlowFile("data", {{"legit-header", "value1"}, {"invalid header", "value2"}});
-  const auto result = test_controller.trigger();
+  const auto result = test_controller.trigger("data", {{"legit-header", "value1"}, {"invalid header", "value2"}});
   auto file_contents = result.at(InvokeHTTP::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller.plan->getContent(file_contents[0]) == "data");
@@ -429,8 +424,7 @@
   invokehttp->setProperty(InvokeHTTP::InvalidHTTPHeaderFieldHandlingStrategy, "drop");
   invokehttp->setProperty(InvokeHTTP::AttributesToSend, "he.*er");
   invokehttp->setAutoTerminatedRelationships({InvokeHTTP::RelNoRetry, InvokeHTTP::RelFailure, InvokeHTTP::RelResponse, InvokeHTTP::RelRetry});
-  test_controller.enqueueFlowFile("data", {{"header1", "value1"}, {"header", "value2"}});
-  const auto result = test_controller.trigger();
+  const auto result = test_controller.trigger("data", {{"header1", "value1"}, {"header", "value2"}});
   auto file_contents = result.at(InvokeHTTP::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller.plan->getContent(file_contents[0]) == "data");
diff --git a/extensions/librdkafka/tests/PublishKafkaTests.cpp b/extensions/librdkafka/tests/PublishKafkaTests.cpp
index b508ee0..78d65b0 100644
--- a/extensions/librdkafka/tests/PublishKafkaTests.cpp
+++ b/extensions/librdkafka/tests/PublishKafkaTests.cpp
@@ -31,8 +31,7 @@
   publish_kafka->setProperty(processors::PublishKafka::SeedBrokers, "test_seedbroker");
   publish_kafka->setProperty(processors::PublishKafka::QueueBufferMaxMessage, "1000");
   publish_kafka->setProperty(processors::PublishKafka::BatchSize, "1500");
-  test_controller.enqueueFlowFile("");
-  REQUIRE_THROWS_WITH(test_controller.trigger(), "Process Schedule Operation: Invalid configuration: Batch Size cannot be larger than Queue Max Message");
+  REQUIRE_THROWS_WITH(test_controller.trigger(""), "Process Schedule Operation: Invalid configuration: Batch Size cannot be larger than Queue Max Message");
 }
 
 }  // namespace org::apache::nifi::minifi::test
diff --git a/extensions/standard-processors/tests/unit/FetchFileTests.cpp b/extensions/standard-processors/tests/unit/FetchFileTests.cpp
index a6cd1d0..2643edc 100644
--- a/extensions/standard-processors/tests/unit/FetchFileTests.cpp
+++ b/extensions/standard-processors/tests/unit/FetchFileTests.cpp
@@ -87,8 +87,7 @@
 
 TEST_CASE_METHOD(FetchFileTestFixture, "Test fetching file with default but non-existent file path", "[testFetchFile]") {
   attributes_["filename"] = "non_existent.file";
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::NotFound);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == "");
@@ -99,8 +98,7 @@
 TEST_CASE_METHOD(FetchFileTestFixture, "FileToFetch property set to a non-existent file path", "[testFetchFile]") {
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::FileToFetch, "/tmp/non_existent.file");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::LogLevelWhenFileNotFound, "INFO");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::NotFound);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == "");
@@ -113,8 +111,7 @@
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::FileToFetch,
     input_dir_ + utils::file::FileUtils::get_separator() + permission_denied_file_name_);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::LogLevelWhenPermissionDenied, "WARN");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::PermissionDenied);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == "");
@@ -124,8 +121,7 @@
 #endif
 
 TEST_CASE_METHOD(FetchFileTestFixture, "Test fetching file with default file path", "[testFetchFile]") {
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -137,8 +133,7 @@
   utils::putFileToDir(input_dir_ + utils::file::FileUtils::get_separator() + "sub", input_file_name_, file_content_);
   auto file_path = input_dir_ + utils::file::FileUtils::get_separator() + "sub" + utils::file::FileUtils::get_separator() + input_file_name_;
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::FileToFetch, file_path);
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -147,8 +142,7 @@
 
 TEST_CASE_METHOD(FetchFileTestFixture, "Flow scheduling fails due to missing move destination directory when completion strategy is set to move file", "[testFetchFile]") {
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
-  test_controller_->enqueueFlowFile("", attributes_);
-  REQUIRE_THROWS_AS(test_controller_->trigger(), minifi::Exception);
+  REQUIRE_THROWS_AS(test_controller_->trigger("", attributes_), minifi::Exception);
 }
 
 TEST_CASE_METHOD(FetchFileTestFixture, "Flow fails due to move conflict", "[testFetchFile]") {
@@ -157,8 +151,7 @@
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveConflictStrategy, "Fail");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Failure);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == "");
@@ -173,8 +166,7 @@
   utils::putFileToDir(move_dir, input_file_name_, "old content");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveConflictStrategy, "Fail");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -186,8 +178,7 @@
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveConflictStrategy, "Replace File");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -203,8 +194,7 @@
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveConflictStrategy, "Rename");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -221,8 +211,7 @@
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveConflictStrategy, "Keep Existing");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -236,8 +225,7 @@
   auto move_dir = test_controller_->createTempDirectory();
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -252,8 +240,7 @@
   move_dir = move_dir + utils::file::FileUtils::get_separator() + "temp";
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -269,8 +256,7 @@
   utils::file::FileUtils::set_permissions(move_dir, 0);
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Move File");
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::MoveDestinationDirectory, move_dir);
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
@@ -283,8 +269,7 @@
 
 TEST_CASE_METHOD(FetchFileTestFixture, "Fetched file is deleted after flow completion", "[testFetchFile]") {
   fetch_file_processor_->setProperty(org::apache::nifi::minifi::processors::FetchFile::CompletionStrategy, "Delete File");
-  test_controller_->enqueueFlowFile("", attributes_);
-  const auto result = test_controller_->trigger();
+  const auto result = test_controller_->trigger("", attributes_);
   auto file_contents = result.at(minifi::processors::FetchFile::Success);
   REQUIRE(file_contents.size() == 1);
   REQUIRE(test_controller_->plan->getContent(file_contents[0]) == file_content_);
diff --git a/extensions/standard-processors/tests/unit/PutUDPTests.cpp b/extensions/standard-processors/tests/unit/PutUDPTests.cpp
index 6095d72..49f4e3f 100644
--- a/extensions/standard-processors/tests/unit/PutUDPTests.cpp
+++ b/extensions/standard-processors/tests/unit/PutUDPTests.cpp
@@ -88,8 +88,7 @@
 
   {
     const char* const message = "first message: hello";
-    controller.enqueueFlowFile(message);
-    const auto result = controller.trigger();
+    const auto result = controller.trigger(message);
     const auto& success_flow_files = result.at(PutUDP::Success);
     REQUIRE(success_flow_files.size() == 1);
     REQUIRE(result.at(PutUDP::Failure).empty());
@@ -101,8 +100,7 @@
 
   {
     const char* const message = "longer message AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";  // NOLINT
-    controller.enqueueFlowFile(message);
-    const auto result = controller.trigger();
+    const auto result = controller.trigger(message);
     const auto& success_flow_files = result.at(PutUDP::Success);
     REQUIRE(success_flow_files.size() == 1);
     REQUIRE(result.at(PutUDP::Failure).empty());
diff --git a/libminifi/include/core/ProcessContext.h b/libminifi/include/core/ProcessContext.h
index 1423bc4..6450f72 100644
--- a/libminifi/include/core/ProcessContext.h
+++ b/libminifi/include/core/ProcessContext.h
@@ -404,7 +404,7 @@
   std::string string_value;
   if (!getProperty(property, string_value, flow_file)) return std::nullopt;
   try {
-    if (!state::response::Value{string_value}.template convertValue(value)) return std::nullopt;
+    if (!state::response::Value{string_value}.template convertValue<>(value)) return std::nullopt;
   } catch (const utils::internal::ValueException&) {
     return std::nullopt;
   }
diff --git a/libminifi/test/SingleProcessorTestController.h b/libminifi/test/SingleProcessorTestController.h
index 1e4707c..b91b4c0 100644
--- a/libminifi/test/SingleProcessorTestController.h
+++ b/libminifi/test/SingleProcessorTestController.h
@@ -51,9 +51,10 @@
     return result;
   }
 
-  void enqueueFlowFile(const std::string_view input_flow_file_content, std::unordered_map<std::string, std::string> input_flow_file_attributes = {}) {
+  auto trigger(const std::string_view input_flow_file_content, std::unordered_map<std::string, std::string> input_flow_file_attributes = {}) {
     const auto new_flow_file = createFlowFile(input_flow_file_content, std::move(input_flow_file_attributes));
     input_->put(new_flow_file);
+    return trigger();
   }
 
   core::Relationship addDynamicRelationship(std::string name) {