diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java index f1bf98527a..46c0084ee5 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java @@ -332,7 +332,7 @@ private void handleSourceReaderFailOver() { @VisibleForTesting protected void handleTableTieringReachMaxDuration( - TablePath tablePath, long tableId, long tieringEpoch) { + TablePath tablePath, long tableId, long tieringEpoch, long maxTieringDurationMs) { Long currentEpoch = tieringTableEpochs.get(tableId); if (currentEpoch != null && currentEpoch.equals(tieringEpoch)) { LOG.info("Table {}-{} reached max duration. Force completing.", tablePath, tableId); @@ -355,9 +355,40 @@ protected void handleTableTieringReachMaxDuration( LOG.info("Send {} to reader {}", tieringReachMaxDurationEvent, reader); context.sendEventToSourceReader(reader, tieringReachMaxDurationEvent); } + + timerService.schedule( + () -> + context.runInCoordinatorThread( + () -> + failTieringJobIfNotFinished( + tablePath, + tableId, + tieringEpoch, + maxTieringDurationMs)), + maxTieringDurationMs, + TimeUnit.MILLISECONDS); } } + @VisibleForTesting + void failTieringJobIfNotFinished( + TablePath tablePath, long tableId, long tieringEpoch, long maxTieringDurationMs) { + Long currentEpoch = tieringTableEpochs.get(tableId); + if (currentEpoch != null && currentEpoch.equals(tieringEpoch)) { + throw new FlinkRuntimeException( + String.format( + "Tiering table %s-%d did not finish within %d ms after reaching max duration. " + + "Failing the tiering job to recover the stuck source readers.", + tablePath, tableId, maxTieringDurationMs)); + } + } + + @VisibleForTesting + @Nullable + Long getTieringEpoch(long tableId) { + return tieringTableEpochs.get(tableId); + } + @VisibleForTesting void generateAndAssignSplits( @Nullable Tuple3 tieringTable, Throwable throwable) { @@ -476,6 +507,8 @@ private void generateTieringSplits(Tuple3 tieringTable) finishedTables.put(tieringTable.f0, TieringFinishInfo.from(tieringTable.f1)); } else { pendingSplits.addAll(tieringSplits); + long maxTieringDurationMs = + tableInfo.getTableConfig().getDataLakeFreshness().toMillis(); timerService.schedule( () -> @@ -484,10 +517,11 @@ private void generateTieringSplits(Tuple3 tieringTable) handleTableTieringReachMaxDuration( tablePath, tieringTable.f0, - tieringTable.f1)), + tieringTable.f1, + maxTieringDurationMs)), // for simplicity, we use the freshness as - tableInfo.getTableConfig().getDataLakeFreshness().toMillis(), + maxTieringDurationMs, TimeUnit.MILLISECONDS); } } catch (Exception e) { diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java index 8d1f62eb34..c6aa5d3b63 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java @@ -63,6 +63,7 @@ import static org.apache.fluss.config.ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Unit tests for {@link TieringSourceEnumerator} and {@link TieringSplitGenerator}. */ @@ -829,11 +830,70 @@ void testNetworkErrorInHeartbeatTriggersFailover() throws Exception { } } + @Test + void testFailTieringJobWhenReadersDoNotFinishAfterMaxDuration() throws Throwable { + TablePath tablePath = TablePath.of(DEFAULT_DB, "tiering-completion-timeout-test"); + long tableId = createTable(tablePath, DEFAULT_LOG_TABLE_DESCRIPTOR); + Duration completionTimeout = Duration.ofMillis(100); + conn.getAdmin() + .alterTable( + tablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(), "10min")), + false) + .get(); + + appendRow(tablePath, DEFAULT_LOG_TABLE_DESCRIPTOR, 0, 10); + + AtomicReference coordinatorError = new AtomicReference<>(); + try (FlussMockSplitEnumeratorContext context = + new FlussMockSplitEnumeratorContext(1) { + @Override + public void runInCoordinatorThread(Runnable runnable) { + try { + runnable.run(); + } catch (Throwable t) { + coordinatorError.compareAndSet(null, t); + } + } + }; + TieringSourceEnumerator enumerator = + createTieringSourceEnumerator(flussConf, context)) { + enumerator.start(); + registerSingleReaderAndHandleSplitRequests(context, enumerator, 0, 0); + + assertThat(context.getSplitsAssignmentSequence()).isNotEmpty(); + Long tieringEpoch = enumerator.getTieringEpoch(tableId); + assertThat(tieringEpoch).isNotNull(); + enumerator.handleTableTieringReachMaxDuration( + tablePath, tableId, tieringEpoch, completionTimeout.toMillis()); + retry( + Duration.ofSeconds(30), + () -> + assertThat(coordinatorError.get()) + .isInstanceOf(FlinkRuntimeException.class) + .hasMessageContaining( + "did not finish within " + + completionTimeout.toMillis() + + " ms after reaching max duration")); + } + } + @Test void testTableReachMaxTieringDuration() throws Throwable { TablePath tablePath = TablePath.of(DEFAULT_DB, "tiering-max-duration-test-log-table"); long tableId = createTable(tablePath, DEFAULT_LOG_TABLE_DESCRIPTOR); int numSubtasks = 2; + Duration maxTieringDuration = Duration.ofMinutes(10); + conn.getAdmin() + .alterTable( + tablePath, + Collections.singletonList( + TableChange.set( + ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(), "10min")), + false) + .get(); try (FlussMockSplitEnumeratorContext context = new FlussMockSplitEnumeratorContext<>(numSubtasks); @@ -855,6 +915,11 @@ void testTableReachMaxTieringDuration() throws Throwable { // Wait for initial assignment waitUntilTieringTableSplitAssignmentReady(context, 2, 200L); + Long firstTieringEpoch = enumerator.getTieringEpoch(tableId); + assertThat(firstTieringEpoch).isNotNull(); + enumerator.handleTableTieringReachMaxDuration( + tablePath, tableId, firstTieringEpoch, maxTieringDuration.toMillis()); + retry( Duration.ofSeconds(30), () -> { @@ -885,17 +950,6 @@ void testTableReachMaxTieringDuration() throws Throwable { assertThat(assignedSplits).hasSize(1); assertThat(assignedSplits.get(0).shouldSkipCurrentRound()).isTrue(); - // alter table freshness to 10 min to make sure we won't assign in - // normal finish - conn.getAdmin() - .alterTable( - tablePath, - Collections.singletonList( - TableChange.set( - ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(), "10min")), - false) - .get(); - // Mock tiering finished // This simulates the reader finishing tiering after reaching max duration enumerator.handleSourceEvent(0, new FinishedTieringEvent(tableId)); @@ -935,6 +989,15 @@ void testTableReachMaxTieringDuration() throws Throwable { split -> split.getTableBucket().getTableId() == tableId && !split.shouldSkipCurrentRound()); + + assertThatCode( + () -> + enumerator.failTieringJobIfNotFinished( + tablePath, + tableId, + firstTieringEpoch, + maxTieringDuration.toMillis())) + .doesNotThrowAnyException(); } } }