Akash3121 commented on code in PR #823:
URL: https://github.com/apache/incubator-xtable/pull/823#discussion_r4033207872
##########
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);
Review Comment:
cancel(true) only requests interruption - it doesn't gaurantee that
`conversionController.sync()` has stopped. The conversion stack does not
currently define or check a cooperative interruption (contract), and
`shutdownNow()` below also returns without confirming worker termination. A
filesystem/table-format operation that ignores interruption can continue
mutating the target after this method reports a timeout.
this can be especially unsafe in a continuous mode - next scheduled
invocation creates another executor and can start a second sync against the
same table while the timed out sync is still running.
I think it would be better for the task to actually terminated before
processing another dataset or scheduling another run. This likely requires
cooperative cancellation through the sync path plus bounded termination
verification. If termination cannot be confirmed, the runner should stop rather
than treating the timed-out operation as cancelled. (fail fast)
--
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]
