vinishjail97 commented on code in PR #823:
URL: https://github.com/apache/incubator-xtable/pull/823#discussion_r3364519394
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,7 +117,14 @@ public class RunSync {
"continuousModeInterval",
true,
"The interval in seconds to schedule the loop. Requires
--continuousMode to be set. Defaults to 5 seconds.")
+ .addOption(HELP_OPTION, "help", false, "Displays help information to
run this utility")
+ .addOption(
+ SYNC_TIMEOUT_OPTION,
+ "syncTimeout",
+ true,
+ "The maximum time in seconds allowed for a single table sync
before timing out. Defaults to no timeout.")
.addOption(HELP_OPTION, "help", false, "Displays help information to
run this utility");
+ ;
Review Comment:
Stray empty statement `;` (and `HELP_OPTION` is now added twice — the new
one on L120 plus the original on L126). Remove both.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -182,10 +194,47 @@ static void syncTableMetdata(
.targetTables(targetTables)
.syncMode(SyncMode.INCREMENTAL)
.build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
+ if (timeoutInSeconds > 0) {
+ CompletableFuture<Void> syncFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ conversionController.sync(conversionConfig,
conversionSourceProvider);
+ },
+ syncExecutor);
+
+ delayer.schedule(
+ () -> {
+ if (!syncFuture.isDone()) {
+ // end complete the future exceptionally with a
TimeoutException
+ syncFuture.completeExceptionally(
+ new java.util.concurrent.TimeoutException(
+ "Sync timed out for " + table.getTableBasePath()));
+ syncFuture.cancel(true); // send interrupt signal to the
executing worker thread
Review Comment:
`CompletableFuture.cancel(true)` does not interrupt the running task — per
the Javadoc, `mayInterruptIfRunning` has no effect for `CompletableFuture`. And
since `completeExceptionally(...)` on the line above has already completed the
future, this `cancel` returns `false` and is a no-op. Net effect: the worker
running `conversionController.sync()` keeps running to completion after the
"timeout." Switch to `ExecutorService.submit(...).get(timeout, SECONDS)` — a
real `FutureTask`'s `cancel(true)` actually sends the interrupt (best-effort,
since `sync()` must honor it).
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -288,14 +337,24 @@ 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);
+ long timeoutInSeconds =
+ getValueFromConfig(cmd, SYNC_TIMEOUT_OPTION) == null
+ || getValueFromConfig(cmd, SYNC_TIMEOUT_OPTION).isEmpty()
+ ? 0L
+ : Long.parseLong(getValueFromConfig(cmd, SYNC_TIMEOUT_OPTION));
Review Comment:
`Long.parseLong` is unguarded: a non-numeric `--timeout` throws
`NumberFormatException` (uncaught), and a negative value silently falls through
as "no timeout" via the `> 0` check. Validate the input and fail fast with a
clear message, e.g. `Preconditions.checkArgument(timeout >= 0, ...)`.
##########
xtable-core/src/test/java/org/apache/xtable/ITConversionController.java:
##########
@@ -223,6 +223,9 @@ private ConversionSourceProvider<?>
getConversionSourceProvider(String sourceTab
@MethodSource("generateTestParametersForFormatsSyncModesAndPartitioning")
public void testVariousOperations(
String sourceTableFormat, SyncMode syncMode, boolean isPartitioned) {
+ // Disable the lock manager and heartbeat threads for both Catalog and
Table configurations
Review Comment:
This lock-impl/heartbeat disabling looks unrelated to a per-table sync
timeout. Is it masking the executor/heartbeat thread leak from the RunSync
change (e.g. the IT not terminating)? Please split it into its own PR or
explain the connection in the description — keeping this PR scoped to the
timeout feature makes it easier to review and revert.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -155,13 +164,16 @@ 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);
+ java.util.concurrent.ExecutorService syncExecutor =
Executors.newCachedThreadPool();
Review Comment:
`newCachedThreadPool()` is unnecessary here — syncs run sequentially (you
`join` before the next loop iteration), so a single-thread executor suffices.
Checklist: prefer fixed pools over `newCachedThreadPool()`. Also: please use
proper imports rather than inline fully-qualified `java.util.concurrent.*`
names (the file already imports `Executors`/`TimeUnit`).
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -155,13 +164,16 @@ 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);
+ java.util.concurrent.ExecutorService syncExecutor =
Executors.newCachedThreadPool();
+ java.util.concurrent.ScheduledExecutorService delayer =
Executors.newScheduledThreadPool(1);
Review Comment:
**Blocking: these executors are never shut down.** `syncExecutor` and
`delayer` are created on every `syncTableMetadata` call but there is no
`shutdown()`/`shutdownNow()`. In continuous mode, `main` schedules `runSync`
via `scheduleAtFixedRate`, so `syncTableMetadata` runs every interval and leaks
two thread pools each tick — unbounded thread growth, and the non-daemon
threads will keep the JVM alive. Create the executor once (or shut it down in a
`finally`) and reuse it.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -182,10 +194,47 @@ static void syncTableMetdata(
.targetTables(targetTables)
.syncMode(SyncMode.INCREMENTAL)
.build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
+ if (timeoutInSeconds > 0) {
+ CompletableFuture<Void> syncFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ conversionController.sync(conversionConfig,
conversionSourceProvider);
+ },
+ syncExecutor);
+
+ delayer.schedule(
+ () -> {
+ if (!syncFuture.isDone()) {
+ // end complete the future exceptionally with a
TimeoutException
+ syncFuture.completeExceptionally(
+ new java.util.concurrent.TimeoutException(
+ "Sync timed out for " + table.getTableBasePath()));
+ syncFuture.cancel(true); // send interrupt signal to the
executing worker thread
+ }
+ },
+ timeoutInSeconds,
+ TimeUnit.SECONDS);
+
+ // wait for whichever happens first (completion or timeout exception)
+ try {
+ syncFuture.join();
Review Comment:
**Blocking: no tests for the feature.** The timeout paths (timeout fires,
completes before timeout, timeout disabled) are the core of this PR and are
easily testable by injecting a fake/sleeping sync. Please add unit tests before
landing — also covers that a slow sync actually surfaces the timeout log and
the executor is cleaned up.
--
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]
