Copilot commented on code in PR #3503:
URL: https://github.com/apache/fluss/pull/3503#discussion_r3447939674
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java:
##########
@@ -54,7 +55,8 @@ public TieringSplit toSourceSplit() {
split.getPartitionName(),
split.getStartingOffset(),
split.getStoppingOffset(),
- split.getNumberOfSplits());
+ split.getNumberOfSplits(),
+ split.getTag());
Review Comment:
`TieringSplitState#toSourceSplit()` drops the `skipCurrentRound` flag for
log splits as well. This can change behavior after converting split state (the
reader treats skipped splits as empty). Preserve the flag when reconstructing
the `TieringLogSplit`.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -494,6 +495,24 @@ private List<TieringSplit>
populateNumberOfTieringSplits(List<TieringSplit> tier
.collect(Collectors.toList());
}
+ private List<TieringSplit> populateTagsForTieringSplits(List<TieringSplit>
tieringSplits) {
+ if (tieringSplits.isEmpty()) {
+ return tieringSplits;
+ }
+ String tieringRoundId = String.valueOf(System.currentTimeMillis());
Review Comment:
Using `System.currentTimeMillis()` as the tiering-round id can produce
duplicate tags if two rounds are generated within the same millisecond (e.g.,
under fast retries/tests). Since the tag is meant to coordinate a tiering
round, it should be unique; using a UUID avoids collisions.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java:
##########
@@ -45,7 +45,8 @@ public TieringSplit toSourceSplit() {
split.getPartitionName(),
split.getSnapshotId(),
split.getLogOffsetOfSnapshot(),
- split.getNumberOfSplits());
+ split.getNumberOfSplits(),
+ split.getTag());
Review Comment:
`TieringSplitState#toSourceSplit()` drops the `skipCurrentRound` flag. That
flag is part of split semantics (used by
`TieringSplitReader#handleSplitsChanges`), so reconstructing a split from state
should preserve it to avoid accidentally processing a split that was marked to
be skipped.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java:
##########
@@ -31,12 +31,15 @@ public abstract class TieringSplit implements SourceSplit {
public static final byte TIERING_SNAPSHOT_SPLIT_FLAG = 1;
public static final byte TIERING_LOG_SPLIT_FLAG = 2;
+ public static final String EMPTY_TAG = "";
+ public static final String FIRST_SPLIT_TAG_PREFIX = "firstSplit-";
protected static final int UNKNOWN_NUMBER_OF_SPLITS = -1;
protected final TablePath tablePath;
protected final TableBucket tableBucket;
@Nullable protected final String partitionName;
+ protected final String tag;
Review Comment:
A new `tag` field is introduced on `TieringSplit`, but the class’ `equals()`
/ `hashCode()` implementations (further down) do not incorporate it. If tags
are intended to be part of split state (they're serialized/copied and passed
into writer init), ignoring them can cause different tagged splits to compare
equal and be deduplicated in sets/maps, and can hide tag regressions in tests.
Consider including `tag` in `equals()` / `hashCode()`, or documenting
explicitly why it is intentionally excluded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
