Add retry logic for COPY operation in SnowflakeCopyBatchInsert with configurable max retries

[kentoyoshida]

Overview

This pull request introduces retry logic for the COPY operation in the SnowflakeCopyBatchInsert class and ensures better handling of transient errors during data load processes.

Detailed Description

1. Retry Logic:
   • Implemented a retry mechanism for the COPY operation in the CopyTask class.
   • Added a new configuration parameter max_copy_retries to allow users to define the maximum number of retry attempts. The default value is set to 3.
   • Included the method isRetryableForCopyTask to classify errors as retryable based on specific error messages (e.g., "JDBC driver encountered communication error").
2. Configuration Updates:
   • Modified the SnowflakeOutputPlugin to accept and pass the new max_copy_retries parameter.
3. Logging Improvements:
   • Enhanced logging for retry attempts, including error messages and retry counts, for better debugging and monitoring.

Related Issues

This changes addresses potential failures in the COPY operation caused by transient communication errors. Previously, such errors could lead to incomplete data loads without any retry attempts.

Impact

• Improved Reliability:
  This update ensures better handling of transient errors and increases the likelihood of successful data loads, even under unstable network conditions.
• Breaking Changes:
  There are no significant breaking changes, as the default value for max_copy_retries ensures backward compatibility.

Additional Notes

• Please review the retry mechanism and configuration changes to confirm they align with project requirements and coding standards.

[yas-okadatech]

LGTM!

[chikamura]

Could you add a description of max_copy_retries in the configuration section of README.md, together with max_upload_retries?

[chikamura]

Memo: There is already a class in embulk that handles retries.

https://dev.embulk.org/embulk-util-retryhelper/0.8.0/javadoc/org/embulk/util/retryhelper/RetryExecutor.html
https://github.com/embulk/embulk-output-jdbc/blob/1a04c539bf879f8771245178f9bbc0d2cee71d30/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java#L1324-L1329

But in the existing code, they're making changes like this one.

embulk-output-snowflake/src/main/java/org/embulk/output/snowflake/SnowflakeCopyBatchInsert.java

Lines 367 to 405 in 76dc706

	public Void call() throws IOException, SQLException, InterruptedException {
	int retries = 0;
	try {
	long startTime = System.currentTimeMillis();
	// put file to snowflake internal storage
	SnowflakeOutputConnection con = (SnowflakeOutputConnection) connector.connect(true);
	while (true) {
	try {
	logger.info(
	String.format(
	"Uploading file id %s to Snowflake (%,d bytes %,d rows)",
	snowflakeStageFileName, file.length(), batchRows));
	FileInputStream fileInputStream = new FileInputStream(file);
	con.runUploadFile(stageIdentifier, snowflakeStageFileName, fileInputStream);
	break;
	} catch (SQLException e) {
	retries++;
	if (retries > this.maxUploadRetries) {
	throw e;
	}
	logger.warn(
	String.format(
	"Upload error %s file %s retries: %d", e, snowflakeStageFileName, retries));
	Thread.sleep(retries * retries * 1000);
	}
	}
	double seconds = (System.currentTimeMillis() - startTime) / 1000.0;
	logger.info(
	String.format("Uploaded file %s (%.2f seconds)", snowflakeStageFileName, seconds));
	} finally {
	file.delete();
	}
	return null;
	}
	}

[chikamura]

If possible, could you add a description of max_copy_retries in the configuration section of README.md, together with max_upload_retries?

[chikamura]

The error occurs because the connection that was closed on the second retry is being accessed again.

[chikamura]

LGTM

[kentoyoshida]

@chikamura
Thank you very much for your review 🙇
I also appreciate you identifying the connection resource release bug.

The issue where the connection was being closed during retries has been addressed in the following commit:
5a6318a

The README update has been handled in the following commit:
7a3f515