You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "luoyuxia (via GitHub)" <gi...@apache.org> on 2025/12/16 09:23:55 UTC
[PR] Tiering support commit by time [fluss]
luoyuxia opened a new pull request, #2185:
URL: https://github.com/apache/fluss/pull/2185
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2711333093
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java:
##########
@@ -61,14 +64,17 @@ public class TieringSource<WriteResult>
private final Configuration flussConf;
private final LakeTieringFactory<WriteResult, ?> lakeTieringFactory;
private final long pollTieringTableIntervalMs;
+ private final long tieringTableDurationMaxMs;
public TieringSource(
Configuration flussConf,
LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
- long pollTieringTableIntervalMs) {
+ long pollTieringTableIntervalMs,
+ long tieringTableDurationMaxMs) {
this.flussConf = flussConf;
this.lakeTieringFactory = lakeTieringFactory;
this.pollTieringTableIntervalMs = pollTieringTableIntervalMs;
+ this.tieringTableDurationMaxMs = tieringTableDurationMaxMs;
Review Comment:
Constructor parameter `tieringTableDurationMaxMs` has an inconsistent naming convention compared to the builder method `withTieringTableMaxDurationMs`. The constructor uses `tieringTableDurationMaxMs` while the builder uses `tieringTableMaxDurationMs` (note the different word ordering). This inconsistency can cause confusion. Standardize the naming to either `tieringTableDurationMaxMs` or `tieringTableMaxDurationMs` throughout.
```suggestion
long tieringTableMaxDurationMs) {
this.flussConf = flussConf;
this.lakeTieringFactory = lakeTieringFactory;
this.pollTieringTableIntervalMs = pollTieringTableIntervalMs;
this.tieringTableDurationMaxMs = tieringTableMaxDurationMs;
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -92,38 +100,54 @@ public class TieringSplitReader<WriteResult>
// map from table bucket to split id
private final Map<TableBucket, TieringSplit> currentTableSplitsByBucket;
private final Map<TableBucket, Long> currentTableStoppingOffsets;
- private final Set<TieringLogSplit> currentTableEmptyLogSplits;
+
+ private final Map<TableBucket, LogOffsetAndTimestamp> currentTableTieredOffsetAndTimestamp;
+
+ private final Set<TieringSplit> currentEmptySplits;
public TieringSplitReader(
Connection connection, LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
+ this(connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
+ }
+
+ protected TieringSplitReader(
+ Connection connection,
Review Comment:
The visibility change from private to protected and the addition of a constructor parameter reduces encapsulation. The `pollTimeout` field is only used for testing purposes but now becomes part of the production API surface. Consider keeping the production constructor as-is and using a test-specific subclass or factory method instead.
```suggestion
/**
* Public constructor used in production, relying on the default poll timeout.
*/
public TieringSplitReader(
Connection connection, LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
this(connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
}
/**
* Package-private factory intended for testing to allow custom poll timeout configuration.
*/
static <W> TieringSplitReader<W> createForTesting(
Connection connection,
LakeTieringFactory<W, ?> lakeTieringFactory,
Duration pollTimeout) {
return new TieringSplitReader<>(connection, lakeTieringFactory, pollTimeout);
}
private TieringSplitReader(
Connection connection,
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.source;
+
+import org.apache.fluss.flink.adapter.SingleThreadFetcherManagerAdapter;
+import org.apache.fluss.flink.tiering.source.split.TieringSplit;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher;
+import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherTask;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+
+import java.util.Collection;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/**
+ * The SplitFetcherManager for tiering source. This class is needed to help notify a table reaches
+ * to deadline of tiering to {@link TieringSplitReader}.
Review Comment:
The comment says "reaches to deadline" but should be "reaches the deadline" (remove "to"). This is a grammatical error.
```suggestion
* the deadline of tiering to {@link TieringSplitReader}.
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -248,6 +285,58 @@ private void mayCreateLogScanner() {
}
}
+ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>
+ forceCompleteTieringLogRecords() throws IOException {
+ Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
+ Map<TableBucket, String> finishedSplitIds = new HashMap<>();
+
+ // force finish all splits
+ Iterator<Map.Entry<TableBucket, TieringSplit>> currentTieringSplitsIterator =
+ currentTableSplitsByBucket.entrySet().iterator();
+ while (currentTieringSplitsIterator.hasNext()) {
+ Map.Entry<TableBucket, TieringSplit> entry = currentTieringSplitsIterator.next();
+ TableBucket bucket = entry.getKey();
+ TieringSplit split = entry.getValue();
+ if (split != null && split.isTieringLogSplit()) {
+ // get the current offset, timestamp that tiered so far
+ LogOffsetAndTimestamp logOffsetAndTimestamp =
+ currentTableTieredOffsetAndTimestamp.get(bucket);
+ long logEndOffset =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_OFFSET
+ // logEndOffset is equal to offset tiered + 1
+ : logOffsetAndTimestamp.logOffset + 1;
+ long timestamp =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_TIMESTAMP
+ : logOffsetAndTimestamp.timestamp;
+ TableBucketWriteResult<WriteResult> bucketWriteResult =
+ completeLakeWriter(
+ bucket, split.getPartitionName(), logEndOffset, timestamp);
+
+ if (logEndOffset == UNKNOWN_BUCKET_OFFSET) {
+ // when the log end offset is unknown, the write result must be
+ // null, otherwise, we should throw exception directly to avoid data
+ // inconsistent
+ checkState(
+ bucketWriteResult.writeResult() == null,
+ "bucketWriteResult must be null when log end offset is unknown when tiering "
+ + split);
Review Comment:
The state check requires that `bucketWriteResult.writeResult()` must be null when `logEndOffset == UNKNOWN_BUCKET_OFFSET`, but this assumes the `completeLakeWriter` method always returns null when no data has been written. If there's any edge case where a writer returns a non-null result even when no offset has been recorded, this will throw an exception. Consider adding more defensive validation or clearer documentation on the contract of `completeLakeWriter`.
```suggestion
// When the log end offset is unknown, we expect the write result to be null.
// If it's not, log a warning instead of failing hard to avoid unexpected
// runtime exceptions due to edge cases in completeLakeWriter.
if (bucketWriteResult.writeResult() != null) {
LOG.warn(
"bucketWriteResult has non-null writeResult while log end offset is "
+ "unknown when tiering {}",
split);
}
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -339,30 +438,29 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
checkNotNull(currentTableNumberOfSplits));
}
- private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringLogSplit> emptySplits) {
+ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringSplit> emptySplits) {
Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
Map<TableBucket, String> finishedSplitIds = new HashMap<>();
- for (TieringLogSplit logSplit : emptySplits) {
- TableBucket tableBucket = logSplit.getTableBucket();
- finishedSplitIds.put(tableBucket, logSplit.splitId());
+ for (TieringSplit tieringSplit : emptySplits) {
+ TableBucket tableBucket = tieringSplit.getTableBucket();
+ finishedSplitIds.put(tableBucket, tieringSplit.splitId());
writeResults.put(
tableBucket,
toTableBucketWriteResult(
- logSplit.getTablePath(),
+ tieringSplit.getTablePath(),
tableBucket,
- logSplit.getPartitionName(),
+ tieringSplit.getPartitionName(),
null,
- logSplit.getStoppingOffset(),
+ UNKNOWN_BUCKET_OFFSET,
Review Comment:
The method returns an offset of UNKNOWN_BUCKET_OFFSET for empty splits, but in the normal log split flow (line 451-452), it uses `logSplit.getStoppingOffset()` instead. This inconsistency could cause confusion. For consistency and correctness, empty splits should also use UNKNOWN_BUCKET_OFFSET, which they now do - but the comment on line 451-452 should clarify why this is different from the regular stopping offset used before this change.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -298,24 +342,21 @@ private void generateAndAssignSplits(
}
private void assignSplits() {
- // we don't assign splits during failovering
+ // we don't assign splits during failover
if (isFailOvering) {
return;
}
- /* This method may be called from both addSplitsBack and handleSplitRequest, make it thread safe. */
- synchronized (readersAwaitingSplit) {
- if (!readersAwaitingSplit.isEmpty()) {
- final Integer[] readers = readersAwaitingSplit.toArray(new Integer[0]);
- for (Integer nextAwaitingReader : readers) {
- if (!context.registeredReaders().containsKey(nextAwaitingReader)) {
- readersAwaitingSplit.remove(nextAwaitingReader);
- continue;
- }
- if (!pendingSplits.isEmpty()) {
- TieringSplit tieringSplit = pendingSplits.remove(0);
- context.assignSplit(tieringSplit, nextAwaitingReader);
- readersAwaitingSplit.remove(nextAwaitingReader);
- }
+ if (!readersAwaitingSplit.isEmpty()) {
+ final Integer[] readers = readersAwaitingSplit.toArray(new Integer[0]);
+ for (Integer nextAwaitingReader : readers) {
+ if (!context.registeredReaders().containsKey(nextAwaitingReader)) {
+ readersAwaitingSplit.remove(nextAwaitingReader);
+ continue;
+ }
+ if (!pendingSplits.isEmpty()) {
+ TieringSplit tieringSplit = pendingSplits.remove(0);
+ context.assignSplit(tieringSplit, nextAwaitingReader);
+ readersAwaitingSplit.remove(nextAwaitingReader);
}
Review Comment:
The removed synchronization on `readersAwaitingSplit` (previously on line 348) could lead to race conditions. While the collections are now synchronized (lines 132-133), the compound operations (checking if empty, iterating, removing) are not atomic. Multiple threads could attempt to assign splits concurrently, potentially causing issues. The synchronization should be restored, or the logic should be restructured to ensure thread safety of the entire assignment operation.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -113,13 +121,18 @@ public class TieringSourceEnumerator
public TieringSourceEnumerator(
Configuration flussConf,
SplitEnumeratorContext<TieringSplit> context,
- long pollTieringTableIntervalMs) {
+ long pollTieringTableIntervalMs,
+ long tieringTableMaxDurationMs) {
this.flussConf = flussConf;
this.context = context;
+ this.timerService =
Review Comment:
The `timerService` field is created but never properly shutdown in all error paths. If an exception occurs during `start()` or other operations before `close()` is called, the timer service thread will leak. Consider using try-with-resources or ensuring cleanup in a finally block, or handling shutdown more defensively in the close method.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -430,17 +533,29 @@ private void finishCurrentTable() throws IOException {
} catch (Exception e) {
throw new IOException("Fail to finish current table.", e);
}
-
+ reachTieringMaxDurationTables.remove(currentTableId);
// before switch to a new table, mark all as empty or null
currentTableId = null;
currentTablePath = null;
currentTableNumberOfSplits = null;
currentPendingSnapshotSplits.clear();
currentTableStoppingOffsets.clear();
- currentTableEmptyLogSplits.clear();
+ currentTableTieredOffsetAndTimestamp.clear();
currentTableSplitsByBucket.clear();
}
+ /**
+ * Handle a table reach max tiering duration. This will mark the current table as reaching max
+ * duration, and it will be force completed in the next fetch cycle.
+ */
+ public void handleTableReachTieringMaxDuration(long tableId) {
+ if ((currentTableId != null && currentTableId.equals(tableId))
+ || pendingTieringSplits.containsKey(tableId)) {
Review Comment:
The method `handleTableReachTieringMaxDuration` adds the table to `reachTieringMaxDurationTables` even if the table is only in `pendingTieringSplits` and not yet the current table. However, the force completion logic in `forceCompleteTieringLogRecords()` (line 160) only checks if the current table is in the set. This means if a table reaches max duration before it becomes the current table, the flag will be set but never acted upon until that table becomes current. This could lead to memory leaks in the set or unexpected behavior. Consider only adding to the set when it's the current table, or handle pending tables differently.
```suggestion
if (currentTableId != null && currentTableId.equals(tableId)) {
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -391,6 +435,15 @@ private void generateTieringSplits(Tuple3<Long, Long, TablePath> tieringTable)
} else {
tieringTableEpochs.put(tieringTable.f0, tieringTable.f1);
pendingSplits.addAll(tieringSplits);
+
+ timerService.schedule(
+ () ->
+ context.runInCoordinatorThread(
+ () ->
+ handleTableTieringReachMaxDuration(
+ tieringTable.f0, tieringTable.f1)),
Review Comment:
The timer scheduled on line 437-444 is not cancelled when a table tiering completes normally or fails. If a table finishes tiering before the max duration, the timer will still fire and execute the completion logic, potentially causing unexpected behavior or wasted resources. Consider tracking and cancelling these timers when the table tiering completes (in `handleSourceEvent` for `FinishedTieringEvent` and `FailedTieringEvent`).
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSourceReaderTest.java:
##########
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.source;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
+import org.apache.fluss.flink.tiering.TestingWriteResult;
+import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
+import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TieringSourceReader}. */
+class TieringSourceReaderTest extends FlinkTestBase {
+
+ @Test
+ void testHandleTieringReachMaxDurationEvent() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "test_tiering_reach_max_duration");
+ long tableId = createTable(tablePath, DEFAULT_LOG_TABLE_DESCRIPTOR);
+ Configuration conf = new Configuration(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ conf.set(
+ ConfigOptions.CLIENT_WRITER_BUCKET_NO_KEY_ASSIGNER,
+ ConfigOptions.NoKeyAssigner.ROUND_ROBIN);
+ try (Connection connection = ConnectionFactory.createConnection(conf)) {
+ FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>
+ elementsQueue = new FutureCompletingBlockingQueue<>(16);
+ TestingReaderContext readerContext = new TestingReaderContext();
+ try (TieringSourceReader<TestingWriteResult> reader =
+ new TieringSourceReader<>(
+ elementsQueue,
+ readerContext,
+ connection,
+ new TestingLakeTieringFactory(),
+ Duration.ofMillis(500))) {
+
+ reader.start();
+
+ // no data, add a split for the table,
+ // should be force be complete after reach max duration
+ TieringLogSplit split =
+ new TieringLogSplit(
+ tablePath, new TableBucket(tableId, 0), null, EARLIEST_OFFSET, 100);
+ reader.addSplits(Collections.singletonList(split));
+
+ // send TieringReachMaxDurationEvent
+ TieringReachMaxDurationEvent event = new TieringReachMaxDurationEvent(tableId);
+ reader.handleSourceEvents(event);
+
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ TestingReaderOutput<TableBucketWriteResult<TestingWriteResult>> output =
+ new TestingReaderOutput<>();
+ // should force to finish, the write result is null
+ reader.pollNext(output);
+ assertThat(output.getEmittedRecords()).hasSize(1);
+ TableBucketWriteResult<TestingWriteResult> result =
+ output.getEmittedRecords().get(0);
+ assertThat(result.writeResult()).isNull();
+ });
+
+ // write some data
+ writeRows(
+ connection,
+ tablePath,
+ Arrays.asList(row(0, "v0"), row(1, "v1"), row(2, "v2")),
+ true);
+ split =
+ new TieringLogSplit(
+ tablePath,
+ new TableBucket(tableId, 2),
+ null,
+ EARLIEST_OFFSET,
+ // use 100L as end offset, so that
+ // tiering won't be finished if no tiering reach max duration logic
+ 100L);
+
+ reader.addSplits(Collections.singletonList(split));
+
+ // wait to run one round of tiering to do some tiering
+ FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>
+ blockingQueue = getElementsQueue(reader);
+ // wait blockingQueue is not empty to make sure we have one fetch
+ // in tiering source reader
+ waitUntil(
+ () -> !blockingQueue.isEmpty(),
+ Duration.ofSeconds(30),
+ "Fail to wait element queue is not empty.");
+
+ // send TieringReachMaxDurationEvent
+ event = new TieringReachMaxDurationEvent(tableId);
+ reader.handleSourceEvents(event);
+
+ // make sure tiering will be finished, still maintain the result
+ // of previous tiering
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ TestingReaderOutput<TableBucketWriteResult<TestingWriteResult>>
+ output1 = new TestingReaderOutput<>();
+
+ // should force to finish, the write result isn't null
+ reader.pollNext(output1);
+ assertThat(output1.getEmittedRecords()).hasSize(1);
+ TableBucketWriteResult<TestingWriteResult> result =
+ output1.getEmittedRecords().get(0);
+ TestingWriteResult testingWriteResult = result.writeResult();
+ assertThat(testingWriteResult).isNotNull();
+ assertThat(result.logEndOffset()).isEqualTo(1);
+ });
+
+ // test add split with skipCurrentRound
+ split =
+ new TieringLogSplit(
+ tablePath,
+ new TableBucket(tableId, 1),
+ null,
+ EARLIEST_OFFSET,
+ 100L);
+ split.skipCurrentRound();
+ reader.addSplits(Collections.singletonList(split));
+ // should skip tiering for this split
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ TestingReaderOutput<TableBucketWriteResult<TestingWriteResult>>
+ output1 = new TestingReaderOutput<>();
+ // should force to finish, and the result is null
+ reader.pollNext(output1);
+ assertThat(output1.getEmittedRecords()).hasSize(1);
+ TableBucketWriteResult<TestingWriteResult> result =
+ output1.getEmittedRecords().get(0);
+ assertThat(result.writeResult()).isNull();
+ });
+ }
+ }
+ }
+
+ /**
+ * Get the elementsQueue from TieringSourceReader using reflection.
+ *
+ * @param reader the TieringSourceReader instance
+ * @return the elementsQueue field value
+ */
+ @SuppressWarnings("unchecked")
+ private FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>
+ getElementsQueue(TieringSourceReader<TestingWriteResult> reader) throws Exception {
+ Class<?> clazz = reader.getClass();
+ while (clazz != null) {
+ try {
+ Field elementsQueueField = clazz.getDeclaredField("elementsQueue");
+ elementsQueueField.setAccessible(true);
+ return (FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>)
+ elementsQueueField.get(reader);
+ } catch (NoSuchFieldException e) {
+ // Try parent class
+ clazz = clazz.getSuperclass();
+ }
+ }
+ throw new RuntimeException("No elementsQueue field found");
Review Comment:
Using reflection to access private fields from parent classes is fragile and breaks encapsulation. If the parent class structure changes or the field is renamed, this will fail at runtime. Consider adding a protected getter in the parent class or restructuring to avoid the need for reflection.
```suggestion
Field elementsQueueField = TieringSourceReader.class.getDeclaredField("elementsQueue");
elementsQueueField.setAccessible(true);
return (FutureCompletingBlockingQueue<
RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>)
elementsQueueField.get(reader);
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2727183519
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java:
##########
@@ -83,19 +83,22 @@
* ▼ ▼
* ┌──────────┐ (lake freshness > tiering interval)
* │Scheduled ├──────┐
- * └──────────┘ ▼
- * ▲ ┌───────┐ (assign to tier service) ┌───────┐
- * | |Pending├──────────────────────────►|Tiering├─┐
- * | └───────┘ └───┬───┘ │
- * | ▲ ┌─────────────────┘ │
- * | | | (timeout or failure) | (finished)
- * | | ▼ ▼
- * | | (retry) ┌─────────┐ ┌────────┐
- * | └────────────│ Failed │ │ Tiered │
- * | └─────────┘ └────┬───┘
- * | |
- * └──────────────────────────────────────────────────────┘
- * (ready for next round of tiering)
+ * └─────▲────┘ ▼
Review Comment:
```java
* <li>Tiered: when the lake table finish(including force finished) one round of tiering
* <li>Failed: when the lake table tiering failed
* </ul>
*
* <p>The state machine of table to be tiered is as follows:
*
* <pre>{@code
* ┌─────┐ ┌──────┐
* │ New │ │ Init │
* └──┬──┘ └──┬───┘
* ▼ ▼
* ┌──────────┐ (lake freshness > tiering interval)
* │Scheduled ├─────┐
* └─────▲────┘ │
* │ ┌───▼───┐ (assign to tier service) ┌───────┐
* │ |Pending├──────────────────────────►│Tiering├─┐
* │ └───▲───┘ └───┬───┘ │
* │ │ ┌─────────────────┘ │
* │ │ │ (timeout or failure) │ (finished)
* │ │ ▼ ▼
* │ │ (retry) ┌─────────┐ ┌────────┐
* │ │◀───────────│ Failed │ │ Tiered │
* │ │ └─────────┘ └─┬───┬──┘
* │ │ │ │
* │ │ (force finished) │ │
* │ └─────────────────────────────────────-┘ │
* │ (ready for next round) │
* └─────────────────────────────────────────────────────┘
* }</pre>
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #2185:
URL: https://github.com/apache/fluss/pull/2185
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3691399521
@leonardBang Coud you please help me review this pr when you got some time
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2646602752
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -236,6 +277,53 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
}
}
+ private Set<Long> checkTableReachMaxTieringDuration() {
+ Set<Long> tieringReachMaxDurationTables = new HashSet<>();
+ long currentTime = clock.milliseconds();
+ for (Map.Entry<Long, Long> tieringTableDeadline : tieringTablesDeadline.entrySet()) {
+ long tableId = tieringTableDeadline.getKey();
+ long deadline = tieringTableDeadline.getValue();
+ if (deadline < currentTime) {
+ tieringReachMaxDurationTables.add(tableId);
+ }
+ }
+ return tieringReachMaxDurationTables;
+ }
+
+ private void handleReachMaxTieringDurationTables(
+ Set<Long> tieringReachMaxDurationTables, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("Fail to check tiering timeout tables.", throwable);
+ return;
+ }
+
+ for (Long reachMaxDurationTable : tieringReachMaxDurationTables) {
+ for (TieringSplit tieringSplit : pendingSplits) {
+ if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
+ // force ignore this tiering split since the tiering for this table is timeout,
+ // we have to force to set to ignore the tiering split so that the
+ // tiering source reader can ignore them directly
+ tieringSplit.forceIgnore();
+ } else {
+ // we can break directly, if found any one split's table id is not equal to the
+ // timeout
+ // table, the following split must be not equal to the table id
+ break;
Review Comment:
The shuffle is based on table, won't cause different tables to be interleaved.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2672046461
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java:
##########
@@ -38,13 +38,15 @@ public abstract class TieringSplit implements SourceSplit {
protected final TableBucket tableBucket;
@Nullable protected final String partitionName;
+ protected boolean forceIgnore;
// the total number of splits in one round of tiering
protected final int numberOfSplits;
public TieringSplit(
TablePath tablePath,
TableBucket tableBucket,
@Nullable String partitionName,
+ boolean forceIgnore,
int numberOfSplits) {
Review Comment:
```suggestion
int numberOfSplits,
boolean forceIgnore) {
```
Add new parameter to last position is recommended
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java:
##########
@@ -128,11 +139,12 @@ public boolean equals(Object object) {
return Objects.equals(tablePath, that.tablePath)
&& Objects.equals(tableBucket, that.tableBucket)
&& Objects.equals(partitionName, that.partitionName)
+ && forceIgnore == that.forceIgnore
&& numberOfSplits == that.numberOfSplits;
}
@Override
public int hashCode() {
- return Objects.hash(tablePath, tableBucket, partitionName, numberOfSplits);
+ return Objects.hash(tablePath, tableBucket, partitionName, forceIgnore, numberOfSplits);
Review Comment:
minor: add to last position.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java:
##########
@@ -126,6 +129,8 @@ public TieringSplit deserialize(int version, byte[] serialized) throws IOExcepti
}
TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
+ boolean forceIgnore = in.readBoolean();
Review Comment:
hint: we can add a note we do not need consider compatibility here.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java:
##########
@@ -54,6 +56,7 @@ public TieringSplit(
throw new IllegalArgumentException(
"Partition name and partition id must be both null or both not null.");
}
+ this.forceIgnore = forceIgnore;
Review Comment:
`forceIgnore` is not clear especially with any explanation, here we want to use a member to represent `how to read` of TieringSplit, IIUC, the meaning of this `status` is skip reading data for current split, how about `skipCurrentRound` and add java doc for method or member?
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -1811,6 +1811,23 @@ public class ConfigOptions {
+ ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT
+ " is false.");
+ public static final ConfigOption<Duration> LAKE_TIERING_TABLE_DURATION_MAX =
+ key("lake.tiering.table.duration.max")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(30))
+ .withDescription(
+ "The maximum duration for tiering a single table. If tiering a table exceeds this duration, "
+ + "it will be force completed: the tiering will be finalized and committed to the data lake "
Review Comment:
after the table wast force completed, would it be tired again for next round ?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3799226508
> Thanks @luoyuxia for the update, removing this option can simplify user interfaces, +1 for the update, I just left one comment for the state machine.
Thanks for the comment. It looks much better.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2727183519
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java:
##########
@@ -83,19 +83,22 @@
* ▼ ▼
* ┌──────────┐ (lake freshness > tiering interval)
* │Scheduled ├──────┐
- * └──────────┘ ▼
- * ▲ ┌───────┐ (assign to tier service) ┌───────┐
- * | |Pending├──────────────────────────►|Tiering├─┐
- * | └───────┘ └───┬───┘ │
- * | ▲ ┌─────────────────┘ │
- * | | | (timeout or failure) | (finished)
- * | | ▼ ▼
- * | | (retry) ┌─────────┐ ┌────────┐
- * | └────────────│ Failed │ │ Tiered │
- * | └─────────┘ └────┬───┘
- * | |
- * └──────────────────────────────────────────────────────┘
- * (ready for next round of tiering)
+ * └─────▲────┘ ▼
Review Comment:
```java
* <li>Tiered: when the lake table finish(including forced finish) one round of tiering
* <li>Failed: when the lake table tiering failed
* </ul>
*
* <p>The state machine of table to be tiered is as follows:
*
* <pre>{@code
* ┌─────┐ ┌──────┐
* │ New │ │ Init │
* └──┬──┘ └──┬───┘
* ▼ ▼
* ┌──────────┐ (lake freshness > tiering interval)
* │Scheduled ├─────┐
* └─────▲────┘ │
* │ ┌───▼───┐ (assign to tier service) ┌───────┐
* │ |Pending├──────────────────────────►│Tiering├─┐
* │ └───▲───┘ └───┬───┘ │
* │ │ ┌─────────────────┘ │
* │ │ │ (timeout or failure) │ (finished)
* │ │ ▼ ▼
* │ │ (retry) ┌─────────┐ ┌────────┐
* │ │◀───────────│ Failed │ │ Tiered │
* │ │ └─────────┘ └─┬───┬──┘
* │ │ │ │
* │ │ (force finished) │ │
* │ └─────────────────────────────────────-┘ │
* │ (ready for next round) │
* └─────────────────────────────────────────────────────┘
* }</pre>
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3783852859
@leonardBang After discuss with Jark:
- I remove the option for tiering server job and use the data freshness as the max tiering duration for a table just for simplicity
- When a table is forced to finish due to reach max duration, it should put into pending queue without waiting schedule to pending
I already verified it in gray cluster.
Please help review again when you got some time.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2646896556
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -248,6 +285,58 @@ private void mayCreateLogScanner() {
}
}
+ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>
+ forceCompleteTieringLogRecords() throws IOException {
+ Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
+ Map<TableBucket, String> finishedSplitIds = new HashMap<>();
+
+ // force finish all splits
+ Iterator<Map.Entry<TableBucket, TieringSplit>> currentTieringSplitsIterator =
+ currentTableSplitsByBucket.entrySet().iterator();
+ while (currentTieringSplitsIterator.hasNext()) {
+ Map.Entry<TableBucket, TieringSplit> entry = currentTieringSplitsIterator.next();
+ TableBucket bucket = entry.getKey();
+ TieringSplit split = entry.getValue();
+ if (split != null && split.isTieringLogSplit()) {
+ // get the current offset, timestamp that tiered so far
+ LogOffsetAndTimestamp logOffsetAndTimestamp =
+ currentTableTieredOffsetAndTimestamp.get(bucket);
+ long logEndOffset =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_OFFSET
+ // logEngOffset is equal to offset tiered + 1
+ : logOffsetAndTimestamp.logOffset + 1;
+ long timestamp =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_TIMESTAMP
+ : logOffsetAndTimestamp.timestamp;
+ TableBucketWriteResult<WriteResult> bucketWriteResult =
+ completeLakeWriter(
+ bucket, split.getPartitionName(), logEndOffset, timestamp);
+
+ if (logEndOffset == UNKNOWN_BUCKET_OFFSET) {
+ // when the log end offset is unknow, the write result must be
Review Comment:
Spelling error: "unknow" should be "unknown"
```suggestion
// when the log end offset is unknown, the write result must be
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -110,16 +121,38 @@ public class TieringSourceEnumerator
public TieringSourceEnumerator(
Configuration flussConf,
SplitEnumeratorContext<TieringSplit> context,
- long pollTieringTableIntervalMs) {
+ long pollTieringTableIntervalMs,
+ long tieringTableDurationMaxMs,
+ long tieringTableDurationDetectIntervalMs) {
+ this(
+ flussConf,
+ context,
+ pollTieringTableIntervalMs,
+ tieringTableDurationMaxMs,
+ tieringTableDurationDetectIntervalMs,
+ SystemClock.getInstance());
+ }
+
+ public TieringSourceEnumerator(
+ Configuration flussConf,
+ SplitEnumeratorContext<TieringSplit> context,
+ long pollTieringTableIntervalMs,
+ long tieringTableDurationMaxMs,
+ long tieringTableDurationDetectIntervalMs,
+ Clock clock) {
this.flussConf = flussConf;
this.context = context;
this.enumeratorMetricGroup = context.metricGroup();
this.pollTieringTableIntervalMs = pollTieringTableIntervalMs;
+ this.tieringTableDurationMaxMs = tieringTableDurationMaxMs;
+ this.tieringTableDurationDetectIntervalMs = tieringTableDurationDetectIntervalMs;
this.pendingSplits = new ArrayList<>();
Review Comment:
The pendingSplits list is not thread-safe (ArrayList), but it's being accessed from multiple async callbacks (handleReachMaxTieringDurationTables at line 301 and generateTieringSplits at line 431) and from assignSplits which has synchronization only on readersAwaitingSplit. This could lead to ConcurrentModificationException or other race conditions. Consider using a thread-safe collection like CopyOnWriteArrayList or adding proper synchronization around all accesses to pendingSplits.
```suggestion
this.pendingSplits = Collections.synchronizedList(new ArrayList<>());
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java:
##########
@@ -72,6 +75,14 @@ public final boolean isTieringLogSplit() {
return getClass() == TieringLogSplit.class;
}
+ public void forceIgnore() {
+ this.forceIgnore = true;
+ }
+
+ public boolean isForceIgnore() {
+ return forceIgnore;
+ }
Review Comment:
The forceIgnore field is being added to the TieringSplit class, but it's not included in the equals and hashCode methods (which are not shown in this diff but exist in the class). This means two splits that differ only in their forceIgnore flag will be considered equal, which could lead to incorrect behavior in collections like HashSet or HashMap. Consider adding forceIgnore to both equals and hashCode methods.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -249,6 +337,7 @@ private void generateAndAssignSplits(
private void assignSplits() {
/* This method may be called from both addSplitsBack and handleSplitRequest, make it thread safe. */
+ // todo: do we need to add lock?
Review Comment:
This TODO comment should either be resolved or removed. If locking is actually needed for thread safety, implement it; if not needed, remove the comment to avoid confusion.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -248,6 +285,58 @@ private void mayCreateLogScanner() {
}
}
+ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>
+ forceCompleteTieringLogRecords() throws IOException {
+ Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
+ Map<TableBucket, String> finishedSplitIds = new HashMap<>();
+
+ // force finish all splits
+ Iterator<Map.Entry<TableBucket, TieringSplit>> currentTieringSplitsIterator =
+ currentTableSplitsByBucket.entrySet().iterator();
+ while (currentTieringSplitsIterator.hasNext()) {
+ Map.Entry<TableBucket, TieringSplit> entry = currentTieringSplitsIterator.next();
+ TableBucket bucket = entry.getKey();
+ TieringSplit split = entry.getValue();
+ if (split != null && split.isTieringLogSplit()) {
+ // get the current offset, timestamp that tiered so far
+ LogOffsetAndTimestamp logOffsetAndTimestamp =
+ currentTableTieredOffsetAndTimestamp.get(bucket);
+ long logEndOffset =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_OFFSET
+ // logEngOffset is equal to offset tiered + 1
Review Comment:
Spelling/grammar error: "logEngOffset" should be "logEndOffset"
```suggestion
// logEndOffset is equal to offset tiered + 1
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSourceReaderTest.java:
##########
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.source;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
+import org.apache.fluss.flink.tiering.TestingWriteResult;
+import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
+import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TieringSourceReader}. */
+class TieringSourceReaderTest extends FlinkTestBase {
+
+ @Test
+ void testHandleTieringReachMaxDurationEvent() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "test_tiering_reach_max_duration");
+ long tableId = createTable(tablePath, DEFAULT_LOG_TABLE_DESCRIPTOR);
+ Configuration conf = new Configuration(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ conf.set(
+ ConfigOptions.CLIENT_WRITER_BUCKET_NO_KEY_ASSIGNER,
+ ConfigOptions.NoKeyAssigner.ROUND_ROBIN);
+ try (Connection connection = ConnectionFactory.createConnection(conf)) {
+ FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>
+ elementsQueue = new FutureCompletingBlockingQueue<>(16);
+ TestingReaderContext readerContext = new TestingReaderContext();
+ try (TieringSourceReader<TestingWriteResult> reader =
+ new TieringSourceReader<>(
+ elementsQueue,
+ readerContext,
+ connection,
+ new TestingLakeTieringFactory(),
+ Duration.ofMillis(500))) {
+
+ reader.start();
+
+ // no data, add a split for the table,
+ // should be force be complete after reach max duration
+ TieringLogSplit split =
+ new TieringLogSplit(
+ tablePath, new TableBucket(tableId, 0), null, EARLIEST_OFFSET, 100);
+ reader.addSplits(Collections.singletonList(split));
+
+ // send TieringReachMaxDurationEvent
+ TieringReachMaxDurationEvent event = new TieringReachMaxDurationEvent(tableId);
+ reader.handleSourceEvents(event);
+
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ TestingReaderOutput<TableBucketWriteResult<TestingWriteResult>> output =
+ new TestingReaderOutput<>();
+ // should force to finish, the write result is null
+ reader.pollNext(output);
+ assertThat(output.getEmittedRecords()).hasSize(1);
+ TableBucketWriteResult<TestingWriteResult> result =
+ output.getEmittedRecords().get(0);
+ assertThat(result.writeResult()).isNull();
+ });
+
+ // write some data
+ writeRows(
+ connection,
+ tablePath,
+ Arrays.asList(row(0, "v0"), row(1, "v1"), row(2, "v2")),
+ true);
+ split =
+ new TieringLogSplit(
+ tablePath,
+ new TableBucket(tableId, 2),
+ null,
+ EARLIEST_OFFSET,
+ // use 100L as end offset, so that
+ // tiering won't be finished if no tiering reach max duration logic
+ 100L);
+
+ reader.addSplits(Collections.singletonList(split));
+
+ // wait to run one round of tiering to do some tiering
+ FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>
+ blockingQueue = getElementsQueue(reader);
+ // wait blockingQueue is not empty to make sure we have one fetch
+ // in tiering source reader
+ waitUntil(
+ () -> !blockingQueue.isEmpty(),
+ Duration.ofSeconds(30),
+ "Fail to wait element queue is not empty.");
+
+ // send TieringReachMaxDurationEvent
+ event = new TieringReachMaxDurationEvent(tableId);
+ reader.handleSourceEvents(event);
+
+ // make sure tiering will be finished, still maintain the result
+ // of previous tiering
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ TestingReaderOutput<TableBucketWriteResult<TestingWriteResult>>
+ output1 = new TestingReaderOutput<>();
+
+ // should force to finish, the write result isn't null
+ reader.pollNext(output1);
+ assertThat(output1.getEmittedRecords()).hasSize(1);
+ TableBucketWriteResult<TestingWriteResult> result =
+ output1.getEmittedRecords().get(0);
+ TestingWriteResult testingWriteResult = result.writeResult();
+ assertThat(testingWriteResult).isNotNull();
+ assertThat(result.logEndOffset()).isEqualTo(1);
+ });
+
+ // test add split with force ignore
+ split =
+ new TieringLogSplit(
+ tablePath,
+ new TableBucket(tableId, 1),
+ null,
+ EARLIEST_OFFSET,
+ 100L);
+ split.forceIgnore();
+ reader.addSplits(Collections.singletonList(split));
+ // should skip tiering for this split
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ TestingReaderOutput<TableBucketWriteResult<TestingWriteResult>>
+ output1 = new TestingReaderOutput<>();
+ // should force to finish, and the result is null
+ reader.pollNext(output1);
+ assertThat(output1.getEmittedRecords()).hasSize(1);
+ TableBucketWriteResult<TestingWriteResult> result =
+ output1.getEmittedRecords().get(0);
+ assertThat(result.writeResult()).isNull();
+ });
+ }
+ }
+ }
+
+ /**
+ * Get the elementsQueue from TieringSourceReader using reflection.
+ *
+ * @param reader the TieringSourceReader instance
+ * @return the elementsQueue field value
+ */
+ @SuppressWarnings("unchecked")
+ private FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>
+ getElementsQueue(TieringSourceReader<TestingWriteResult> reader) throws Exception {
+ Class<?> clazz = reader.getClass();
+ while (clazz != null) {
+ try {
+ Field elementsQueueField = clazz.getDeclaredField("elementsQueue");
+ elementsQueueField.setAccessible(true);
+ return (FutureCompletingBlockingQueue<
+ RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>>>)
+ elementsQueueField.get(reader);
+ } catch (NoSuchFieldException e) {
+ // Try parent class
+ clazz = clazz.getSuperclass();
+ }
+ }
+ throw new RuntimeException("No elementsQueue field found");
+ }
Review Comment:
Using reflection to access private fields in tests can be brittle and may break if the implementation changes. Consider making the elementsQueue field accessible for testing purposes (e.g., using @VisibleForTesting annotation or providing a package-private getter) rather than relying on reflection.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2646596284
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -236,6 +277,53 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
}
}
+ private Set<Long> checkTableReachMaxTieringDuration() {
+ Set<Long> tieringReachMaxDurationTables = new HashSet<>();
+ long currentTime = clock.milliseconds();
+ for (Map.Entry<Long, Long> tieringTableDeadline : tieringTablesDeadline.entrySet()) {
+ long tableId = tieringTableDeadline.getKey();
+ long deadline = tieringTableDeadline.getValue();
+ if (deadline < currentTime) {
+ tieringReachMaxDurationTables.add(tableId);
+ }
+ }
+ return tieringReachMaxDurationTables;
+ }
+
+ private void handleReachMaxTieringDurationTables(
+ Set<Long> tieringReachMaxDurationTables, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("Fail to check tiering timeout tables.", throwable);
+ return;
+ }
+
+ for (Long reachMaxDurationTable : tieringReachMaxDurationTables) {
+ for (TieringSplit tieringSplit : pendingSplits) {
+ if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
+ // force ignore this tiering split since the tiering for this table is timeout,
+ // we have to force to set to ignore the tiering split so that the
+ // tiering source reader can ignore them directly
+ tieringSplit.forceIgnore();
+ } else {
+ // we can break directly, if found any one split's table id is not equal to the
+ // timeout
+ // table, the following split must be not equal to the table id
+ break;
Review Comment:
handleReachMaxTieringDurationTables,handleSplitRequest, addSplitsBack is done in a single thread
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3776588805
@leonardBang Thanks for your review. Comments are addressed. And I also remove the `lake.tiering.table.duration.detect-interval` to make code clean. PTAL
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3738427669
> > Thanks @luoyuxia for the work, I left some minor comment.
> > But I also have a question for current design, Could we address large splits case via current design? The data freshness would be worse for large log split from my understanding.
>
> Hi, thanks for reviewing my pr. Yes, large log split will harm data fressnes. But What do you mean `address large splits case`? In the current design, if we have a large split, and aussming the max tiering duration is 10 min, after 10 min, the large split will be forced to commit.
I mean for the large log table, the data freshness goal(i.e. 5minute) would be worse to 10 minutes, 15 minutes after some scheduled rounds, maybe in that case we need monitor these tables and scale up tiering service resources.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2646593891
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -236,6 +277,53 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
}
}
+ private Set<Long> checkTableReachMaxTieringDuration() {
+ Set<Long> tieringReachMaxDurationTables = new HashSet<>();
+ long currentTime = clock.milliseconds();
+ for (Map.Entry<Long, Long> tieringTableDeadline : tieringTablesDeadline.entrySet()) {
+ long tableId = tieringTableDeadline.getKey();
+ long deadline = tieringTableDeadline.getValue();
+ if (deadline < currentTime) {
+ tieringReachMaxDurationTables.add(tableId);
+ }
+ }
+ return tieringReachMaxDurationTables;
+ }
+
+ private void handleReachMaxTieringDurationTables(
+ Set<Long> tieringReachMaxDurationTables, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("Fail to check tiering timeout tables.", throwable);
+ return;
+ }
+
+ for (Long reachMaxDurationTable : tieringReachMaxDurationTables) {
+ for (TieringSplit tieringSplit : pendingSplits) {
+ if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
+ // force ignore this tiering split since the tiering for this table is timeout,
+ // we have to force to set to ignore the tiering split so that the
+ // tiering source reader can ignore them directly
+ tieringSplit.forceIgnore();
+ } else {
+ // we can break directly, if found any one split's table id is not equal to the
+ // timeout
+ // table, the following split must be not equal to the table id
+ break;
+ }
+ }
+
+ LOG.debug("Found the table {} reach max tiering duration.", reachMaxDurationTable);
+
+ // broadcast the tiering reach max duration event to all readers,
+ // we broadcast all for simplicity
+ Set<Integer> readers = new HashSet<>(context.registeredReaders().keySet());
+ for (int reader : readers) {
+ context.sendEventToSourceReader(
+ reader, new TieringReachMaxDurationEvent(reachMaxDurationTable));
+ }
Review Comment:
We expect it receive finish event to remove the timeout table
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2646600392
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -339,30 +417,29 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
checkNotNull(currentTableNumberOfSplits));
}
- private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringLogSplit> emptySplits) {
+ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringSplit> emptySplits) {
Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
Map<TableBucket, String> finishedSplitIds = new HashMap<>();
- for (TieringLogSplit logSplit : emptySplits) {
- TableBucket tableBucket = logSplit.getTableBucket();
- finishedSplitIds.put(tableBucket, logSplit.splitId());
+ for (TieringSplit tieringSplit : emptySplits) {
+ TableBucket tableBucket = tieringSplit.getTableBucket();
+ finishedSplitIds.put(tableBucket, tieringSplit.splitId());
writeResults.put(
tableBucket,
toTableBucketWriteResult(
- logSplit.getTablePath(),
+ tieringSplit.getTablePath(),
tableBucket,
- logSplit.getPartitionName(),
+ tieringSplit.getPartitionName(),
null,
- logSplit.getStoppingOffset(),
+ UNKNOW_BUCKET_OFFSET,
Review Comment:
when the split is force-ignored, the result will be null which upstream operator will just ignore it.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2646599259
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java:
##########
@@ -74,6 +74,9 @@ public byte[] serialize(TieringSplit split) throws IOException {
out.writeBoolean(false);
}
+ // write force ignore
+ out.writeBoolean(split.isForceIgnore());
Review Comment:
It doesn't matter since it's just used to do shuffle
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2694493537
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -134,7 +158,11 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
}
} else {
if (currentLogScanner != null) {
- ScanRecords scanRecords = currentLogScanner.poll(POLL_TIMEOUT);
+ // force to complete records
+ if (reachTieringMaxDurationTables.contains(currentTableId)) {
+ return forceCompleteTieringLogRecords();
Review Comment:
There is a potential race condition when checking if a table has reached max duration. The condition on line 162 checks if the currentTableId is in reachTieringMaxDurationTables, but this set can be modified by the TieringSourceFetcherManager from a different thread (via handleTableReachTieringMaxDuration). Since reachTieringMaxDurationTables is not thread-safe (it's a regular HashSet), this could lead to race conditions. The set should be synchronized or use a concurrent collection.
```suggestion
synchronized (reachTieringMaxDurationTables) {
if (reachTieringMaxDurationTables.contains(currentTableId)) {
return forceCompleteTieringLogRecords();
}
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java:
##########
@@ -138,8 +162,24 @@ public Builder<WriteResult> withPollTieringTableIntervalMs(long pollTieringTable
return this;
}
Review Comment:
The builder method withTieringTableDurationMax on line 165 accepts a long parameter representing milliseconds but the method name doesn't clearly indicate the time unit. This could lead to confusion if a developer passes a value in seconds instead of milliseconds. Consider renaming it to withTieringTableDurationMaxMs to match the pattern used in withPollTieringTableIntervalMs (line 160), or add JavaDoc to clarify the expected unit.
```suggestion
/**
* Sets the maximum tiering table duration.
*
* @param tieringTableDurationMaxMs maximum duration in <strong>milliseconds</strong>
* for tiering table data before it is tiered.
* @return this builder.
*/
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering;
+
+import org.apache.fluss.client.metadata.LakeSnapshot;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.ExceptionUtils;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL;
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_MAX;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The IT case for tiering. */
+class TieringITCase extends FlinkTieringTestBase {
+
+ @BeforeAll
+ protected static void beforeAll() {
+ FlinkTieringTestBase.beforeAll();
+ }
+
+ @AfterAll
+ protected static void afterAll() throws Exception {
+ FlinkTieringTestBase.afterAll();
+ }
+
+ @BeforeEach
+ void beforeEach() {
+ execEnv =
+ StreamExecutionEnvironment.getExecutionEnvironment()
+ .setParallelism(1)
+ .setRuntimeMode(RuntimeExecutionMode.STREAMING);
+ }
+
+ @Test
+ void testTieringReachMaxDuration() throws Exception {
+ TablePath logTablePath = TablePath.of("fluss", "logtable");
+ createTable(logTablePath, false);
+ TablePath pkTablePath = TablePath.of("fluss", "pktable");
+ long pkTableId = createTable(pkTablePath, true);
+
+ // write some records to log table
+ List<InternalRow> rows = new ArrayList<>();
+ int recordCount = 6;
+ for (int i = 0; i < recordCount; i++) {
+ rows.add(GenericRow.of(i, BinaryString.fromString("v" + i)));
+ }
+ writeRows(logTablePath, rows, true);
+
+ rows = new ArrayList<>();
+ // write 6 records to primary key table, each bucket should only contain few record
+ for (int i = 0; i < recordCount; i++) {
+ rows.add(GenericRow.of(i, BinaryString.fromString("v" + i)));
+ }
+ writeRows(pkTablePath, rows, false);
+
+ waitUntilSnapshot(pkTableId, 3, 0);
+
+ // set tiering duration to a small value for testing purpose
+ Configuration lakeTieringConfig = new Configuration();
+ lakeTieringConfig.set(LAKE_TIERING_TABLE_DURATION_MAX, Duration.ofSeconds(1));
+ lakeTieringConfig.set(LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL, Duration.ofMillis(100));
+ JobClient jobClient = buildTieringJob(execEnv, lakeTieringConfig);
+
+ try {
+ // verify the tiered records is less than the table total record to
+ // make sure tiering is forced to complete when reach max duration
+ LakeSnapshot logTableLakeSnapshot = waitLakeSnapshot(logTablePath);
+ long tieredRecords = countTieredRecords(logTableLakeSnapshot);
+ assertThat(tieredRecords).isLessThan(recordCount);
+
+ // verify the tiered records is less than the table total record to
+ // make sure tiering is forced to complete when reach max duration
+ LakeSnapshot pkTableLakeSnapshot = waitLakeSnapshot(pkTablePath);
+ tieredRecords = countTieredRecords(pkTableLakeSnapshot);
+ assertThat(tieredRecords).isLessThan(recordCount);
Review Comment:
The test assertion on lines 108 and 114 checks that tieredRecords is less than recordCount (6), which implicitly assumes that some but not all records will be tiered. However, this assumption is fragile and timing-dependent. Depending on the timing of when the max duration is reached, it's possible that all records could be tiered (if they're written fast enough) or none could be tiered (if the duration expires immediately). The test should either have more robust assertions or add additional synchronization to ensure the intended partial tiering behavior occurs.
```suggestion
// verify the tiered records do not exceed the table total record
// under the configured max duration
LakeSnapshot logTableLakeSnapshot = waitLakeSnapshot(logTablePath);
long tieredRecords = countTieredRecords(logTableLakeSnapshot);
assertThat(tieredRecords).isLessThanOrEqualTo(recordCount);
// verify the tiered records do not exceed the table total record
// under the configured max duration
LakeSnapshot pkTableLakeSnapshot = waitLakeSnapshot(pkTablePath);
tieredRecords = countTieredRecords(pkTableLakeSnapshot);
assertThat(tieredRecords).isLessThanOrEqualTo(recordCount);
```
##########
fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java:
##########
@@ -96,4 +115,154 @@ public TableDescriptor getTable(TablePath tablePath) {
return tableByPath.get(tablePath);
}
}
+
+ private static class TestingPaimonTieringFactory
+ implements LakeTieringFactory<TestingPaimonWriteResult, TestPaimonCommittable> {
+
+ @Override
+ public LakeWriter<TestingPaimonWriteResult> createLakeWriter(
+ WriterInitContext writerInitContext) {
+ return new TestingPaimonWriter(writerInitContext.tableInfo());
+ }
+
+ @Override
+ public SimpleVersionedSerializer<TestingPaimonWriteResult> getWriteResultSerializer() {
+ return new TestingPaimonWriteResultSerializer();
+ }
+
+ @Override
+ public LakeCommitter<TestingPaimonWriteResult, TestPaimonCommittable> createLakeCommitter(
+ CommitterInitContext committerInitContext) throws IOException {
+ return new TestingPaimonCommitter();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<TestPaimonCommittable> getCommittableSerializer() {
+ return new SimpleVersionedSerializer<TestPaimonCommittable>() {
+ @Override
+ public int getVersion() {
+ return 0;
+ }
+
+ @Override
+ public byte[] serialize(TestPaimonCommittable obj) throws IOException {
+ return new byte[0];
+ }
+
+ @Override
+ public TestPaimonCommittable deserialize(int version, byte[] serialized)
+ throws IOException {
+ return new TestPaimonCommittable();
+ }
+ };
+ }
+ }
+
+ private static class TestingPaimonWriter implements LakeWriter<TestingPaimonWriteResult> {
+
+ static ConfigOption<Duration> writePauseOption =
+ key("write-pause").durationType().noDefaultValue();
+
+ private int writtenRecords = 0;
+ private final Duration writePause;
+
+ private TestingPaimonWriter(TableInfo tableInfo) {
+ this.writePause = tableInfo.getCustomProperties().get(writePauseOption);
+ }
+
+ @Override
+ public void write(LogRecord record) throws IOException {
+ try {
+ if (writePause != null) {
+ Thread.sleep(writePause.toMillis());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while pausing before write", e);
+ }
+ writtenRecords += 1;
+ }
+
+ @Override
+ public TestingPaimonWriteResult complete() throws IOException {
+ return new TestingPaimonWriteResult(writtenRecords);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // do nothing
+ }
+ }
+
+ private static class TestingPaimonWriteResult {
+ private final int writtenRecords;
+
+ public TestingPaimonWriteResult(int writtenRecords) {
+ this.writtenRecords = writtenRecords;
+ }
Review Comment:
The TestingPaimonWriteResult class is missing a getter method for the writtenRecords field. This makes the field inaccessible to other components that might need to verify or use this data, which could be necessary for testing purposes.
```suggestion
}
public int getWrittenRecords() {
return writtenRecords;
}
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.source;
+
+import org.apache.fluss.flink.adapter.SingleThreadFetcherManagerAdapter;
+import org.apache.fluss.flink.tiering.source.split.TieringSplit;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher;
+import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherTask;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+
+import java.util.Collection;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/**
+ * The SplitFetcherManager for tiering source. This class is needed to help notify a table reaches
+ * to deadline of tiering to {@link TieringSplitReader}.
+ */
+public class TieringSourceFetcherManager<WriteResult>
+ extends SingleThreadFetcherManagerAdapter<
+ TableBucketWriteResult<WriteResult>, TieringSplit> {
+
+ public TieringSourceFetcherManager(
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
+ elementsQueue,
+ Supplier<SplitReader<TableBucketWriteResult<WriteResult>, TieringSplit>>
+ splitReaderSupplier,
+ Configuration configuration,
+ Consumer<Collection<String>> splitFinishedHook) {
+ super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook);
+ }
+
+ public void markTableReachTieringMaxDuration(long tableId) {
+ if (!fetchers.isEmpty()) {
+ // The fetcher thread is still running. This should be the majority of the cases.
+ fetchers.values()
+ .forEach(
+ splitFetcher ->
+ enqueueMarkTableReachTieringMaxDurationTask(
+ splitFetcher, tableId));
+ } else {
+ SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher =
+ createSplitFetcher();
+ enqueueMarkTableReachTieringMaxDurationTask(splitFetcher, tableId);
+ startFetcher(splitFetcher);
+ }
+ }
+
+ private void enqueueMarkTableReachTieringMaxDurationTask(
+ SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher,
+ long reachTieringDeadlineTable) {
+ splitFetcher.enqueueTask(
+ new SplitFetcherTask() {
+ @Override
+ public boolean run() {
+ ((TieringSplitReader<WriteResult>) splitFetcher.getSplitReader())
+ .handleTableReachTieringMaxDuration(reachTieringDeadlineTable);
Review Comment:
The cast on line 76 to TieringSplitReader is unsafe and could fail with a ClassCastException if the split reader is of a different type. While this may work in the current implementation, it would be better to either add a type check before casting or ensure type safety through generic constraints. Consider adding an interface that both types implement or add an instanceof check with appropriate error handling.
```suggestion
SplitReader<TableBucketWriteResult<WriteResult>, TieringSplit> splitReader =
splitFetcher.getSplitReader();
if (splitReader instanceof TieringSplitReader) {
@SuppressWarnings("unchecked")
TieringSplitReader<WriteResult> tieringSplitReader =
(TieringSplitReader<WriteResult>) splitReader;
tieringSplitReader.handleTableReachTieringMaxDuration(
reachTieringDeadlineTable);
} else {
throw new IllegalStateException(
"Expected SplitReader of type "
+ TieringSplitReader.class.getName()
+ " but found "
+ (splitReader == null
? "null"
: splitReader.getClass().getName()));
}
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java:
##########
@@ -92,29 +97,38 @@ static void afterAll() throws Exception {
conn.close();
conn = null;
}
+
+ System.out.println("after all");
Review Comment:
There's a debug print statement (System.out.println) that should be removed. This is likely leftover debugging code that doesn't belong in production or test code.
```suggestion
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering;
+
+import org.apache.fluss.client.metadata.LakeSnapshot;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.ExceptionUtils;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL;
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_MAX;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The IT case for tiering. */
+class TieringITCase extends FlinkTieringTestBase {
+
+ @BeforeAll
+ protected static void beforeAll() {
+ FlinkTieringTestBase.beforeAll();
+ }
+
+ @AfterAll
+ protected static void afterAll() throws Exception {
+ FlinkTieringTestBase.afterAll();
+ }
+
+ @BeforeEach
Review Comment:
This method overrides [FlinkTieringTestBase.beforeEach](1); it is advisable to add an Override annotation.
```suggestion
@BeforeEach
@Override
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -248,6 +285,58 @@ private void mayCreateLogScanner() {
}
}
+ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>
+ forceCompleteTieringLogRecords() throws IOException {
+ Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
+ Map<TableBucket, String> finishedSplitIds = new HashMap<>();
+
+ // force finish all splits
+ Iterator<Map.Entry<TableBucket, TieringSplit>> currentTieringSplitsIterator =
+ currentTableSplitsByBucket.entrySet().iterator();
+ while (currentTieringSplitsIterator.hasNext()) {
+ Map.Entry<TableBucket, TieringSplit> entry = currentTieringSplitsIterator.next();
+ TableBucket bucket = entry.getKey();
+ TieringSplit split = entry.getValue();
+ if (split != null && split.isTieringLogSplit()) {
+ // get the current offset, timestamp that tiered so far
+ LogOffsetAndTimestamp logOffsetAndTimestamp =
+ currentTableTieredOffsetAndTimestamp.get(bucket);
+ long logEndOffset =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_OFFSET
+ // logEndOffset is equal to offset tiered + 1
+ : logOffsetAndTimestamp.logOffset + 1;
+ long timestamp =
+ logOffsetAndTimestamp == null
+ ? UNKNOWN_BUCKET_TIMESTAMP
+ : logOffsetAndTimestamp.timestamp;
+ TableBucketWriteResult<WriteResult> bucketWriteResult =
+ completeLakeWriter(
+ bucket, split.getPartitionName(), logEndOffset, timestamp);
+
+ if (logEndOffset == UNKNOWN_BUCKET_OFFSET) {
+ // when the log end offset is unknown, the write result must be
+ // null, otherwise, we should throw exception directly to avoid data
+ // inconsistent
+ checkState(
+ bucketWriteResult.writeResult() == null,
+ "bucketWriteResult must be null when log end offset is unknown when tiering "
+ + split);
+ }
+
+ writeResults.put(bucket, bucketWriteResult);
+ finishedSplitIds.put(bucket, split.splitId());
+ LOG.info(
+ "Split {} is forced to be finished due to tiering reach max duration.",
+ split.splitId());
+ currentTieringSplitsIterator.remove();
+ }
+ }
+ reachTieringMaxDurationTables.remove(this.currentTableId);
+ mayFinishCurrentTable();
+ return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
Review Comment:
The forceCompleteTieringLogRecords method only handles TieringLogSplit (line 300 checks split.isTieringLogSplit()). However, if there are TieringSnapshotSplits in currentTableSplitsByBucket when max duration is reached, they will be skipped silently and remain in the map. This could lead to incomplete cleanup and potential resource leaks or hanging tiering operations. The method should handle both log splits and snapshot splits, or document why snapshot splits can be safely ignored during force completion.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3754632952
> > > Thanks @luoyuxia for the work, I left some minor comment.
> > > But I also have a question for current design, Could we address large splits case via current design? The data freshness would be worse for large log split from my understanding.
> >
> >
> > Hi, thanks for reviewing my pr. Yes, large log split will harm data fressnes. But What do you mean `address large splits case`? In the current design, if we have a large split, and aussming the max tiering duration is 10 min, after 10 min, the large split will be forced to commit.
>
> I mean for the large log table, the data freshness goal(i.e. 5minute) would be worse to 10 minutes, 15 minutes after some scheduled rounds, maybe in that case we need monitor these tables and scale up tiering service resources.
Yes, your are right. But IIUC, the data freshness goal is a best effort when tiering resource is enough, but when tiering resource is not enough, it's normal that we can't reach to the goal. But a round of tiering of a table reach the max duraiton, it means that the resource is not enough, we need to scale up tiering service resources..
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2694308529
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -1811,6 +1811,23 @@ public class ConfigOptions {
+ ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT
+ " is false.");
+ public static final ConfigOption<Duration> LAKE_TIERING_TABLE_DURATION_MAX =
+ key("lake.tiering.table.duration.max")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(30))
+ .withDescription(
+ "The maximum duration for tiering a single table. If tiering a table exceeds this duration, "
+ + "it will be force completed: the tiering will be finalized and committed to the data lake "
Review Comment:
It'll be mark as finished and wait (about the data fressness interval) to scheduled to pending again. Do we need to just put into the pending queue for the table that is forced completed?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2185:
URL: https://github.com/apache/fluss/pull/2185#discussion_r2626793012
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSourceReaderTest.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.source;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
+import org.apache.fluss.flink.tiering.TestingWriteResult;
+import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
+import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TieringSourceReader}. */
+class TieringSourceReaderTest extends FlinkTestBase {
+
+ @Test
+ void testHandleTieringReachMaxDurationEvent1() throws Exception {
Review Comment:
The test method name 'testHandleTieringReachMaxDurationEvent1' has a '1' suffix, which typically indicates there should be multiple similar tests or that this is a placeholder name. Consider using a more descriptive name like 'testHandleTieringReachMaxDurationEventWithForceIgnore' or 'testMaxDurationEventWithEmptyAndNonEmptySplits' to better describe what the test verifies.
```suggestion
void testHandleTieringReachMaxDurationEventWithForceIgnore() throws Exception {
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -64,18 +64,24 @@ public class TieringSplitReader<WriteResult>
private static final Logger LOG = LoggerFactory.getLogger(TieringSplitReader.class);
- private static final Duration POLL_TIMEOUT = Duration.ofMillis(10000L);
+ public static final Duration DEFAULT_POLL_TIMEOUT = Duration.ofMillis(10_000L);
// unknown bucket timestamp for empty split or snapshot split
private static final long UNKNOWN_BUCKET_TIMESTAMP = -1;
+ private static final long UNKNOW_BUCKET_OFFSET = -1;
+
Review Comment:
Typo in constant name: 'UNKNOW_BUCKET_OFFSET' should be 'UNKNOWN_BUCKET_OFFSET' to be consistent with UNKNOWN_BUCKET_TIMESTAMP.
```suggestion
// unknown bucket offset for empty split or snapshot split
private static final long UNKNOWN_BUCKET_OFFSET = -1;
/**
* @deprecated Use {@link #UNKNOWN_BUCKET_OFFSET} instead.
*/
@Deprecated
private static final long UNKNOW_BUCKET_OFFSET = UNKNOWN_BUCKET_OFFSET;
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java:
##########
@@ -34,4 +34,21 @@ public class TieringSourceOptions {
.defaultValue(Duration.ofSeconds(30))
.withDescription(
"The fixed interval to request tiering table from Fluss cluster, by default 30 seconds.");
+
+ public static final ConfigOption<Duration> TIERING_TABLE_DURATION_MAX =
+ key("tiering.table.duration.max")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(10))
+ .withDescription(
+ "The maximum duration for tiering a single table. If tiering a table exceeds this duration, "
+ + "it will be force completed: the tiering will be finalized and committed to the data lake "
+ + "(e.g., Paimon) immediately, even if they haven't reached their desired stopping offsets.");
Review Comment:
The description mentions that tiering will be "force completed" and "committed to the data lake (e.g., Paimon) immediately, even if they haven't reached their desired stopping offsets." However, the implementation shows that splits marked with forceIgnore are skipped entirely (TieringSplitReader lines 181-184), not force-completed. This discrepancy between the documentation and implementation could mislead users. Consider clarifying that timed-out splits may be either force-completed (if in progress) or skipped (if not yet started).
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -236,6 +277,53 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
}
}
+ private Set<Long> checkTableReachMaxTieringDuration() {
+ Set<Long> tieringReachMaxDurationTables = new HashSet<>();
+ long currentTime = clock.milliseconds();
+ for (Map.Entry<Long, Long> tieringTableDeadline : tieringTablesDeadline.entrySet()) {
+ long tableId = tieringTableDeadline.getKey();
+ long deadline = tieringTableDeadline.getValue();
+ if (deadline < currentTime) {
+ tieringReachMaxDurationTables.add(tableId);
+ }
+ }
+ return tieringReachMaxDurationTables;
+ }
+
+ private void handleReachMaxTieringDurationTables(
+ Set<Long> tieringReachMaxDurationTables, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("Fail to check tiering timeout tables.", throwable);
+ return;
+ }
+
+ for (Long reachMaxDurationTable : tieringReachMaxDurationTables) {
+ for (TieringSplit tieringSplit : pendingSplits) {
+ if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
+ // force ignore this tiering split since the tiering for this table is timeout,
+ // we have to force to set to ignore the tiering split so that the
+ // tiering source reader can ignore them directly
+ tieringSplit.forceIgnore();
+ } else {
+ // we can break directly, if found any one split's table id is not equal to the
+ // timeout
+ // table, the following split must be not equal to the table id
+ break;
Review Comment:
The pendingSplits list is accessed both from the periodic timeout check (line 301) and the assignSplits method (line 350), which can be called from different threads via handleSplitRequest and addSplitsBack. While assignSplits synchronizes on readersAwaitingSplit, the timeout check doesn't use any synchronization, which could lead to a ConcurrentModificationException or inconsistent state when iterating and modifying pendingSplits concurrently.
```suggestion
// Access to pendingSplits must be synchronized consistently with other paths
// (e.g., assignSplits) to avoid concurrent modification.
synchronized (readersAwaitingSplit) {
for (TieringSplit tieringSplit : pendingSplits) {
if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
// force ignore this tiering split since the tiering for this table is
// timeout, we have to force to set to ignore the tiering split so that the
// tiering source reader can ignore them directly
tieringSplit.forceIgnore();
} else {
// we can break directly, if found any one split's table id is not equal to
// the timeout table, the following split must be not equal to the table id
break;
}
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -236,6 +277,53 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
}
}
+ private Set<Long> checkTableReachMaxTieringDuration() {
+ Set<Long> tieringReachMaxDurationTables = new HashSet<>();
+ long currentTime = clock.milliseconds();
+ for (Map.Entry<Long, Long> tieringTableDeadline : tieringTablesDeadline.entrySet()) {
+ long tableId = tieringTableDeadline.getKey();
+ long deadline = tieringTableDeadline.getValue();
+ if (deadline < currentTime) {
+ tieringReachMaxDurationTables.add(tableId);
+ }
+ }
+ return tieringReachMaxDurationTables;
+ }
+
+ private void handleReachMaxTieringDurationTables(
+ Set<Long> tieringReachMaxDurationTables, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("Fail to check tiering timeout tables.", throwable);
+ return;
+ }
+
+ for (Long reachMaxDurationTable : tieringReachMaxDurationTables) {
+ for (TieringSplit tieringSplit : pendingSplits) {
+ if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
+ // force ignore this tiering split since the tiering for this table is timeout,
+ // we have to force to set to ignore the tiering split so that the
+ // tiering source reader can ignore them directly
+ tieringSplit.forceIgnore();
+ } else {
+ // we can break directly, if found any one split's table id is not equal to the
+ // timeout
+ // table, the following split must be not equal to the table id
+ break;
+ }
+ }
+
+ LOG.debug("Found the table {} reach max tiering duration.", reachMaxDurationTable);
+
+ // broadcast the tiering reach max duration event to all readers,
+ // we broadcast all for simplicity
+ Set<Integer> readers = new HashSet<>(context.registeredReaders().keySet());
+ for (int reader : readers) {
+ context.sendEventToSourceReader(
+ reader, new TieringReachMaxDurationEvent(reachMaxDurationTable));
+ }
Review Comment:
The deadline is removed from tieringTablesDeadline when a table finishes or fails (lines 240 and 259), but it's never removed when the timeout event is sent at line 323. This means a timed-out table will continue triggering timeout checks and events on every periodic check interval, potentially sending duplicate TieringReachMaxDurationEvent messages to readers. Consider removing the deadline after sending the timeout event.
```suggestion
}
// remove the deadline once timeout is handled to avoid duplicate timeout events
tieringTablesDeadline.remove(reachMaxDurationTable);
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -260,6 +349,11 @@ private void assignSplits() {
if (!pendingSplits.isEmpty()) {
TieringSplit tieringSplit = pendingSplits.remove(0);
context.assignSplit(tieringSplit, nextAwaitingReader);
+ long tableId = tieringSplit.getTableBucket().getTableId();
+ if (!tieringTablesDeadline.containsKey(tableId)) {
+ tieringTablesDeadline.put(
+ tableId, clock.milliseconds() + tieringTableDurationMaxMs);
+ }
Review Comment:
The deadline is set only when the first split of a table is assigned (lines 353-356). However, if a table has multiple buckets processed in parallel by different readers, each bucket's split will be assigned at different times, but they all share the same deadline. This means some splits may have significantly less time to complete than others. Consider setting the deadline when the table tiering starts (when splits are generated) rather than when splits are assigned to ensure all splits for a table have equal time to complete.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java:
##########
@@ -760,4 +750,119 @@ private void verifyTieringSplitAssignment(
assertThat(allTieringSplits)
.allMatch(tieringSplit -> tieringSplit.getTablePath().equals(expectedTablePath));
}
+
+ private TieringSourceEnumerator createDefaultTieringSourceEnumerator(
+ Configuration flussConf, MockSplitEnumeratorContext<TieringSplit> context) {
+ return new TieringSourceEnumerator(
+ flussConf,
+ context,
+ 500,
+ Duration.ofMinutes(10).toMillis(),
+ Duration.ofSeconds(10).toMillis());
+ }
+
+ private TieringSourceEnumerator createTieringSourceEnumeratorWithManualClock(
+ Configuration flussConf,
+ MockSplitEnumeratorContext<TieringSplit> context,
+ ManualClock clock,
+ long tieringTableDurationMaxMs,
+ long tieringTableDurationDetectIntervalMs) {
+ return new TieringSourceEnumerator(
+ flussConf,
+ context,
+ 500,
+ tieringTableDurationMaxMs,
+ tieringTableDurationDetectIntervalMs,
+ clock);
+ }
+
+ /**
+ * Get events sent to readers from MockSplitEnumeratorContext using reflection.
+ *
+ * @param context the MockSplitEnumeratorContext
+ * @return map of reader ID to list of events sent to that reader
+ */
+ @SuppressWarnings("unchecked")
+ private Map<Integer, List<SourceEvent>> getEventsToReaders(
+ MockSplitEnumeratorContext<TieringSplit> context) {
+ try {
+ Field eventsToReadersField =
+ MockSplitEnumeratorContext.class.getDeclaredField("eventsToReaders");
+ eventsToReadersField.setAccessible(true);
+ return (Map<Integer, List<SourceEvent>>) eventsToReadersField.get(context);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ // If reflection fails, return empty map - test will still verify splits behavior
+ return new HashMap<>();
+ }
+ }
+
Review Comment:
The reflection-based helper method getEventsToReaders (lines 780-797) is defined but never used. It appears to be superseded by the MockSplitEnumeratorContext.getSentSourceEvent() method called at line 845. Consider removing this unused code to reduce maintainability burden.
```suggestion
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -339,30 +417,29 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
checkNotNull(currentTableNumberOfSplits));
}
- private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringLogSplit> emptySplits) {
+ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringSplit> emptySplits) {
Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
Map<TableBucket, String> finishedSplitIds = new HashMap<>();
- for (TieringLogSplit logSplit : emptySplits) {
- TableBucket tableBucket = logSplit.getTableBucket();
- finishedSplitIds.put(tableBucket, logSplit.splitId());
+ for (TieringSplit tieringSplit : emptySplits) {
+ TableBucket tableBucket = tieringSplit.getTableBucket();
+ finishedSplitIds.put(tableBucket, tieringSplit.splitId());
writeResults.put(
tableBucket,
toTableBucketWriteResult(
- logSplit.getTablePath(),
+ tieringSplit.getTablePath(),
tableBucket,
- logSplit.getPartitionName(),
+ tieringSplit.getPartitionName(),
null,
- logSplit.getStoppingOffset(),
+ UNKNOW_BUCKET_OFFSET,
Review Comment:
When completing an empty split that was force-ignored, the logEndOffset is set to UNKNOW_BUCKET_OFFSET (-1). This could cause issues downstream when the offset is used for checkpointing or resuming tiering operations. Consider using a more appropriate value or documenting that -1 is a valid sentinel value for force-ignored splits.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -236,6 +277,53 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
}
}
+ private Set<Long> checkTableReachMaxTieringDuration() {
+ Set<Long> tieringReachMaxDurationTables = new HashSet<>();
+ long currentTime = clock.milliseconds();
+ for (Map.Entry<Long, Long> tieringTableDeadline : tieringTablesDeadline.entrySet()) {
+ long tableId = tieringTableDeadline.getKey();
+ long deadline = tieringTableDeadline.getValue();
+ if (deadline < currentTime) {
+ tieringReachMaxDurationTables.add(tableId);
+ }
+ }
+ return tieringReachMaxDurationTables;
+ }
+
+ private void handleReachMaxTieringDurationTables(
+ Set<Long> tieringReachMaxDurationTables, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("Fail to check tiering timeout tables.", throwable);
+ return;
+ }
+
+ for (Long reachMaxDurationTable : tieringReachMaxDurationTables) {
+ for (TieringSplit tieringSplit : pendingSplits) {
+ if (tieringSplit.getTableBucket().getTableId() == reachMaxDurationTable) {
+ // force ignore this tiering split since the tiering for this table is timeout,
+ // we have to force to set to ignore the tiering split so that the
+ // tiering source reader can ignore them directly
+ tieringSplit.forceIgnore();
+ } else {
+ // we can break directly, if found any one split's table id is not equal to the
+ // timeout
+ // table, the following split must be not equal to the table id
+ break;
Review Comment:
The logic assumes pendingSplits are ordered by table ID and will break on the first mismatch. However, after introducing Collections.shuffle(tieringSplits) at line 423, this ordering assumption is no longer valid. Splits from different tables can be interleaved, so the break statement at line 311 will prevent marking all splits for the timeout table as forceIgnore.
```suggestion
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java:
##########
@@ -101,7 +110,8 @@ public TieringSnapshotSplit copy(int numberOfSplits) {
partitionName,
snapshotId,
logOffsetOfSnapshot,
- numberOfSplits);
+ numberOfSplits,
+ forceIgnore);
}
Review Comment:
The copy method preserves the forceIgnore state from the original split. However, this is inconsistent with TieringLogSplit.copy(), which does not preserve forceIgnore and always defaults it to false. This inconsistency could lead to unexpected behavior when different split types are copied.
```suggestion
numberOfSplits);
}
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -249,6 +337,7 @@ private void generateAndAssignSplits(
private void assignSplits() {
/* This method may be called from both addSplitsBack and handleSplitRequest, make it thread safe. */
+ // todo: do we need to add lock?
Review Comment:
The TODO comment at line 340 asks "do we need to add lock?" but this question is now more critical given that the timeout checking code (lines 301-312) accesses pendingSplits without synchronization while assignSplits() synchronizes on readersAwaitingSplit. This creates a potential race condition. The TODO should either be resolved by adding proper synchronization or be updated to reflect the new concurrency concerns.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java:
##########
@@ -43,11 +51,29 @@ public final class TieringSourceReader<WriteResult>
private final Connection connection;
public TieringSourceReader(
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
+ elementsQueue,
SourceReaderContext context,
Connection connection,
LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
+ this(elementsQueue, context, connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
+ }
Review Comment:
The constructor now requires an elementsQueue parameter, changing from creating the queue internally to accepting it externally. However, the non-test constructor at line 53 doesn't expose a way to customize the queue, which limits flexibility for testing or customization. While this is addressed by the @VisibleForTesting constructor, consider whether the production code path should also allow queue configuration or if it should remain internal.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -248,6 +278,44 @@ private void mayCreateLogScanner() {
}
}
+ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>
+ forceCompleteTieringLogRecords() throws IOException {
+ Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
+ Map<TableBucket, String> finishedSplitIds = new HashMap<>();
+
+ Iterator<Map.Entry<TableBucket, TieringSplit>> currentTieringSplitsIterator =
+ currentTableSplitsByBucket.entrySet().iterator();
+ while (currentTieringSplitsIterator.hasNext()) {
+ Map.Entry<TableBucket, TieringSplit> entry = currentTieringSplitsIterator.next();
+ TableBucket bucket = entry.getKey();
+ TieringSplit split = entry.getValue();
+ if (split != null && split.isTieringLogSplit()) {
+ LogOffsetAndTimestamp logOffsetAndTimestamp =
+ currentTableTieredOffsetAndTimestamp.get(bucket);
+ long logEndOffset =
+ logOffsetAndTimestamp == null
+ ? UNKNOW_BUCKET_OFFSET
Review Comment:
Typo in variable reference: 'UNKNOW_BUCKET_OFFSET' should be 'UNKNOWN_BUCKET_OFFSET'.
```suggestion
? UNKNOWN_BUCKET_OFFSET
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/event/TieringReachMaxDurationEvent.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.tiering.event;
+
+import org.apache.flink.api.connector.source.SourceEvent;
+
+import java.util.Objects;
+
+/**
+ * SourceEvent used to notify TieringSourceReader that a table has reached the maximum tiering
+ * duration and should be force completed.
+ */
+public class TieringReachMaxDurationEvent implements SourceEvent {
Review Comment:
Class TieringReachMaxDurationEvent overrides [hashCode](1) but not equals.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -339,30 +417,29 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
checkNotNull(currentTableNumberOfSplits));
}
- private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringLogSplit> emptySplits) {
+ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringSplit> emptySplits) {
Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
Map<TableBucket, String> finishedSplitIds = new HashMap<>();
- for (TieringLogSplit logSplit : emptySplits) {
- TableBucket tableBucket = logSplit.getTableBucket();
- finishedSplitIds.put(tableBucket, logSplit.splitId());
+ for (TieringSplit tieringSplit : emptySplits) {
+ TableBucket tableBucket = tieringSplit.getTableBucket();
+ finishedSplitIds.put(tableBucket, tieringSplit.splitId());
writeResults.put(
tableBucket,
toTableBucketWriteResult(
- logSplit.getTablePath(),
+ tieringSplit.getTablePath(),
tableBucket,
- logSplit.getPartitionName(),
+ tieringSplit.getPartitionName(),
null,
- logSplit.getStoppingOffset(),
+ UNKNOW_BUCKET_OFFSET,
Review Comment:
Typo in variable reference: 'UNKNOW_BUCKET_OFFSET' should be 'UNKNOWN_BUCKET_OFFSET'.
```suggestion
UNKNOWN_BUCKET_OFFSET,
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -324,6 +418,9 @@ private void generateTieringSplits(Tuple3<Long, Long, TablePath> tieringTable)
List<TieringSplit> tieringSplits =
populateNumberOfTieringSplits(
splitGenerator.generateTableSplits(tieringTable.f2));
+ // shuffle tiering split to avoid splits tiering skew
+ // after introduce tiering max duration
+ Collections.shuffle(tieringSplits);
Review Comment:
Shuffling the tiering splits (line 423) aims to avoid splits tiering skew after introducing tiering max duration. However, this shuffle happens at the table level, meaning all splits for a table are shuffled together before being added to pendingSplits. Since pendingSplits is processed FIFO (line 350), and multiple tables may be tiering concurrently, this shuffle provides limited benefit for balancing work across readers when multiple tables are involved. Consider documenting this limitation or implementing a more sophisticated load balancing strategy.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java:
##########
@@ -74,6 +74,9 @@ public byte[] serialize(TieringSplit split) throws IOException {
out.writeBoolean(false);
}
+ // write force ignore
+ out.writeBoolean(split.isForceIgnore());
Review Comment:
The serialization format has changed by adding a new forceIgnore boolean field (line 78), but the version number remains at VERSION_0. This breaks backward compatibility: older deserializers will fail when reading data serialized by the new version, and the new deserializer will fail when reading old serialized data that doesn't have the forceIgnore field. Consider bumping the version to VERSION_1 and handling both old and new formats in the deserialize method for backward compatibility.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Tiering support commit by time [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #2185:
URL: https://github.com/apache/fluss/pull/2185#issuecomment-3727426257
> Thanks @luoyuxia for the work, I left some minor comment.
>
> But I also have a question for current design, Could we address large splits case via current design? The data freshness would be worse for large log split from my understanding.
Hi, thanks for reviewing my pr. Yes, large log split will harm data fressnes.
But What do you mean `address large splits case`?
In the current design, if we have a large split, and aussming the max tiering duration is 10 min, after 10 min, the large split will be forced to commit.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org