added dlq implementation v1

[bharadwaj-aditya]

Added support for writing errors to DLQ.

This PR does not include the files in a format that can be retried using the existing retrial scripts. That will be taken up in a follow up PR.

[codecov]

Codecov Report

Attention: Patch coverage is 64.12214% with 47 lines in your changes are missing coverage. Please review.

Project coverage is 43.54%. Comparing base (8dfb6de) to head (80ad3ae).
Report is 55 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #1577      +/-   ##
============================================
+ Coverage     40.93%   43.54%   +2.61%     
+ Complexity     2808      654    -2154     
============================================
  Files           739      294     -445     
  Lines         42855    15542   -27313     
  Branches       4582     1548    -3034     
============================================
- Hits          17542     6768   -10774     
+ Misses        23814     8255   -15559     
+ Partials       1499      519     -980     

Components	Coverage Δ
spanner-templates	55.73% <64.12%> (-4.13%)	⬇️
spanner-import-export	∅ <ø> (∅)
spanner-live-forward-migration	73.64% <75.00%> (-0.04%)	⬇️
spanner-live-reverse-replication	48.55% <75.00%> (-0.09%)	⬇️
spanner-bulk-migration	78.07% <64.06%> (+0.02%)	⬆️

Files	Coverage Δ
...le/cloud/teleport/v2/constants/MetricCounters.java	0.00% <ø> (ø)
.../v2/source/reader/io/schema/SourceTableSchema.java	100.00% <ø> (ø)
...google/cloud/teleport/v2/templates/RowContext.java	100.00% <100.00%> (ø)
...eport/v2/constants/SourceDbToSpannerConstants.java	66.66% <66.66%> (ø)
...leport/v2/transformer/SourceRowToMutationDoFn.java	97.36% <90.90%> (-2.64%)	⬇️
.../v2/spanner/migrations/avro/AvroToValueMapper.java	90.90% <75.00%> (-1.56%)	⬇️
...google/cloud/teleport/v2/writer/SpannerWriter.java	50.00% <50.00%> (ø)
...ogle/cloud/teleport/v2/writer/DeadLetterQueue.java	73.33% <73.33%> (ø)
...cloud/teleport/v2/templates/SourceDbToSpanner.java	24.28% <25.00%> (+24.28%)	⬆️

... and 466 files with indirect coverage changes

[Deep1998]

LGTM

[manitgupta]

Can we structure this a little better? The code flow is very hard to understand, especially wrt how failed mutations and failed transformations are getting written to the DLQ.

We used TupleTag for transformer errors to collect the errors and we directly write the records for mutation errors. We should be consistent.

IMO if we should collect the failed records, attributed to the failure reason (transform, retryable, permanent etc) and then write them out to a DLQ...

The PCollections of all types of errors (categorized by permanent or retryable) can be merged and the DLQ can be written in a single stage, something like -

// example for class of permanent errors
PCollectionList.of(errorTransformedRecords)
            .and(spannerWriteResults.permanentErrors())
            .apply(Flatten.pCollections())
            .apply(DLQWriteTransform....)

[bharadwaj-aditya]

Can we structure this a little better? The code flow is very hard to understand, especially wrt how failed mutations and failed transformations are getting written to the DLQ.

We used TupleTag for transformer errors to collect the errors and we directly write the records for mutation errors. We should be consistent.

IMO if we should collect the failed records, attributed to the failure reason (transform, retryable, permanent etc) and then write them out to a DLQ...

The PCollections of all types of errors (categorized by permanent or retryable) can be merged and the DLQ can be written in a single stage, something like -

// example for class of permanent errors
PCollectionList.of(errorTransformedRecords)
            .and(spannerWriteResults.permanentErrors())
            .apply(Flatten.pCollections())
            .apply(DLQWriteTransform....)

That was the original idea, but this is currently not possible in this flow owing to the different objects being created. I will catch up with you to explain.