This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 1b2bd2924 [lake] Refactor tiering split assertions to ignore round
metadata (#3506)
1b2bd2924 is described below
commit 1b2bd2924c1e46459f8349824e9fafa1fef2a224
Author: yuxia Luo <[email protected]>
AuthorDate: Mon Jun 22 22:19:41 2026 +0800
[lake] Refactor tiering split assertions to ignore round metadata (#3506)
---
.../flink/tiering/source/split/TieringSplit.java | 21 ------
.../enumerator/TieringSourceEnumeratorTest.java | 83 +++++++++++-----------
.../source/split/TieringSplitSerializerTest.java | 26 +------
3 files changed, 42 insertions(+), 88 deletions(-)
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
index b0ea25693..b41aef3f7 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
@@ -87,8 +87,6 @@ public abstract class TieringSplit implements SourceSplit {
}
this.numberOfSplits = numberOfSplits;
this.skipCurrentRound = skipCurrentRound;
- validateSplitIndex(numberOfSplits, splitIndex);
- validateTieringRoundTimestamp(tieringRoundTimestamp);
this.splitIndex = splitIndex;
this.tieringRoundTimestamp = tieringRoundTimestamp;
}
@@ -215,23 +213,4 @@ public abstract class TieringSplit implements SourceSplit {
splitIndex,
tieringRoundTimestamp);
}
-
- private static void validateSplitIndex(int numberOfSplits, int splitIndex)
{
- if (splitIndex < UNKNOWN_SPLIT_INDEX) {
- throw new IllegalArgumentException("Split index must be -1 or
non-negative.");
- }
- if (splitIndex != UNKNOWN_SPLIT_INDEX
- && numberOfSplits != UNKNOWN_NUMBER_OF_SPLITS
- && (numberOfSplits <= 0 || splitIndex >= numberOfSplits)) {
- throw new IllegalArgumentException(
- "Split index must be smaller than the number of splits.");
- }
- }
-
- private static void validateTieringRoundTimestamp(long
tieringRoundTimestamp) {
- if (tieringRoundTimestamp != UNKNOWN_TIERING_ROUND_TIMESTAMP
- && tieringRoundTimestamp <= 0) {
- throw new IllegalArgumentException("Tiering round timestamp must
be -1 or positive.");
- }
- }
}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
index 9e7975d0c..237e29037 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
@@ -52,6 +52,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static
org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
import static
org.apache.fluss.config.ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE;
@@ -142,10 +143,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualLogAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualLogAssignment::addAll));
- assertTieringRoundMetadata(actualLogAssignment);
- clearTieringSplitMetadata(actualLogAssignment);
- assertThat(actualLogAssignment)
-
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
+ assertTieringSplitsMatch(actualLogAssignment,
expectedLogAssignment);
}
}
@@ -187,10 +185,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
- assertTieringRoundMetadata(actualAssignment);
- clearTieringSplitMetadata(actualAssignment);
- assertThat(actualAssignment)
-
.containsExactlyInAnyOrderElementsOf(expectedSnapshotAssignment);
+ assertTieringSplitsMatch(actualAssignment,
expectedSnapshotAssignment);
// mock finished tiered this round, check second round
context.getSplitsAssignmentSequence().clear();
@@ -234,9 +229,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualLogAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualLogAssignment::addAll));
- clearTieringSplitMetadata(actualLogAssignment);
- assertThat(actualLogAssignment)
-
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
+ assertTieringSplitsMatch(actualLogAssignment,
expectedLogAssignment);
}
}
@@ -278,8 +271,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
- clearTieringSplitMetadata(actualAssignment);
-
assertThat(actualAssignment).containsExactlyInAnyOrderElementsOf(expectedAssignment);
+ assertTieringSplitsMatch(actualAssignment, expectedAssignment);
// mock finished tiered this round, check second round
context.getSplitsAssignmentSequence().clear();
@@ -322,9 +314,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualLogAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualLogAssignment::addAll));
- clearTieringSplitMetadata(actualLogAssignment);
- assertThat(actualLogAssignment)
-
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
+ assertTieringSplitsMatch(actualLogAssignment,
expectedLogAssignment);
}
}
@@ -423,9 +413,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()) {
splitsAssignment.assignment().values().forEach(actualLogAssignment::addAll);
}
- clearTieringSplitMetadata(actualLogAssignment);
- assertThat(actualLogAssignment)
-
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
+ assertTieringSplitsMatch(actualLogAssignment,
expectedLogAssignment);
}
}
@@ -480,8 +468,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()) {
splitsAssignment.assignment().values().forEach(actualAssignment::addAll);
}
- clearTieringSplitMetadata(actualAssignment);
-
assertThat(actualAssignment).containsExactlyInAnyOrderElementsOf(expectedAssignment);
+ assertTieringSplitsMatch(actualAssignment, expectedAssignment);
// mock finished tiered this round, check second round
context.getSplitsAssignmentSequence().clear();
@@ -548,9 +535,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()) {
splitsAssignment.assignment().values().forEach(actualLogAssignment::addAll);
}
- clearTieringSplitMetadata(actualLogAssignment);
- assertThat(actualLogAssignment)
-
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
+ assertTieringSplitsMatch(actualLogAssignment,
expectedLogAssignment);
}
}
@@ -589,8 +574,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
- clearTieringSplitMetadata(actualAssignment);
-
assertThat(actualAssignment).containsExactlyInAnyOrderElementsOf(expectedAssignment);
+ assertTieringSplitsMatch(actualAssignment, expectedAssignment);
// mock tiering fail by send tiering fail event
context.getSplitsAssignmentSequence().clear();
@@ -604,8 +588,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualAssignment1 = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment1::addAll));
- clearTieringSplitMetadata(actualAssignment1);
-
assertThat(actualAssignment1).containsExactlyInAnyOrderElementsOf(expectedAssignment);
+ assertTieringSplitsMatch(actualAssignment1, expectedAssignment);
}
}
@@ -685,13 +668,41 @@ class TieringSourceEnumeratorTest extends TieringTestBase
{
}
// --------------------- Test Utils ---------------------
- private static void assertTieringRoundMetadata(List<TieringSplit>
tieringSplits) {
+
+ /**
+ * Asserts that the actual assigned splits match the expected ones. {@code
splitIndex} and
+ * {@code tieringRoundTimestamp} are assigned at split-generation time and
unknown to the
+ * expected splits constructed in tests, so they are validated as round
metadata via {@link
+ * #assertValidTieringRound} and then normalized away before comparing the
remaining fields with
+ * the regular {@code equals}.
+ */
+ private static void assertTieringSplitsMatch(
+ List<TieringSplit> actualSplits, List<TieringSplit>
expectedSplits) {
+ assertValidTieringRound(actualSplits);
+ List<TieringSplit> normalizedActualSplits =
+ actualSplits.stream()
+ .map(
+ split ->
+ split.copy(
+ split.getNumberOfSplits(),
+
TieringSplit.UNKNOWN_SPLIT_INDEX,
+
TieringSplit.UNKNOWN_TIERING_ROUND_TIMESTAMP))
+ .collect(Collectors.toList());
+
assertThat(normalizedActualSplits).containsExactlyInAnyOrderElementsOf(expectedSplits);
+ }
+
+ /**
+ * Validates the per-round metadata of the assigned splits: the split
indexes cover {@code
+ * 0..size-1} with exactly one first split, every split reports the round
size as its number of
+ * splits, and all splits share the same positive tiering round timestamp.
+ */
+ private static void assertValidTieringRound(List<TieringSplit>
tieringSplits) {
assertThat(tieringSplits).isNotEmpty();
assertThat(tieringSplits).filteredOn(TieringSplit::isFirstSplit).hasSize(1);
assertThat(tieringSplits)
.extracting(TieringSplit::getSplitIndex)
.containsExactlyInAnyOrderElementsOf(
- java.util.stream.IntStream.range(0,
tieringSplits.size())
+ IntStream.range(0, tieringSplits.size())
.boxed()
.collect(Collectors.toList()));
long tieringRoundTimestamp =
tieringSplits.get(0).getTieringRoundTimestamp();
@@ -705,18 +716,6 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
});
}
- private static void clearTieringSplitMetadata(List<TieringSplit>
tieringSplits) {
- for (int i = 0; i < tieringSplits.size(); i++) {
- TieringSplit tieringSplit = tieringSplits.get(i);
- tieringSplits.set(
- i,
- tieringSplit.copy(
- tieringSplit.getNumberOfSplits(),
- TieringSplit.UNKNOWN_SPLIT_INDEX,
- TieringSplit.UNKNOWN_TIERING_ROUND_TIMESTAMP));
- }
- }
-
private void registerReaderAndHandleSplitRequests(
FlussMockSplitEnumeratorContext<TieringSplit> context,
TieringSourceEnumerator enumerator,
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
index ef7de7f08..f9c0913b9 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
@@ -153,7 +153,7 @@ class TieringSplitSerializerTest {
}
@Test
- void testTieringRoundMetadataSerde() throws Exception {
+ void testTieringRoundTimestampSerde() throws Exception {
TieringSnapshotSplit snapshotSplit =
new TieringSnapshotSplit(tablePath, tableBucket, null, 0L,
200L, 10, 0, 1000L);
byte[] serialized = serializer.serialize(snapshotSplit);
@@ -172,28 +172,4 @@ class TieringSplitSerializerTest {
assertThat(deserializedLogSplit.isFirstSplit()).isFalse();
assertThat(deserializedLogSplit.getTieringRoundTimestamp()).isEqualTo(2000L);
}
-
- @Test
- void testTieringRoundMetadataParticipatesInSplitIdentity() {
- TieringSnapshotSplit snapshotSplit =
- new TieringSnapshotSplit(tablePath, tableBucket, null, 0L,
200L, 10, 0, 1000L);
- TieringSnapshotSplit snapshotSplitWithDifferentMetadata =
- new TieringSnapshotSplit(tablePath, tableBucket, null, 0L,
200L, 10, 1, 2000L);
-
assertThat(snapshotSplit).isNotEqualTo(snapshotSplitWithDifferentMetadata);
-
- TieringLogSplit logSplit =
- new TieringLogSplit(tablePath, tableBucket, null, 100, 200,
40, 0, 1000L);
- TieringLogSplit logSplitWithDifferentMetadata =
- new TieringLogSplit(tablePath, tableBucket, null, 100, 200,
40, 1, 2000L);
- assertThat(logSplit).isNotEqualTo(logSplitWithDifferentMetadata);
- }
-
- @Test
- void testLogSplitDefaultTieringRoundMetadataIsUnknown() {
- TieringLogSplit logSplit = new TieringLogSplit(tablePath, tableBucket,
null, 100, 200);
-
-
assertThat(logSplit.getSplitIndex()).isEqualTo(TieringSplit.UNKNOWN_SPLIT_INDEX);
- assertThat(logSplit.getTieringRoundTimestamp())
- .isEqualTo(TieringSplit.UNKNOWN_TIERING_ROUND_TIMESTAMP);
- }
}
