gcp_cloud_storage(input) lack of randomize Objects List

Hi,

I have a pipeline that reads from a GCS bucket using the "gcp_cloud_storage" component and deletes the file after it has been acknowledged. This works well when using one thread/broker/pod, but when multiple threads are used, a race condition issue arises causing only one "worker" to successfully complete the operation while the others trigger an error.
"level=error msg="Failed to read message: storage: object doesn't exist"

The root cause of the issue is the Go package "cloud.google.com/go/storage" which returns objects in a lexicographical order. When multiple workers are used, they both do the listing and only one of them can successfully perform the operation and delete the file while the other receives an error that the file is missing.

I have implemented a cache in my pipeline to prevent duplication, but the input reads the file before it and triggers the error.

Possible solutions:

1)Use a workaround
2)Add a Randomize feature for gcp_cloud_storage component: By adding a preliminary rand.Shuffle for the pending objects list, every "worker" would have a different order. I am (more than) willing to create a pull request for this solution, if this is the right way

My pipeline is as follows:

input:
read_until:
restart_input: true
check: false # tried use errored() to recreate the pod - no go
input:
gcp_cloud_storage:
bucket: "$INPUT_Bucket"
prefix: "$INPUT_Bucket_Prefix"
codec: all-bytes
delete_objects: true

pipeline:
  processors:
  - bloblang: |
      meta file_name = meta("gcs_key").filepath_split().slice(-1).join("")
  - cache:
      resource: main_cache_redis
      operator: add
      key: '${!  meta("file_name") }' 
      value: '${!   meta("gcs_last_modified_unix") }'
  - mapping: root = if errored() { deleted() }

Thanks,
Almog

Start at the gcp_cloud_storage input and its use of cloud.google.com/go/storage; inspect how pending objects are listed and deleted. Reproduce the multi-worker behavior using the pipeline shown in the issue, with completion defined as workers receiving different object orders without the missing-object error.