[Improve][log] Add logs in transform/sink error

[hailin0]

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config
• Update the release-note.

[Hisoka-X]

Please rebase on dev to trigger ci.

[Copilot]

Pull Request Overview

This PR aims to improve error logging in transform and sink operations by wrapping critical method calls in try-catch blocks and logging more detailed error messages.

• Introduces try-catch blocks in map/flatMap transforms to capture transformation errors.
• Adds additional error logging for record processing and sink commit/snapshot operations.
• Refactors conditional checks in transformation flows to log filtered rows.

Reviewed Changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogMapTransform.java	Enhanced error handling in map transform by wrapping transform.map(row) in try-catch and logging errors.
seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java	Similar error logging improvements for flatMap transform operations.
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/TransformFlowLifeCycle.java	Added try-catch around record processing and refined the logic to check for empty output lists.
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java	Inserted additional error logging when record reception fails.
seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java	Wrapped snapshotState and prepareCommit calls in try-catch blocks with enhanced logging.
seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkCommitter.java	Added try-catch to log errors during commit operations.
seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java	Enhanced error handling for aggregated commit operations using try-catch and logging.

Comments suppressed due to low confidence (6)

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogMapTransform.java:55

• [nitpick] The error message contains a grammatical issue ('throw' should be 'threw' or rephrased to 'encountered an error while mapping'). Consider revising it for clarity.

String.format("The transform %s map table %s data throw an error", transform.getPluginName(), tableId);

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java:55

• [nitpick] The error message should be revised to correct the grammatical error ('throw' should be 'threw' or reworded to improve readability).

String.format("The transform %s map table %s data throw an error", transform.getPluginName(), tableId);

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java:227

• [nitpick] The error message in snapshotState has a grammatical mistake; consider changing it to 'threw an error' or 'encountered an error during snapshotState'.

String.format("table %s snapshotState throw an error", sinkWriterEntry.getKey().getTableIdentifier());

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java:270

• [nitpick] Consider updating the error message in prepareCommit to correct the grammar (e.g., 'threw an error during commit') for better clarity.

String.format("table %s prepareCommit throw an error", sinkWriterEntry.getKey().getTableIdentifier());

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkCommitter.java:64

• [nitpick] The error message in the commit method contains a grammatical error; please consider revising it to 'threw an error' or another clear phrasing.

String.format("table %s commit throw an error", sinkIdentifier);

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java:93

• [nitpick] The error message here should be rephrased to correct the grammar (for example, 'threw an error during commit').

String.format("table %s commit throw an error", sinkIdentifier);