Skip to content

[tiering] fail tiering job when readers exceed max duration #3810

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zuston wants to merge 1 commit into
base: main
Choose a base branch
from
+111 −14
Open

[tiering] fail tiering job when readers exceed max duration #3810

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 37 additions & 3 deletions ...c/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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<Long, Long, TablePath> tieringTable, Throwable throwable) {
Expand Down Expand Up @@ -476,6 +507,8 @@ private void generateTieringSplits(Tuple3<Long, Long, TablePath> tieringTable)
finishedTables.put(tieringTable.f0, TieringFinishInfo.from(tieringTable.f1));
} else {
pendingSplits.addAll(tieringSplits);
long maxTieringDurationMs =
tableInfo.getTableConfig().getDataLakeFreshness().toMillis();

timerService.schedule(
() ->
Expand All @@ -484,10 +517,11 @@ private void generateTieringSplits(Tuple3<Long, Long, TablePath> 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) {
Expand Down
85 changes: 74 additions & 11 deletions ...st/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}. */
Expand Down Expand Up @@ -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<Throwable> coordinatorError = new AtomicReference<>();
try (FlussMockSplitEnumeratorContext<TieringSplit> context =
new FlussMockSplitEnumeratorContext<TieringSplit>(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<TieringSplit> context =
new FlussMockSplitEnumeratorContext<>(numSubtasks);
Expand All @@ -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),
() -> {
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -935,6 +989,15 @@ void testTableReachMaxTieringDuration() throws Throwable {
split ->
split.getTableBucket().getTableId() == tableId
&& !split.shouldSkipCurrentRound());

assertThatCode(
() ->
enumerator.failTieringJobIfNotFinished(
tablePath,
tableId,
firstTieringEpoch,
maxTieringDuration.toMillis()))
.doesNotThrowAnyException();
}
}
}
Loading