Support Tap for SMB writes (addresses #5080)

[clairemcginty]

addresses #5080

Simple implementation of Tap for SMB writes

todo:

• Tap for SMB writes
• Tap for SMB pre-keyed writes
• Tap for SMB transforms

[clairemcginty]

We could get around the signature change by adding a new method to SortedBucketIO.Write like:

public Iterator<T> tapBucketFile(ResourceId resourceId) {
   return getFileOperations().open(resourceId);
}

but this seems overtly hacky to me - plus, we re-create a FileOperations instance for every bucket/shard combo

[clairemcginty]

tbh it might be simplest to just make all these getters public 🤷‍♀️

[clairemcginty]

this is set here: https://github.com/spotify/scio/blob/v0.13.6/scio-smb/src/main/java/org/apache/beam/sdk/extensions/smb/SortedBucketSink.java#L512

[clairemcginty]

We could keep the BucketShardId so the tap returns a mapping from Bucket ID --> records in that bucket - although we'd have to also find a way to implement this in test context (probably by making another SortedBucketIO.Write#getBucketMetadata public too and mapping the bucketing function over each elements)

[codecov]

Codecov Report

Attention: 4 lines in your changes are missing coverage. Please review.

Comparison is base (54027fc) 63.42% compared to head (c3e7ca5) 63.55%.
Report is 2 commits behind head on main.

Files	Patch %	Lines
...cala/com/spotify/scio/smb/SortMergeTransform.scala	50.00%	4 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #5144      +/-   ##
==========================================
+ Coverage   63.42%   63.55%   +0.13%     
==========================================
  Files         291      291              
  Lines       10832    10845      +13     
  Branches      755      758       +3     
==========================================
+ Hits         6870     6893      +23     
+ Misses       3962     3952      -10     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[RustedBones]

Introduction of duplicated transform