Akash3121 commented on code in PR #823:
URL: https://github.com/apache/incubator-xtable/pull/823#discussion_r4033235631
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/ITRunSync.java:
##########
@@ -93,6 +93,53 @@ void testContinuousSyncMode(@TempDir Path tempDir) throws
IOException {
}
}
+ @Test
+ void testSyncCompletesBeforeTimeout(@TempDir Path tempDir) throws
IOException {
+ String tableName = "tiny-table";
+ try (GenericTable table =
+ TestJavaHudiTable.forStandardSchema(
+ tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
+ table.insertRows(5); // small numRows for it to finish before the
timeout does
+ File configFile = writeConfigFile(tempDir, table, tableName);
+
+ String[] args = new String[] {"--datasetConfig", configFile.getPath(),
"--syncTimeout", "60"};
+ RunSync.main(args);
+ // check successful sync
+ Path icebergMetadataPath = Paths.get(URI.create(table.getBasePath() +
"/metadata"));
+ waitForNumIcebergCommits(icebergMetadataPath, 3);
+ }
+ }
+
+ @Test
+ void testSyncTimeoutFires(@TempDir Path tempDir) throws IOException {
+ String tableName = "large-table";
+ try (GenericTable table =
+ TestJavaHudiTable.forStandardSchema(
+ tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
+
+ table.insertRows(5000);
Review Comment:
this test is timing and machine-dependent. If 5,000 rows finish in under one
second, it passes without exercising the timeout branch. It also passes if
RunSync.main() returns while the worker continues in the background - the
failure mode we need to prevent.
I think better to use a deterministic blocking test double coordinated with
latches rather than table size. The test should assert that the timeout was
observed, cancellation/interruption reached the worker, the worker terminated,
and no sync activity continued after RunSync.main() returned. It would also be
useful to verify the behavior of the next dataset after a timeout.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -155,38 +165,70 @@ static List<TargetTable> targetTableBuilder(
return targetTables;
}
- static void syncTableMetdata(
+ static void syncTableMetadata(
DatasetConfig datasetConfig,
List<String> tableFormatList,
CatalogConfig catalogConfig,
Configuration hadoopConf,
- ConversionSourceProvider conversionSourceProvider) {
+ ConversionSourceProvider conversionSourceProvider,
+ long timeoutInSeconds) {
ConversionController conversionController = new
ConversionController(hadoopConf);
- for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
- log.info(
- "Running sync for basePath {} for following table formats {}",
- table.getTableBasePath(),
- tableFormatList);
- Properties sourceProperties = new Properties();
- if (table.getPartitionSpec() != null) {
- sourceProperties.put(
- HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG,
table.getPartitionSpec());
- }
+ // use a single-thread executor since tasks are processed sequentially
+ ExecutorService syncExecutor = Executors.newSingleThreadExecutor();
+ try {
- SourceTable sourceTable =
- sourceTableBuilder(table, catalogConfig, datasetConfig,
sourceProperties);
- List<TargetTable> targetTables = targetTableBuilder(table,
catalogConfig, tableFormatList);
- ConversionConfig conversionConfig =
- ConversionConfig.builder()
- .sourceTable(sourceTable)
- .targetTables(targetTables)
- .syncMode(SyncMode.INCREMENTAL)
- .build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG,
table.getPartitionSpec());
+ }
+
+ SourceTable sourceTable =
+ sourceTableBuilder(table, catalogConfig, datasetConfig,
sourceProperties);
+ List<TargetTable> targetTables = targetTableBuilder(table,
catalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ if (timeoutInSeconds > 0) {
+ Future<?> standardFuture =
+ syncExecutor.submit(
+ () -> {
+ conversionController.sync(conversionConfig,
conversionSourceProvider);
+ });
+ try {
+ standardFuture.get(timeoutInSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ log.error(
+ "Sync timed out for {} after {} seconds. Triggering thread
interrupt.",
+ table.getTableBasePath(),
+ timeoutInSeconds);
+ standardFuture.cancel(true);
+ } catch (ExecutionException e) {
+ log.error("Error running sync for {}", table.getTableBasePath(),
e.getCause());
+ } catch (InterruptedException e) {
+ log.error("Sync main runner thread was interrupted", e);
+ standardFuture.cancel(true);
+ Thread.currentThread().interrupt();
Review Comment:
After restoring the runner’s interrupt status, better to stop processing and
return or break from the dataset loop. The current code continues to the next
dataset, submits another task, and then get() immediately throws because the
runner is still interrupted. The submitted task may already have started, which
means shutdown can trigger additional sync work after cancellation was
requested.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -155,38 +165,70 @@ static List<TargetTable> targetTableBuilder(
return targetTables;
}
- static void syncTableMetdata(
+ static void syncTableMetadata(
DatasetConfig datasetConfig,
List<String> tableFormatList,
CatalogConfig catalogConfig,
Configuration hadoopConf,
- ConversionSourceProvider conversionSourceProvider) {
+ ConversionSourceProvider conversionSourceProvider,
+ long timeoutInSeconds) {
ConversionController conversionController = new
ConversionController(hadoopConf);
- for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
- log.info(
- "Running sync for basePath {} for following table formats {}",
- table.getTableBasePath(),
- tableFormatList);
- Properties sourceProperties = new Properties();
- if (table.getPartitionSpec() != null) {
- sourceProperties.put(
- HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG,
table.getPartitionSpec());
- }
+ // use a single-thread executor since tasks are processed sequentially
+ ExecutorService syncExecutor = Executors.newSingleThreadExecutor();
+ try {
- SourceTable sourceTable =
- sourceTableBuilder(table, catalogConfig, datasetConfig,
sourceProperties);
- List<TargetTable> targetTables = targetTableBuilder(table,
catalogConfig, tableFormatList);
- ConversionConfig conversionConfig =
- ConversionConfig.builder()
- .sourceTable(sourceTable)
- .targetTables(targetTables)
- .syncMode(SyncMode.INCREMENTAL)
- .build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG,
table.getPartitionSpec());
+ }
+
+ SourceTable sourceTable =
+ sourceTableBuilder(table, catalogConfig, datasetConfig,
sourceProperties);
+ List<TargetTable> targetTables = targetTableBuilder(table,
catalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ if (timeoutInSeconds > 0) {
+ Future<?> standardFuture =
+ syncExecutor.submit(
+ () -> {
+ conversionController.sync(conversionConfig,
conversionSourceProvider);
+ });
+ try {
+ standardFuture.get(timeoutInSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ log.error(
Review Comment:
timeout is logged and swallowed, so a one-shot RunSync invocation returns
normally with a successful process exit even though the table was not
synchronized. Callers and schedulers need a machine-detectable failure. Could
we propagate the timeout from one-shot mode, while letting continuous mode
catch it at the scheduling boundary, log it, and retry according to its
documented behavior?
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -288,14 +330,29 @@ private static void runSync(CommandLine cmd) throws
IOException {
String icebergCatalogConfigpath = getValueFromConfig(cmd,
ICEBERG_CATALOG_CONFIG_PATH);
String hadoopConfigpath = getValueFromConfig(cmd, HADOOP_CONFIG_PATH);
String conversionProviderConfigpath = getValueFromConfig(cmd,
CONVERTERS_CONFIG_PATH);
+ String rawTimeout = getValueFromConfig(cmd, SYNC_TIMEOUT_OPTION);
+ if (rawTimeout != null && !rawTimeout.isEmpty()) {
Review Comment:
In continuous mode, runSync(cmd) runs inside scheduleAtFixedRate , whose
callback catches only IO Exception . An invalid timeout throws
IllegalArgumentException, and a digit-only value bigger than Long.MAX_VALUE
throws NumberFormatException . Either exception escapes the scheduled task,
which permanently suppresses later executions while the main thread keeps
sleeping, leaving a live process that performs no more syncs.
better to parse and validate the timeout before starting the scheduler,
including numeric overflow, and surface a clear configuration error rather than
silently disabling continuous synchronization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
