Add metrics/DLQ record for records getting silently dropped with a warning in reverse replication

[aasthabharill]

https://b.corp.google.com/issues/432145898

Summary

This PR addresses critical issues in the SpannerToSourceDb pipeline where certain error conditions were being silently dropped or causing inefficient worker retries. A thorough investigation of LOG warnings and exceptions revealed that invalid DML generation and specific JSON parsing failures were not being handled robustly.

Changes

DML Generation Integrity

• Problem: Previously, when the DMLGenerator encountered invalid requests (e.g., missing primary keys or schema mismatches), it returned an empty string ("") and logged a warning. This caused SourceWriterFn to mistakenly treat the record as "processed" or simply skip it, leading to silent data loss without alerting the user or routing to the Dead Letter Queue (DLQ).
• Fix: Updated MySQLDMLGenerator and CassandraDMLGenerator to throw a new InvalidDMLGenerationException
  instead of returning an empty string.
• Classification: This exception is classified as a Severe Error because its caused due to schema issues that will not resolve upon retry.
• Propagation: The exception is now propagated from DMLGenerator -> InputRecordProcessor -> SourceWriterFn
  It is then caught in SourceWriterFn, correctly unwrapped and identified by SpannerToSourceDbExceptionClassifier, and finally routed to the DLQ with a PERMANENT_ERROR_TAG to ensure visibility and prevent data loss.

Testing

{"message":{"commitTimestamp":{"seconds":1767956884,"nanos":38751000},"serverTransactionId":"MTc5NTY4OTg1MjU2MjY4MDcxMzU=","recordSequence":"00000000","tableName":"sedges","mod":{"keysJson":"{\"created\":\"23\",\"ddrkey\":\"354\"}","oldValuesJson":"{}","newValuesJson":"{\"content\":null,\"flags\":\"-1326272220\",\"from\":\"nodeA\",\"to\":\"nodeB\",\"type\":\"edgeType1\",\"value\":\"edgeValue1\"}"},"modType":"INSERT","numberOfRecordsInTransaction":1,"transactionTag":"","shard":"shard2_00","isRetryRecord":false,"_metadata_retry_count":0,"_metadata_historical_retry_count":0},"error_message":"UNKNOWN: Cannot reverse replicate for source table sedges without primary key, skipping the record. Source Table: SourceTable{name=sedges, schema=custom_transform_test00, columns=[SourceColumn{name=ddrkey, type=bigint, isNullable=false, isPrimaryKey=false, size=null, precision=19, scale=0, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=from, type=varchar, isNullable=false, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=to, type=varchar, isNullable=true, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=type, type=varchar, isNullable=true, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=value, type=varchar, isNullable=true, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=content, type=varbinary, isNullable=true, isPrimaryKey=false, size=60000, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=flags, type=bigint, isNullable=false, isPrimaryKey=false, size=null, precision=19, scale=0, columnOptions=[], sourceType=MYSQL}], primaryKeyColumns=[], sourceType=MYSQL, comment=null}, Caused by: Cannot reverse replicate for source table sedges without primary key, skipping the record. Source Table: SourceTable{name=sedges, schema=custom_transform_test00, columns=[SourceColumn{name=ddrkey, type=bigint, isNullable=false, isPrimaryKey=false, size=null, precision=19, scale=0, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=from, type=varchar, isNullable=false, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=to, type=varchar, isNullable=true, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=type, type=varchar, isNullable=true, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=value, type=varchar, isNullable=true, isPrimaryKey=false, size=255, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=content, type=varbinary, isNullable=true, isPrimaryKey=false, size=60000, precision=null, scale=null, columnOptions=[], sourceType=MYSQL}, SourceColumn{name=flags, type=bigint, isNullable=false, isPrimaryKey=false, size=null, precision=19, scale=0, columnOptions=[], sourceType=MYSQL}], primaryKeyColumns=[], sourceType=MYSQL, comment=null}"}

[gemini-code-assist]

Summary of Changes

Hello @aasthabharill, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request focuses on enhancing the observability and understanding of record processing within the reverse replication pipeline. It introduces a series of detailed comments to existing logging statements across several core components. These comments serve to identify and categorize various error and warning conditions, explicitly noting where records might be silently dropped, where exceptions are handled (or not), and how different issues interact with the Dead Letter Queue (DLQ) and retry mechanisms. This work lays the groundwork for future improvements in error handling, metrics, and DLQ integration by clearly documenting current behaviors and potential points of failure.

Highlights

• Identification of Silently Dropped Records: Extensive comments have been added to LOG.warn statements, particularly within DML generators (CassandraDMLGenerator, MySQLDMLGenerator) and InputRecordProcessor, to explicitly flag instances where records are currently dropped without being sent to a Dead Letter Queue (DLQ).
• Clarification of Error Handling Paths: New comments delineate the expected behavior of various error and warning logs, indicating whether an issue is "severe", "retryable", "seems fine", or "problematic" and how it impacts record processing or DLQ mechanisms.
• Pinpointing Unhandled Exceptions: Specific comments highlight potential code paths where exceptions might occur but are not adequately caught or handled, such as in AssignShardIdFn and ShardingLogicImplFetcher, suggesting areas for robustness improvements.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[codecov]

Codecov Report

❌ Patch coverage is 65.51724% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 50.85%. Comparing base (368d6f5) to head (2dd5113).
⚠️ Report is 8 commits behind head on main.

Files with missing lines	Patch %	Lines
...rt/v2/templates/dbutils/dml/MySQLDMLGenerator.java	64.28%	5 Missing ⚠️
...2/templates/dbutils/dml/CassandraDMLGenerator.java	71.42%	4 Missing ⚠️
...plates/dbutils/processor/InputRecordProcessor.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #3200      +/-   ##
============================================
+ Coverage     50.80%   50.85%   +0.05%     
- Complexity     5124     5131       +7     
============================================
  Files           976      976              
  Lines         60058    60086      +28     
  Branches       6569     6572       +3     
============================================
+ Hits          30511    30557      +46     
+ Misses        27406    27374      -32     
- Partials       2141     2155      +14     

Components	Coverage Δ
spanner-templates	70.98% <65.51%> (-0.02%)	⬇️
spanner-import-export	68.96% <ø> (-0.04%)	⬇️
spanner-live-forward-migration	80.12% <ø> (+0.02%)	⬆️
spanner-live-reverse-replication	77.69% <65.51%> (-0.03%)	⬇️
spanner-bulk-migration	88.44% <ø> (ø)

Files with missing lines	Coverage Δ
...es/utils/SpannerToSourceDbExceptionClassifier.java	95.23% <ø> (ø)
...plates/dbutils/processor/InputRecordProcessor.java	87.50% <0.00%> (-3.73%)	⬇️
...2/templates/dbutils/dml/CassandraDMLGenerator.java	81.48% <71.42%> (-0.23%)	⬇️
...rt/v2/templates/dbutils/dml/MySQLDMLGenerator.java	88.73% <64.28%> (+0.39%)	⬆️

... and 6 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[shreyakhajanchi]

Can you test for cases where if the schema is fixed and the severe dlq errors classified as part of the current pr are replayed then they get successfully migrated?