offsets = Collections.singletonMap(bucket, 100L);
+ TieringStateEntry corruptEntry =
+ new TieringStateEntry(
+ PartitionMarkDoneState.STATE_KEY,
+ 1,
+ "{not-json".getBytes(StandardCharsets.UTF_8));
+ LakeSnapshot snapshot =
+ new LakeSnapshot(1L, offsets, Collections.singletonList(corruptEntry));
+
+ // bucket offsets remain accessible even though the state payload is corrupt.
+ assertThat(snapshot.getTableBucketsOffset()).containsEntry(bucket, 100L);
+ // only the typed accessor surfaces the parse failure.
+ assertThatThrownBy(snapshot::getPartitionMarkDoneState)
+ .isInstanceOf(RuntimeException.class);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java
new file mode 100644
index 0000000000..3760e7726c
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java
@@ -0,0 +1,171 @@
+/*
+ * 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.lake.committer;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.utils.json.JsonSerdeUtils;
+import org.apache.fluss.utils.json.PartitionMarkDoneStateJsonSerde;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/**
+ * The partition mark-done state, persisted as the {@link TieringStateEntry} keyed by {@link
+ * #STATE_KEY} in the lake offsets file. Each tracked partition records its last update time and
+ * done time ({@link #NOT_DONE} until first marked done); a partition absent from the map has never
+ * been tracked. Entries of dropped partitions are removed by omitting them from the next payload.
+ *
+ * The payload schema version lives in the entry envelope. Evolve the payload by bumping {@link
+ * #CURRENT_VERSION}; {@link #fromStateEntry} rejects a higher version, and the caller must pass the
+ * entry through unchanged so a newer build's state is never dropped.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class PartitionMarkDoneState {
+
+ /** The tiering-state key of the partition mark-done state. */
+ public static final String STATE_KEY = "fluss.partition-mark-done";
+
+ public static final int CURRENT_VERSION = 1;
+
+ /** Sentinel done time of a partition that has not been marked done yet. */
+ public static final long NOT_DONE = -1L;
+
+ private final Map partitionStates;
+
+ public PartitionMarkDoneState(Map partitionStates) {
+ this.partitionStates =
+ partitionStates == null ? Collections.emptyMap() : new HashMap<>(partitionStates);
+ }
+
+ public Map getPartitionStates() {
+ return Collections.unmodifiableMap(partitionStates);
+ }
+
+ /** Wraps this state into its {@link TieringStateEntry} at {@link #CURRENT_VERSION}. */
+ public TieringStateEntry toStateEntry() {
+ return new TieringStateEntry(
+ STATE_KEY,
+ CURRENT_VERSION,
+ JsonSerdeUtils.writeValueAsBytes(this, PartitionMarkDoneStateJsonSerde.INSTANCE));
+ }
+
+ /**
+ * Parses the state from its {@link TieringStateEntry}.
+ *
+ * @throws IllegalArgumentException if the entry has a different key, an unsupported (newer)
+ * version, or a corrupt payload
+ */
+ public static PartitionMarkDoneState fromStateEntry(TieringStateEntry entry) {
+ checkArgument(
+ STATE_KEY.equals(entry.getStateKey()),
+ "Expected state key %s but got %s.",
+ STATE_KEY,
+ entry.getStateKey());
+ if (entry.getStateVersion() > CURRENT_VERSION) {
+ throw new IllegalArgumentException(
+ "Unsupported partition mark-done state version "
+ + entry.getStateVersion()
+ + " > "
+ + CURRENT_VERSION
+ + "; pass the entry through unchanged.");
+ }
+ return JsonSerdeUtils.readValue(
+ entry.getPayload(), PartitionMarkDoneStateJsonSerde.INSTANCE);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PartitionMarkDoneState that = (PartitionMarkDoneState) o;
+ return Objects.equals(partitionStates, that.partitionStates);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partitionStates);
+ }
+
+ @Override
+ public String toString() {
+ return "PartitionMarkDoneState{" + "partitionStates=" + partitionStates + '}';
+ }
+
+ /** The mark-done state of a single partition. */
+ public static class PartitionState {
+
+ private final long updateTime;
+ private final long doneTime;
+
+ public PartitionState(long updateTime, long doneTime) {
+ checkArgument(
+ updateTime >= 0, "updateTime must be non-negative but got %s.", updateTime);
+ checkArgument(
+ doneTime == NOT_DONE || doneTime > 0,
+ "doneTime must be %s (not done) or positive but got %s.",
+ NOT_DONE,
+ doneTime);
+ this.updateTime = updateTime;
+ this.doneTime = doneTime;
+ }
+
+ public long getUpdateTime() {
+ return updateTime;
+ }
+
+ public long getDoneTime() {
+ return doneTime;
+ }
+
+ public boolean isDone() {
+ return doneTime != NOT_DONE;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PartitionState that = (PartitionState) o;
+ return updateTime == that.updateTime && doneTime == that.doneTime;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(updateTime, doneTime);
+ }
+
+ @Override
+ public String toString() {
+ return "PartitionState{" + "updateTime=" + updateTime + ", doneTime=" + doneTime + '}';
+ }
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java
new file mode 100644
index 0000000000..69558606d0
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java
@@ -0,0 +1,101 @@
+/*
+ * 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.lake.committer;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * A keyed, versioned tiering-state entry carried by the lake offsets file alongside the bucket
+ * offsets.
+ *
+ * The entry is an opaque envelope: {@code stateKey} identifies the state owner (e.g. {@link
+ * PartitionMarkDoneState#STATE_KEY}), {@code stateVersion} is the schema version of the payload,
+ * and {@code payload} is a JSON object serialized as bytes. The transport and storage layers never
+ * interpret the payload; the offsets-file serde may re-encode it as a JSON tree (all fields
+ * preserved, byte-level layout not guaranteed). Entries with an unrecognized key or a
+ * higher-than-supported version must be passed through unchanged so state written by a newer build
+ * is never dropped.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class TieringStateEntry {
+
+ private final String stateKey;
+ private final int stateVersion;
+ private final byte[] payload;
+
+ public TieringStateEntry(String stateKey, int stateVersion, byte[] payload) {
+ checkArgument(
+ stateKey != null && !stateKey.isEmpty(), "stateKey must be a non-empty string.");
+ checkArgument(stateVersion > 0, "stateVersion must be positive but got %s.", stateVersion);
+ this.stateKey = stateKey;
+ this.stateVersion = stateVersion;
+ this.payload = checkNotNull(payload, "payload must not be null.");
+ }
+
+ public String getStateKey() {
+ return stateKey;
+ }
+
+ public int getStateVersion() {
+ return stateVersion;
+ }
+
+ public byte[] getPayload() {
+ return payload;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TieringStateEntry that = (TieringStateEntry) o;
+ return stateVersion == that.stateVersion
+ && stateKey.equals(that.stateKey)
+ && Arrays.equals(payload, that.payload);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(stateKey, stateVersion, Arrays.hashCode(payload));
+ }
+
+ @Override
+ public String toString() {
+ return "TieringStateEntry{"
+ + "stateKey='"
+ + stateKey
+ + '\''
+ + ", stateVersion="
+ + stateVersion
+ + ", payload="
+ + Arrays.toString(payload)
+ + '}';
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java
index b6e0794bef..21afdc7d58 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java
@@ -73,5 +73,13 @@ public static T readValue(byte[] json, JsonDeserializer deserializer) {
}
}
+ /**
+ * Parse the given JSON bytes into a tree node using the shared object mapper. Useful for
+ * opaque/pass-through JSON payloads whose concrete type is not known to the caller.
+ */
+ public static JsonNode readTree(byte[] json) throws IOException {
+ return OBJECT_MAPPER_INSTANCE.readTree(json);
+ }
+
private JsonSerdeUtils() {}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/PartitionMarkDoneStateJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/PartitionMarkDoneStateJsonSerde.java
new file mode 100644
index 0000000000..19c7475674
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/PartitionMarkDoneStateJsonSerde.java
@@ -0,0 +1,139 @@
+/*
+ * 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.utils.json;
+
+import org.apache.fluss.lake.committer.PartitionMarkDoneState;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState;
+import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Json serializer and deserializer for the {@link PartitionMarkDoneState} payload, e.g. {@code
+ * {"partitions":{"5":{"update_time":1704153550000,"done_time":-1}}}}. Partitions are written in
+ * ascending id order for deterministic output.
+ *
+ * The payload schema version lives in the {@code TieringStateEntry} envelope, not here. The
+ * serde validates the JSON structure and field types; value ranges are enforced by the {@link
+ * PartitionState} constructor. Unknown fields are tolerated.
+ */
+public class PartitionMarkDoneStateJsonSerde
+ implements JsonSerializer,
+ JsonDeserializer {
+
+ public static final PartitionMarkDoneStateJsonSerde INSTANCE =
+ new PartitionMarkDoneStateJsonSerde();
+
+ private static final String PARTITIONS_KEY = "partitions";
+ private static final String UPDATE_TIME_KEY = "update_time";
+ private static final String DONE_TIME_KEY = "done_time";
+
+ @Override
+ public void serialize(PartitionMarkDoneState state, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+ generator.writeObjectFieldStart(PARTITIONS_KEY);
+ for (Map.Entry entry :
+ new TreeMap<>(state.getPartitionStates()).entrySet()) {
+ generator.writeObjectFieldStart(String.valueOf(entry.getKey()));
+ generator.writeNumberField(UPDATE_TIME_KEY, entry.getValue().getUpdateTime());
+ generator.writeNumberField(DONE_TIME_KEY, entry.getValue().getDoneTime());
+ generator.writeEndObject();
+ }
+ generator.writeEndObject();
+ generator.writeEndObject();
+ }
+
+ @Override
+ public PartitionMarkDoneState deserialize(JsonNode node) {
+ if (node == null || !node.isObject()) {
+ throw new IllegalArgumentException(
+ "Corrupt partition mark-done state: expected a JSON object but got " + node);
+ }
+
+ Map partitionStates = new HashMap<>();
+ JsonNode partitionsNode = node.get(PARTITIONS_KEY);
+ if (partitionsNode != null && !partitionsNode.isNull()) {
+ if (!partitionsNode.isObject()) {
+ throw new IllegalArgumentException(
+ "Corrupt partition mark-done state: '"
+ + PARTITIONS_KEY
+ + "' must be a JSON object but got "
+ + partitionsNode);
+ }
+ Iterator> fields = partitionsNode.fields();
+ while (fields.hasNext()) {
+ Map.Entry field = fields.next();
+ long partitionId;
+ try {
+ partitionId = Long.parseLong(field.getKey());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Corrupt partition mark-done state: invalid partitionId key '"
+ + field.getKey()
+ + "' in "
+ + PARTITIONS_KEY);
+ }
+ if (partitionId <= 0) {
+ throw new IllegalArgumentException(
+ "Corrupt partition mark-done state: partitionId must be positive but got "
+ + partitionId);
+ }
+ partitionStates.put(
+ partitionId, deserializePartitionState(partitionId, field.getValue()));
+ }
+ }
+
+ return new PartitionMarkDoneState(partitionStates);
+ }
+
+ private static PartitionState deserializePartitionState(long partitionId, JsonNode node) {
+ JsonNode updateTimeNode = node == null ? null : node.get(UPDATE_TIME_KEY);
+ JsonNode doneTimeNode = node == null ? null : node.get(DONE_TIME_KEY);
+ if (node == null
+ || !node.isObject()
+ || updateTimeNode == null
+ || !updateTimeNode.canConvertToLong()
+ || doneTimeNode == null
+ || !doneTimeNode.canConvertToLong()) {
+ throw new IllegalArgumentException(
+ "Corrupt partition mark-done state: partition "
+ + partitionId
+ + " must be an object with numeric '"
+ + UPDATE_TIME_KEY
+ + "' and '"
+ + DONE_TIME_KEY
+ + "' but got "
+ + node);
+ }
+ try {
+ return new PartitionState(updateTimeNode.asLong(), doneTimeNode.asLong());
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Corrupt partition mark-done state: partition "
+ + partitionId
+ + ": "
+ + e.getMessage());
+ }
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java
index 030d3da180..32c1e0004f 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java
@@ -18,19 +18,25 @@
package org.apache.fluss.utils.json;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
/**
* Represents the offsets for all buckets of a table. This class stores the mapping from {@link
- * TableBucket} to their corresponding offsets.
+ * TableBucket} to their corresponding offsets, as well as the keyed tiering-state entries.
*
* This class is used to track the log end offsets for each bucket in a table. It supports both
* non-partitioned tables (where buckets are identified only by bucket id) and partitioned tables
* (where buckets are identified by partition id and bucket id).
*
+ *
It also carries the table-level {@link TieringStateEntry}s (e.g. the partition mark-done
+ * state), passed through without parsing the payloads.
+ *
*
The offsets map contains entries for each bucket that has a valid offset. Missing buckets are
* not included in the map.
*
@@ -47,15 +53,31 @@ public class TableBucketOffsets {
*/
private final Map offsets;
+ /** The keyed tiering-state entries, passed through by the serde without parsing payloads. */
+ private final List tieringStates;
+
/**
- * Creates a new {@link TableBucketOffsets} instance.
+ * Creates a new {@link TableBucketOffsets} instance without tiering states.
*
* @param tableId the table ID that all buckets belong to
* @param offsets the mapping from {@link TableBucket} to their offsets
*/
public TableBucketOffsets(long tableId, Map offsets) {
+ this(tableId, offsets, Collections.emptyList());
+ }
+
+ /**
+ * Creates a new {@link TableBucketOffsets} instance with tiering-state entries.
+ *
+ * @param tableId the table ID that all buckets belong to
+ * @param offsets the mapping from {@link TableBucket} to their offsets
+ * @param tieringStates the keyed tiering-state entries (passed through)
+ */
+ public TableBucketOffsets(
+ long tableId, Map offsets, List tieringStates) {
this.tableId = tableId;
this.offsets = offsets;
+ this.tieringStates = tieringStates;
}
/**
@@ -76,6 +98,11 @@ public Map getOffsets() {
return offsets;
}
+ /** Returns the keyed tiering-state entries; empty when absent. */
+ public List getTieringStates() {
+ return tieringStates;
+ }
+
/**
* Serialize to a JSON byte array.
*
@@ -103,16 +130,25 @@ public boolean equals(Object o) {
return false;
}
TableBucketOffsets that = (TableBucketOffsets) o;
- return tableId == that.tableId && Objects.equals(offsets, that.offsets);
+ return tableId == that.tableId
+ && Objects.equals(offsets, that.offsets)
+ && Objects.equals(tieringStates, that.tieringStates);
}
@Override
public int hashCode() {
- return Objects.hash(tableId, offsets);
+ return Objects.hash(tableId, offsets, tieringStates);
}
@Override
public String toString() {
- return "TableBucketOffsets{" + "tableId=" + tableId + ", offsets=" + offsets + '}';
+ return "TableBucketOffsets{"
+ + "tableId="
+ + tableId
+ + ", offsets="
+ + offsets
+ + ", tieringStates="
+ + tieringStates
+ + '}';
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java
index a177c53f66..d1d7d436c7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java
@@ -18,12 +18,15 @@
package org.apache.fluss.utils.json;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -58,6 +61,8 @@
* "bucket_offsets": array of offsets for non-partitioned table buckets (optional)
* "partition_offsets": array of partition offset objects for partitioned table buckets
* (optional)
+ * "tiering_states": array of keyed tiering-state entries, each {@code {"key", "version",
+ * "payload"}} with the payload embedded as a JSON object but not parsed (optional)
*
*/
public class TableBucketOffsetsJsonSerde
@@ -70,8 +75,13 @@ public class TableBucketOffsetsJsonSerde
private static final String BUCKET_OFFSETS_KEY = "bucket_offsets";
private static final String PARTITION_OFFSETS_KEY = "partition_offsets";
private static final String PARTITION_ID_KEY = "partition_id";
+ private static final String TIERING_STATES_KEY = "tiering_states";
+ private static final String STATE_ENTRY_KEY = "key";
+ private static final String STATE_ENTRY_VERSION = "version";
+ private static final String STATE_ENTRY_PAYLOAD = "payload";
- private static final int VERSION = 1;
+ private static final int VERSION_1 = 1;
+ private static final int CURRENT_VERSION = VERSION_1;
private static final long UNKNOWN_OFFSET = -1;
/**
@@ -94,7 +104,7 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener
throws IOException {
generator.writeStartObject();
long expectedTableId = tableBucketOffsets.getTableId();
- generator.writeNumberField(VERSION_KEY, VERSION);
+ generator.writeNumberField(VERSION_KEY, CURRENT_VERSION);
generator.writeNumberField(TABLE_ID_KEY, expectedTableId);
Map offsets = tableBucketOffsets.getOffsets();
@@ -149,6 +159,38 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener
}
}
+ // embed the tiering-state entries (optional); each payload must be a JSON object, its
+ // content is not parsed.
+ List tieringStates = tableBucketOffsets.getTieringStates();
+ if (!tieringStates.isEmpty()) {
+ generator.writeArrayFieldStart(TIERING_STATES_KEY);
+ for (TieringStateEntry entry : tieringStates) {
+ JsonNode payloadNode;
+ try {
+ payloadNode = JsonSerdeUtils.readTree(entry.getPayload());
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "tiering-state payload of key '"
+ + entry.getStateKey()
+ + "' is not valid JSON",
+ e);
+ }
+ if (payloadNode == null || !payloadNode.isObject()) {
+ throw new IllegalArgumentException(
+ "tiering-state payload of key '"
+ + entry.getStateKey()
+ + "' must be a JSON object");
+ }
+ generator.writeStartObject();
+ generator.writeStringField(STATE_ENTRY_KEY, entry.getStateKey());
+ generator.writeNumberField(STATE_ENTRY_VERSION, entry.getStateVersion());
+ generator.writeFieldName(STATE_ENTRY_PAYLOAD);
+ generator.writeTree(payloadNode);
+ generator.writeEndObject();
+ }
+ generator.writeEndArray();
+ }
+
generator.writeEndObject();
}
@@ -165,7 +207,7 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener
@Override
public TableBucketOffsets deserialize(JsonNode node) {
int version = node.get(VERSION_KEY).asInt();
- if (version != VERSION) {
+ if (version != CURRENT_VERSION) {
throw new IllegalArgumentException("Unsupported version: " + version);
}
@@ -211,7 +253,29 @@ public TableBucketOffsets deserialize(JsonNode node) {
}
}
- return new TableBucketOffsets(tableId, offsets);
+ // read the tiering-state entries (optional); payloads are kept as raw JSON bytes.
+ List tieringStates = Collections.emptyList();
+ JsonNode tieringStatesNode = node.get(TIERING_STATES_KEY);
+ if (tieringStatesNode != null && !tieringStatesNode.isNull()) {
+ tieringStates = new ArrayList<>();
+ for (JsonNode entryNode : tieringStatesNode) {
+ JsonNode keyNode = entryNode.get(STATE_ENTRY_KEY);
+ JsonNode versionNode = entryNode.get(STATE_ENTRY_VERSION);
+ JsonNode payloadNode = entryNode.get(STATE_ENTRY_PAYLOAD);
+ if (keyNode == null || versionNode == null || payloadNode == null) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering-state entry: missing key/version/payload in "
+ + entryNode);
+ }
+ tieringStates.add(
+ new TieringStateEntry(
+ keyNode.asText(),
+ versionNode.asInt(),
+ payloadNode.toString().getBytes(StandardCharsets.UTF_8)));
+ }
+ }
+
+ return new TableBucketOffsets(tableId, offsets, tieringStates);
}
private void serializeBucketLogEndOffset(
diff --git a/fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java b/fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java
new file mode 100644
index 0000000000..545deb2d03
--- /dev/null
+++ b/fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.lake.committer;
+
+import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.lake.committer.PartitionMarkDoneState.NOT_DONE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link PartitionMarkDoneState} and its {@link TieringStateEntry} conversion. */
+class PartitionMarkDoneStateTest {
+
+ @Test
+ void testStateEntryRoundTrip() {
+ Map partitionStates = new HashMap<>();
+ partitionStates.put(5L, new PartitionState(1704153550000L, NOT_DONE));
+ partitionStates.put(7L, new PartitionState(1704153560000L, 1704153570000L));
+ PartitionMarkDoneState state = new PartitionMarkDoneState(partitionStates);
+
+ TieringStateEntry entry = state.toStateEntry();
+ assertThat(entry.getStateKey()).isEqualTo(PartitionMarkDoneState.STATE_KEY);
+ assertThat(entry.getStateVersion()).isEqualTo(PartitionMarkDoneState.CURRENT_VERSION);
+ assertThat(PartitionMarkDoneState.fromStateEntry(entry)).isEqualTo(state);
+
+ // empty state round-trips too (null map normalizes to empty).
+ PartitionMarkDoneState empty = new PartitionMarkDoneState(null);
+ assertThat(PartitionMarkDoneState.fromStateEntry(empty.toStateEntry())).isEqualTo(empty);
+ }
+
+ @Test
+ void testPartitionStateValidation() {
+ assertThat(new PartitionState(1000L, NOT_DONE).isDone()).isFalse();
+ assertThat(new PartitionState(1000L, 2000L).isDone()).isTrue();
+ assertThatThrownBy(() -> new PartitionState(-5L, NOT_DONE))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("updateTime");
+ // doneTime 0 and negatives other than the NOT_DONE sentinel are invalid.
+ assertThatThrownBy(() -> new PartitionState(1000L, 0L))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("doneTime");
+ assertThatThrownBy(() -> new PartitionState(1000L, -2L))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("doneTime");
+ }
+
+ @Test
+ void testFromStateEntryRejectsWrongKey() {
+ TieringStateEntry entry =
+ new TieringStateEntry("other-key", 1, "{}".getBytes(StandardCharsets.UTF_8));
+ assertThatThrownBy(() -> PartitionMarkDoneState.fromStateEntry(entry))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Expected state key");
+ }
+
+ @Test
+ void testFromStateEntryRejectsNewerVersion() {
+ // a newer build's payload cannot be interpreted; the entry must be passed through.
+ TieringStateEntry entry =
+ new TieringStateEntry(
+ PartitionMarkDoneState.STATE_KEY,
+ PartitionMarkDoneState.CURRENT_VERSION + 1,
+ "{}".getBytes(StandardCharsets.UTF_8));
+ assertThatThrownBy(() -> PartitionMarkDoneState.fromStateEntry(entry))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("pass the entry through unchanged");
+ }
+
+ @Test
+ void testFromStateEntryRejectsCorruptPayload() {
+ assertThatThrownBy(() -> PartitionMarkDoneState.fromStateEntry(markDoneEntry("[1,2]")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("expected a JSON object");
+ assertThatThrownBy(
+ () ->
+ PartitionMarkDoneState.fromStateEntry(
+ markDoneEntry("{\"partitions\":{\"abc\":{}}}")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("invalid partitionId");
+ // record missing done_time
+ assertThatThrownBy(
+ () ->
+ PartitionMarkDoneState.fromStateEntry(
+ markDoneEntry(
+ "{\"partitions\":{\"5\":{\"update_time\":1}}}")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("done_time");
+ // invalid done_time value, rejected by the PartitionState constructor with context
+ assertThatThrownBy(
+ () ->
+ PartitionMarkDoneState.fromStateEntry(
+ markDoneEntry(
+ "{\"partitions\":{\"5\":{\"update_time\":1,\"done_time\":0}}}")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("partition 5");
+ }
+
+ @Test
+ void testPayloadTolerantToUnknownFields() {
+ // unknown fields (added by a compatible newer build) are ignored at both levels.
+ PartitionMarkDoneState state =
+ PartitionMarkDoneState.fromStateEntry(
+ markDoneEntry(
+ "{\"partitions\":{\"5\":{\"update_time\":1000,\"done_time\":-1,"
+ + "\"max_timestamp\":9}},\"future_field\":123}"));
+ assertThat(state.getPartitionStates())
+ .containsOnlyKeys(5L)
+ .containsEntry(5L, new PartitionState(1000L, NOT_DONE));
+ }
+
+ @Test
+ void testEntryEnvelopeValidation() {
+ byte[] payload = "{}".getBytes(StandardCharsets.UTF_8);
+ assertThatThrownBy(() -> new TieringStateEntry("", 1, payload))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new TieringStateEntry("key", 0, payload))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new TieringStateEntry("key", 1, null))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ private static TieringStateEntry markDoneEntry(String payloadJson) {
+ return new TieringStateEntry(
+ PartitionMarkDoneState.STATE_KEY,
+ PartitionMarkDoneState.CURRENT_VERSION,
+ payloadJson.getBytes(StandardCharsets.UTF_8));
+ }
+}
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java
index 831171af2f..23acb53248 100644
--- a/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java
@@ -18,12 +18,20 @@
package org.apache.fluss.utils.json;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
/** Test for {@link TableBucketOffsetsJsonSerde}. */
class TableBucketOffsetsJsonSerdeTest extends JsonSerdeTestBase {
@@ -74,12 +82,32 @@ protected TableBucketOffsets[] createObjects() {
TableBucketOffsets tableBucketOffsets5 =
new TableBucketOffsets(tableId, bucketLogEndOffset);
+ // Test case 6: Partition table with a tiering-state entry
+ tableId = 8;
+ bucketLogEndOffset = new HashMap<>();
+ bucketLogEndOffset.put(new TableBucket(tableId, 5L, 0), 100L);
+ bucketLogEndOffset.put(new TableBucket(tableId, 5L, 1), 230L);
+ bucketLogEndOffset.put(new TableBucket(tableId, 4L, 0), 420L);
+ TableBucketOffsets tableBucketOffsets6 =
+ new TableBucketOffsets(
+ tableId,
+ bucketLogEndOffset,
+ Collections.singletonList(
+ new PartitionMarkDoneState(
+ Collections.singletonMap(
+ 5L,
+ new PartitionState(
+ 1704153550000L,
+ PartitionMarkDoneState.NOT_DONE)))
+ .toStateEntry()));
+
return new TableBucketOffsets[] {
tableBucketOffsets1,
tableBucketOffsets2,
tableBucketOffsets3,
tableBucketOffsets4,
tableBucketOffsets5,
+ tableBucketOffsets6,
};
}
@@ -102,7 +130,67 @@ protected String[] expectedJsons() {
// Test case 4: Partition table with consecutive bucket ids
"{\"version\":1,\"table_id\":6,\"partition_offsets\":[{\"partition_id\":1,\"bucket_offsets\":[100,200]},{\"partition_id\":2,\"bucket_offsets\":[300,400]}]}",
// Test case 5: Partition table with missing bucket ids
- "{\"version\":1,\"table_id\":7,\"partition_offsets\":[{\"partition_id\":1,\"bucket_offsets\":[100,-1,300]},{\"partition_id\":2,\"bucket_offsets\":[-1,400,-1,600]}]}"
+ "{\"version\":1,\"table_id\":7,\"partition_offsets\":[{\"partition_id\":1,\"bucket_offsets\":[100,-1,300]},{\"partition_id\":2,\"bucket_offsets\":[-1,400,-1,600]}]}",
+ // Test case 6: Partition table with a tiering-state entry
+ "{\"version\":1,\"table_id\":8,\"partition_offsets\":[{\"partition_id\":4,\"bucket_offsets\":[420]},{\"partition_id\":5,\"bucket_offsets\":[100,230]}],\"tiering_states\":[{\"key\":\"fluss.partition-mark-done\",\"version\":1,\"payload\":{\"partitions\":{\"5\":{\"update_time\":1704153550000,\"done_time\":-1}}}}]}"
};
}
+
+ /**
+ * Test that a malformed (non-JSON) or non-object tiering-state payload is rejected on
+ * serialization, so a bad payload never corrupts the persisted offsets file. The happy path
+ * (with/without tiering states, exact JSON and round-trip) is covered by {@link
+ * #createObjects()} / {@link #expectedJsons()}.
+ */
+ @Test
+ void testSerializeRejectsBadTieringStatePayload() {
+ Map offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 1L, 0), 100L);
+
+ assertThatThrownBy(
+ () ->
+ new TableBucketOffsets(
+ 1L,
+ offsets,
+ Collections.singletonList(
+ new TieringStateEntry(
+ "some-key",
+ 1,
+ "{not-json"
+ .getBytes(
+ StandardCharsets
+ .UTF_8))))
+ .toJsonBytes())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("not valid JSON");
+
+ assertThatThrownBy(
+ () ->
+ new TableBucketOffsets(
+ 1L,
+ offsets,
+ Collections.singletonList(
+ new TieringStateEntry(
+ "some-key",
+ 1,
+ "[1,2,3]"
+ .getBytes(
+ StandardCharsets
+ .UTF_8))))
+ .toJsonBytes())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a JSON object");
+ }
+
+ /** A tiering-state entry missing key/version/payload fails as a controlled parse error. */
+ @Test
+ void testDeserializeRejectsIncompleteTieringStateEntry() {
+ String json = "{\"version\":1,\"table_id\":1,\"tiering_states\":[{\"key\":\"some-key\"}]}";
+ assertThatThrownBy(
+ () ->
+ TableBucketOffsets.fromJsonBytes(
+ json.getBytes(StandardCharsets.UTF_8)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("missing key/version/payload");
+ }
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
index e9f18b3d67..b8a43931b4 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
@@ -70,7 +70,8 @@ public enum ApiKeys {
COMMIT_REMOTE_LOG_MANIFEST(1027, 0, 0, PRIVATE),
NOTIFY_REMOTE_LOG_OFFSETS(1028, 0, 0, PRIVATE),
NOTIFY_KV_SNAPSHOT_OFFSET(1029, 0, 0, PRIVATE),
- COMMIT_LAKE_TABLE_SNAPSHOT(1030, 0, 0, PRIVATE),
+ // Version 1: PbLakeTableSnapshotMetadata may carry tiering_epoch for fencing.
+ COMMIT_LAKE_TABLE_SNAPSHOT(1030, 0, 1, PRIVATE),
NOTIFY_LAKE_TABLE_OFFSET(1031, 0, 0, PRIVATE),
GET_LAKE_SNAPSHOT(1032, 0, 0, PUBLIC),
LIMIT_SCAN(1033, 0, 0, PUBLIC),
@@ -97,7 +98,8 @@ public enum ApiKeys {
REBALANCE(1049, 0, 0, PUBLIC),
LIST_REBALANCE_PROGRESS(1050, 0, 0, PUBLIC),
CANCEL_REBALANCE(1051, 0, 0, PUBLIC),
- PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE),
+ // Version 1: request may carry keyed tiering-state entries (for capability detection).
+ PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 1, PRIVATE),
REGISTER_PRODUCER_OFFSETS(1053, 0, 0, PUBLIC),
GET_PRODUCER_OFFSETS(1054, 0, 0, PUBLIC),
DELETE_PRODUCER_OFFSETS(1055, 0, 0, PUBLIC),
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 2e6104b4cf..bb12b70180 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -477,6 +477,8 @@ message GetLakeSnapshotResponse {
required int64 table_id = 1;
required int64 snapshotId = 2;
repeated PbLakeSnapshotForBucket bucket_snapshots = 3;
+ // The keyed tiering-state entries (e.g. partition mark-done state), passed through unparsed.
+ repeated PbTieringStateEntry tiering_states = 4;
}
message GetFileSystemSecurityTokenRequest {
@@ -1250,6 +1252,8 @@ message PbRebalancePlanForBucket {
message PbLakeTableSnapshotMetadata {
required int64 table_id = 1;
+ // The lake snapshot id; may repeat across entries when a state-only round (no new lake commit)
+ // reuses it. Reads resolve to the latest entry for a given id.
required int64 snapshot_id = 2;
required string tiered_bucket_offsets_file_path = 3;
optional string readable_bucket_offsets_file_path = 4;
@@ -1257,6 +1261,9 @@ message PbLakeTableSnapshotMetadata {
// 1. If set, the system will keep all snapshots in the range [earliest_snapshot_id_to_keep, current_snapshot_id].
// 2. If not set, the system defaults to a "Single Snapshot Retention" policy, keeping only the snapshot specified in this request.
optional int64 earliest_snapshot_id_to_keep = 5;
+ // Tiering assignment epoch for fencing (optional for compatibility). When present, the commit is
+ // rejected unless it matches the table's current epoch. Requires COMMIT api version >= 1.
+ optional int64 tiering_epoch = 6;
}
message PbLakeTableSnapshotInfo {
@@ -1289,6 +1296,16 @@ message PbTableOffsets {
required int64 table_id = 1;
required PbTablePath table_path = 2;
repeated PbBucketOffset bucket_offsets = 3;
+ // The keyed tiering-state entries (e.g. partition mark-done state), passed through unparsed.
+ repeated PbTieringStateEntry tiering_states = 4;
+}
+
+// A keyed, versioned tiering-state entry carried by the lake offsets file. The payload is an
+// opaque JSON object owned by the state key (see TieringStateEntry); servers never parse it.
+message PbTieringStateEntry {
+ required string state_key = 1;
+ required int32 state_version = 2;
+ required bytes payload = 3;
}
message PbBucketOffset {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index c5b90f5904..60a0aa5cf8 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -2300,6 +2300,11 @@ private void handleCommitLakeTableSnapshotV2(
throw new FlussRuntimeException(
"Lake snapshot metadata is null for table " + tableId);
}
+ // Fencing: reject a writer from a stale tiering assignment.
+ if (snapshot.getTieringEpoch() != null) {
+ lakeTableTieringManager.validateTieringEpoch(
+ tableId, snapshot.getTieringEpoch());
+ }
lakeTableHelper.registerLakeTableSnapshotV2(
tableId,
snapshot.getLakeSnapshotMetadata(),
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
index 9498c45ecd..31d62e7577 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
@@ -516,6 +516,18 @@ public void renewTieringHeartbeat(long tableId, long tieringEpoch) {
});
}
+ /**
+ * Validates a lake commit's tiering epoch against the coordinator's current assignment epoch,
+ * fencing off a writer from a stale assignment. The epoch is in-memory (reset on restart), so
+ * this is an exact match; a mismatched writer must re-acquire the table via heartbeat first.
+ *
+ * @throws TableNotExistException if the table is not (or no longer) a lake table
+ * @throws FencedTieringEpochException if the epoch does not match the current epoch
+ */
+ public void validateTieringEpoch(long tableId, long tieringEpoch) {
+ inReadLock(lock, () -> validateTieringServiceRequest(tableId, tieringEpoch));
+ }
+
private void validateTieringServiceRequest(long tableId, long tieringEpoch) {
Long currentEpoch = tableTierEpoch.get(tableId);
// the table has been dropped, return false
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java b/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java
index 9d88be9909..1d88552fd0 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java
@@ -64,7 +64,8 @@ public void addTableSnapshot(
@Nullable LakeTableSnapshot lakeTableSnapshot,
@Nullable Map tableMaxTieredTimestamps,
@Nullable LakeTable.LakeSnapshotMetadata lakeSnapshotMetadata,
- @Nullable Long earliestSnapshotIDToKeep) {
+ @Nullable Long earliestSnapshotIDToKeep,
+ @Nullable Long tieringEpoch) {
snapshotMap.put(
tableId,
new CommitLakeTableSnapshot(
@@ -73,7 +74,8 @@ public void addTableSnapshot(
? tableMaxTieredTimestamps
: Collections.emptyMap(),
lakeSnapshotMetadata,
- earliestSnapshotIDToKeep));
+ earliestSnapshotIDToKeep,
+ tieringEpoch));
}
/**
@@ -149,15 +151,20 @@ public static class CommitLakeTableSnapshot {
// The earliest snapshot ID to keep for Paimon DV tables. Null for non-Paimon-DV tables.
@Nullable private final Long earliestSnapshotIDToKeep;
+ // Tiering assignment epoch for fencing; null when the committer did not send it (legacy).
+ @Nullable private final Long tieringEpoch;
+
public CommitLakeTableSnapshot(
@Nullable LakeTableSnapshot lakeTableSnapshot,
@Nullable Map tableMaxTieredTimestamps,
@Nullable LakeTable.LakeSnapshotMetadata lakeSnapshotMetadata,
- @Nullable Long earliestSnapshotIDToKeep) {
+ @Nullable Long earliestSnapshotIDToKeep,
+ @Nullable Long tieringEpoch) {
this.lakeTableSnapshot = lakeTableSnapshot;
this.tableMaxTieredTimestamps = tableMaxTieredTimestamps;
this.lakeSnapshotMetadata = lakeSnapshotMetadata;
this.earliestSnapshotIDToKeep = earliestSnapshotIDToKeep;
+ this.tieringEpoch = tieringEpoch;
}
@Nullable
@@ -174,5 +181,10 @@ public LakeTable.LakeSnapshotMetadata getLakeSnapshotMetadata() {
public Long getEarliestSnapshotIDToKeep() {
return earliestSnapshotIDToKeep;
}
+
+ @Nullable
+ public Long getTieringEpoch() {
+ return tieringEpoch;
+ }
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index e96be16c17..58e6a554ff 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -31,6 +31,7 @@
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.fs.token.ObtainedSecurityToken;
import org.apache.fluss.lake.committer.LakeCommitResult;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.AggFunction;
import org.apache.fluss.metadata.AggFunctionType;
import org.apache.fluss.metadata.AggFunctions;
@@ -159,6 +160,7 @@
import org.apache.fluss.rpc.messages.PbTablePath;
import org.apache.fluss.rpc.messages.PbTableStatsReqForBucket;
import org.apache.fluss.rpc.messages.PbTableStatsRespForBucket;
+import org.apache.fluss.rpc.messages.PbTieringStateEntry;
import org.apache.fluss.rpc.messages.PbValue;
import org.apache.fluss.rpc.messages.PbValueList;
import org.apache.fluss.rpc.messages.PrefixLookupRequest;
@@ -1781,7 +1783,8 @@ public static CommitLakeTableSnapshotsData getCommitLakeTableSnapshotData(
entry.getValue(),
tableBucketsMaxTimestamp.get(tableId),
null, // no metadata for V1
- LakeCommitResult.KEEP_LATEST); // V1: keep only latest snapshot
+ LakeCommitResult.KEEP_LATEST, // V1: keep only latest snapshot
+ null); // V1: no tiering epoch
}
// Add V2 format snapshots (current)
@@ -1802,6 +1805,10 @@ public static CommitLakeTableSnapshotsData getCommitLakeTableSnapshotData(
pbLakeTableSnapshotMetadata.hasEarliestSnapshotIdToKeep()
? pbLakeTableSnapshotMetadata.getEarliestSnapshotIdToKeep()
: null;
+ Long tieringEpoch =
+ pbLakeTableSnapshotMetadata.hasTieringEpoch()
+ ? pbLakeTableSnapshotMetadata.getTieringEpoch()
+ : null;
// If this table already exists in builder (from V1), update it; otherwise add new
builder.addTableSnapshot(
@@ -1809,7 +1816,8 @@ public static CommitLakeTableSnapshotsData getCommitLakeTableSnapshotData(
lakeTableInfoByTableId.get(tableId), // may be null for V2-only
tableBucketsMaxTimestamp.get(tableId), // may be null
lakeSnapshotMetadata,
- earliestSnapshotIDToKeep);
+ earliestSnapshotIDToKeep,
+ tieringEpoch);
}
return builder.build();
@@ -1828,7 +1836,17 @@ public static TableBucketOffsets toTableBucketOffsets(PbTableOffsets pbTableOffs
pbBucketOffset.getBucketId());
bucketOffsets.put(tableBucket, pbBucketOffset.getLogEndOffset());
}
- return new TableBucketOffsets(tableId, bucketOffsets);
+
+ // pass through the tiering-state entries unparsed
+ List tieringStates = new ArrayList<>();
+ for (PbTieringStateEntry pbEntry : pbTableOffsets.getTieringStatesList()) {
+ tieringStates.add(
+ new TieringStateEntry(
+ pbEntry.getStateKey(),
+ pbEntry.getStateVersion(),
+ pbEntry.getPayload()));
+ }
+ return new TableBucketOffsets(tableId, bucketOffsets, tieringStates);
}
/**
@@ -1966,6 +1984,15 @@ public static GetLakeSnapshotResponse makeGetLakeSnapshotResponse(
}
}
+ // pass through the tiering-state entries unparsed
+ for (TieringStateEntry entry : lakeTableSnapshot.getTieringStates()) {
+ getLakeTableSnapshotResponse
+ .addTieringState()
+ .setStateKey(entry.getStateKey())
+ .setStateVersion(entry.getStateVersion())
+ .setPayload(entry.getPayload());
+ }
+
return getLakeTableSnapshotResponse;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
index 9e3e183861..7f38ff97ad 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
@@ -21,7 +21,6 @@
import org.apache.fluss.fs.FSDataInputStream;
import org.apache.fluss.fs.FileSystem;
import org.apache.fluss.fs.FsPath;
-import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.server.zk.data.ZkData;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.json.TableBucketOffsets;
@@ -32,7 +31,6 @@
import java.io.IOException;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Objects;
import static org.apache.fluss.metrics.registry.MetricRegistry.LOG;
@@ -53,8 +51,9 @@
*/
public class LakeTable {
- // Version 2 (current):
- // a list of lake snapshot metadata, record the metadata for different lake snapshots
+ // Version 2 (current): an append-only log of lake snapshot metadata. A lake snapshot id may
+ // repeat across entries (a state-only round appends a new entry reusing it); entries are
+ // immutable once written and reads resolve to the latest matching one.
@Nullable private final List lakeSnapshotMetadatas;
// Version 1 (legacy): the full lake table snapshot info stored in ZK, will be null in version2
@@ -108,7 +107,9 @@ public LakeSnapshotMetadata getLatestLakeSnapshotMetadata() {
@Nullable
private LakeSnapshotMetadata getLakeSnapshotMetadata(long snapshotId) {
if (lakeSnapshotMetadatas != null) {
- for (LakeSnapshotMetadata lakeSnapshotMetadata : lakeSnapshotMetadatas) {
+ // Resolve to the latest (last) entry for the id: it may repeat across entries.
+ for (int i = lakeSnapshotMetadatas.size() - 1; i >= 0; i--) {
+ LakeSnapshotMetadata lakeSnapshotMetadata = lakeSnapshotMetadatas.get(i);
if (lakeSnapshotMetadata.snapshotId == snapshotId) {
return lakeSnapshotMetadata;
}
@@ -203,9 +204,12 @@ private LakeTableSnapshot toLakeTableSnapshot(long snapshotId, FsPath offsetFile
FSDataInputStream inputStream = offsetFilePath.getFileSystem().open(offsetFilePath);
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
IOUtils.copyBytes(inputStream, outputStream, true);
- Map logOffsets =
- TableBucketOffsets.fromJsonBytes(outputStream.toByteArray()).getOffsets();
- return new LakeTableSnapshot(snapshotId, logOffsets);
+ TableBucketOffsets tableBucketOffsets =
+ TableBucketOffsets.fromJsonBytes(outputStream.toByteArray());
+ return new LakeTableSnapshot(
+ snapshotId,
+ tableBucketOffsets.getOffsets(),
+ tableBucketOffsets.getTieringStates());
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
index 0e70d80767..8d0feeb2d3 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
@@ -21,6 +21,7 @@
import org.apache.fluss.fs.FSDataOutputStream;
import org.apache.fluss.fs.FileSystem;
import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -33,6 +34,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -63,12 +65,21 @@ public void registerLakeTableSnapshotV1(long tableId, LakeTableSnapshot lakeTabl
Optional optPreviousLakeTable = zkClient.getLakeTable(tableId);
// Merge with previous snapshot if exists
if (optPreviousLakeTable.isPresent()) {
- TableBucketOffsets tableBucketOffsets =
- mergeTableBucketOffsets(
- optPreviousLakeTable.get(),
- new TableBucketOffsets(
- tableId, lakeTableSnapshot.getBucketLogEndOffset()));
- lakeTableSnapshot = new LakeTableSnapshot(tableId, tableBucketOffsets.getOffsets());
+ LakeTableSnapshot previousSnapshot =
+ optPreviousLakeTable.get().getOrReadLatestTableSnapshot();
+ // The legacy (v1) format cannot carry tiering states. This path is not expected to run
+ // on a state-bearing table; if it does (e.g. an old committer), drop the states with a
+ // warning instead of failing the commit.
+ if (!previousSnapshot.getTieringStates().isEmpty()) {
+ LOG.warn(
+ "Dropping tiering states for table {} on a legacy (v1) lake commit; "
+ + "the v1 format cannot store them.",
+ tableId);
+ }
+ Map bucketLogEndOffset =
+ new HashMap<>(previousSnapshot.getBucketLogEndOffset());
+ bucketLogEndOffset.putAll(lakeTableSnapshot.getBucketLogEndOffset());
+ lakeTableSnapshot = new LakeTableSnapshot(tableId, bucketLogEndOffset);
}
zkClient.upsertLakeTable(
tableId, new LakeTable(lakeTableSnapshot), optPreviousLakeTable.isPresent());
@@ -171,13 +182,28 @@ public TableBucketOffsets mergeTableBucketOffsets(
// Merge current with previous one since the current request
// may not carry all buckets for the table. It typically only carries buckets
// that were written after the previous commit.
+ LakeTableSnapshot previousSnapshot = previousLakeTable.getOrReadLatestTableSnapshot();
// merge log end offsets, current will override the previous
Map bucketLogEndOffset =
- new HashMap<>(
- previousLakeTable.getOrReadLatestTableSnapshot().getBucketLogEndOffset());
+ new HashMap<>(previousSnapshot.getBucketLogEndOffset());
bucketLogEndOffset.putAll(newTableBucketOffsets.getOffsets());
- return new TableBucketOffsets(newTableBucketOffsets.getTableId(), bucketLogEndOffset);
+
+ // merge tiering states by key (upsert): entries not carried by the request are inherited
+ // from the previous snapshot, so a writer only sends the keys it wants to update and never
+ // touches states owned by others (including keys it does not understand).
+ Map mergedStates = new LinkedHashMap<>();
+ for (TieringStateEntry entry : previousSnapshot.getTieringStates()) {
+ mergedStates.put(entry.getStateKey(), entry);
+ }
+ for (TieringStateEntry entry : newTableBucketOffsets.getTieringStates()) {
+ mergedStates.put(entry.getStateKey(), entry);
+ }
+
+ return new TableBucketOffsets(
+ newTableBucketOffsets.getTableId(),
+ bucketLogEndOffset,
+ new ArrayList<>(mergedStates.values()));
}
public FsPath storeLakeTableOffsetsFile(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java
index f567a19d36..68c55b490b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java
@@ -18,8 +18,11 @@
package org.apache.fluss.server.zk.data.lake;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -35,9 +38,20 @@ public class LakeTableSnapshot {
// will be null if log offset is unknown such as reading the snapshot of primary key table
private final Map bucketLogEndOffset;
+ // the keyed tiering-state entries; empty when absent.
+ private final List tieringStates;
+
public LakeTableSnapshot(long snapshotId, Map bucketLogEndOffset) {
+ this(snapshotId, bucketLogEndOffset, Collections.emptyList());
+ }
+
+ public LakeTableSnapshot(
+ long snapshotId,
+ Map bucketLogEndOffset,
+ List tieringStates) {
this.snapshotId = snapshotId;
this.bucketLogEndOffset = bucketLogEndOffset;
+ this.tieringStates = tieringStates;
}
public long getSnapshotId() {
@@ -52,6 +66,11 @@ public Map getBucketLogEndOffset() {
return bucketLogEndOffset;
}
+ /** Returns the keyed tiering-state entries; empty when absent. */
+ public List getTieringStates() {
+ return tieringStates;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -62,12 +81,13 @@ public boolean equals(Object o) {
}
LakeTableSnapshot that = (LakeTableSnapshot) o;
return snapshotId == that.snapshotId
- && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset);
+ && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset)
+ && Objects.equals(tieringStates, that.tieringStates);
}
@Override
public int hashCode() {
- return Objects.hash(snapshotId, bucketLogEndOffset);
+ return Objects.hash(snapshotId, bucketLogEndOffset, tieringStates);
}
@Override
@@ -77,6 +97,8 @@ public String toString() {
+ snapshotId
+ ", bucketLogEndOffset="
+ bucketLogEndOffset
+ + ", tieringStates="
+ + tieringStates
+ '}';
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java
index d1e5a52589..95c44cd8ac 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java
@@ -71,6 +71,12 @@ public class LakeTableSnapshotLegacyJsonSerde
@Override
public void serialize(LakeTableSnapshot lakeTableSnapshot, JsonGenerator generator)
throws IOException {
+ // Tiering states are a v2-only feature; the legacy v1 format cannot carry them (fail fast
+ // rather than silently drop them).
+ if (!lakeTableSnapshot.getTieringStates().isEmpty()) {
+ throw new IllegalStateException(
+ "The legacy (v1) lake table format does not support tiering states.");
+ }
generator.writeStartObject();
generator.writeNumberField(VERSION_KEY, VERSION_1);
generator.writeNumberField(SNAPSHOT_ID, lakeTableSnapshot.getSnapshotId());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java
index e5788f32b6..8745604d83 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java
@@ -19,14 +19,26 @@
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
+import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotResponse;
+import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest;
+import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.PbBucketOffset;
import org.apache.fluss.rpc.messages.PbLakeTableOffsetForBucket;
import org.apache.fluss.rpc.messages.PbLakeTableSnapshotInfo;
+import org.apache.fluss.rpc.messages.PbLakeTableSnapshotMetadata;
+import org.apache.fluss.rpc.messages.PbTableOffsets;
+import org.apache.fluss.rpc.messages.PbTieringStateEntry;
+import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotRequest;
+import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotResponse;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.testutils.FlussClusterExtension;
import org.apache.fluss.server.testutils.RpcMessageTestUtils;
@@ -38,6 +50,7 @@
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -136,6 +149,150 @@ private void checkLakeTableDataInZk(long tableId, LakeTableSnapshot expected) th
assertThat(lakeTableSnapshot).isEqualTo(expected);
}
+ /**
+ * Real-cluster RPC coverage: PREPARE embeds the state in the offsets file, COMMIT registers it
+ * (fencing a stale epoch), and GetLakeSnapshot reads it back. A state-only round reuses the
+ * previous lake snapshot id, and latest/by-id/readable resolve to the latest entry for that id.
+ */
+ @Test
+ void testCommitTieringStateAndReadBackViaRpc() throws Exception {
+ long tableId = createLogTable();
+ CoordinatorGateway coordinatorGateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ Map offsets = new HashMap<>();
+ for (int bucket = 0; bucket < BUCKET_NUM; bucket++) {
+ offsets.put(new TableBucket(tableId, bucket), (bucket + 1) * 10L);
+ }
+
+ // Round 1 (normal round): PREPARE carries the state into the offsets file; COMMIT registers
+ // snapshot 1 with the current tiering epoch (0 for a freshly created lake table).
+ long snapshotId = 1L;
+ PartitionMarkDoneState state1 = stateOf(1000L, PartitionMarkDoneState.NOT_DONE);
+ String path1 = prepareOffsetsFile(coordinatorGateway, tableId, offsets, state1);
+ assertThat(commit(coordinatorGateway, tableId, snapshotId, path1, 0L).hasErrorCode())
+ .isFalse();
+
+ // GetLakeSnapshot latest / by-id / readable all report state1.
+ assertThat(readState(coordinatorGateway, latestRequest())).isEqualTo(state1);
+ assertThat(readState(coordinatorGateway, byIdRequest(snapshotId))).isEqualTo(state1);
+ assertThat(readState(coordinatorGateway, readableRequest())).isEqualTo(state1);
+
+ // Fencing: a commit carrying a stale/wrong epoch is rejected per-table.
+ assertThat(commit(coordinatorGateway, tableId, snapshotId, path1, 999L).hasErrorCode())
+ .isTrue();
+
+ // Round 2 (state-only round): no new lake commit, reuse snapshot id 1, only the state
+ // advances (partition 1 marked done). It appends another entry with the same id; reads
+ // resolve to the latest.
+ PartitionMarkDoneState state2 = stateOf(1000L, 2000L);
+ String path2 = prepareOffsetsFile(coordinatorGateway, tableId, offsets, state2);
+ assertThat(commit(coordinatorGateway, tableId, snapshotId, path2, 0L).hasErrorCode())
+ .isFalse();
+
+ GetLakeSnapshotResponse latest = coordinatorGateway.getLakeSnapshot(latestRequest()).get();
+ // the lake snapshot id is unchanged (state-only round reused it)
+ assertThat(latest.getSnapshotId()).isEqualTo(snapshotId);
+ // latest and by-id both resolve deterministically to the newest state for that id
+ assertThat(parseState(latest)).isEqualTo(state2);
+ assertThat(readState(coordinatorGateway, byIdRequest(snapshotId))).isEqualTo(state2);
+ }
+
+ private static PartitionMarkDoneState stateOf(long updateTime, long doneTime) {
+ return new PartitionMarkDoneState(
+ Collections.singletonMap(1L, new PartitionState(updateTime, doneTime)));
+ }
+
+ private static String prepareOffsetsFile(
+ CoordinatorGateway gateway,
+ long tableId,
+ Map offsets,
+ PartitionMarkDoneState state)
+ throws Exception {
+ PrepareLakeTableSnapshotRequest request = new PrepareLakeTableSnapshotRequest();
+ PbTableOffsets pbTableOffsets = request.addBucketOffset();
+ pbTableOffsets.setTableId(tableId);
+ pbTableOffsets
+ .setTablePath()
+ .setDatabaseName(DATA1_TABLE_PATH.getDatabaseName())
+ .setTableName(DATA1_TABLE_PATH.getTableName());
+ TieringStateEntry entry = state.toStateEntry();
+ pbTableOffsets
+ .addTieringState()
+ .setStateKey(entry.getStateKey())
+ .setStateVersion(entry.getStateVersion())
+ .setPayload(entry.getPayload());
+ for (Map.Entry offsetEntry : offsets.entrySet()) {
+ PbBucketOffset pbBucketOffset = pbTableOffsets.addBucketOffset();
+ pbBucketOffset.setBucketId(offsetEntry.getKey().getBucket());
+ pbBucketOffset.setLogEndOffset(offsetEntry.getValue());
+ }
+ PrepareLakeTableSnapshotResponse response = gateway.prepareLakeTableSnapshot(request).get();
+ return response.getPrepareLakeTableRespsList().get(0).getLakeTableOffsetsPath();
+ }
+
+ private static org.apache.fluss.rpc.messages.PbCommitLakeTableSnapshotRespForTable commit(
+ CoordinatorGateway gateway,
+ long tableId,
+ long snapshotId,
+ String offsetsPath,
+ long tieringEpoch)
+ throws Exception {
+ CommitLakeTableSnapshotRequest request = new CommitLakeTableSnapshotRequest();
+ PbLakeTableSnapshotMetadata metadata = request.addLakeTableSnapshotMetadata();
+ metadata.setTableId(tableId);
+ metadata.setSnapshotId(snapshotId);
+ metadata.setTieredBucketOffsetsFilePath(offsetsPath);
+ // make it readable so getReadableLakeSnapshot returns it
+ metadata.setReadableBucketOffsetsFilePath(offsetsPath);
+ // keep all previous entries so a reused snapshot id yields multiple entries
+ metadata.setEarliestSnapshotIdToKeep(-1L);
+ metadata.setTieringEpoch(tieringEpoch);
+ CommitLakeTableSnapshotResponse response = gateway.commitLakeTableSnapshot(request).get();
+ return response.getTableRespsList().get(0);
+ }
+
+ private static PartitionMarkDoneState readState(
+ CoordinatorGateway gateway, GetLakeSnapshotRequest request) throws Exception {
+ return parseState(gateway.getLakeSnapshot(request).get());
+ }
+
+ private static PartitionMarkDoneState parseState(GetLakeSnapshotResponse response) {
+ for (PbTieringStateEntry pbEntry : response.getTieringStatesList()) {
+ if (PartitionMarkDoneState.STATE_KEY.equals(pbEntry.getStateKey())) {
+ return PartitionMarkDoneState.fromStateEntry(
+ new TieringStateEntry(
+ pbEntry.getStateKey(),
+ pbEntry.getStateVersion(),
+ pbEntry.getPayload()));
+ }
+ }
+ throw new AssertionError("partition mark-done state not found in response");
+ }
+
+ private static GetLakeSnapshotRequest latestRequest() {
+ return newGetLakeSnapshotRequest();
+ }
+
+ private static GetLakeSnapshotRequest byIdRequest(long snapshotId) {
+ GetLakeSnapshotRequest request = newGetLakeSnapshotRequest();
+ request.setSnapshotId(snapshotId);
+ return request;
+ }
+
+ private static GetLakeSnapshotRequest readableRequest() {
+ GetLakeSnapshotRequest request = newGetLakeSnapshotRequest();
+ request.setReadable(true);
+ return request;
+ }
+
+ private static GetLakeSnapshotRequest newGetLakeSnapshotRequest() {
+ GetLakeSnapshotRequest request = new GetLakeSnapshotRequest();
+ request.setTablePath()
+ .setDatabaseName(DATA1_TABLE_PATH.getDatabaseName())
+ .setTableName(DATA1_TABLE_PATH.getTableName());
+ return request;
+ }
+
private static CommitLakeTableSnapshotRequest genCommitLakeTableSnapshotRequest(
long tableId, int buckets, long snapshotId, long logEndOffset, long maxTimestamp) {
CommitLakeTableSnapshotRequest commitLakeTableSnapshotRequest =
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java
new file mode 100644
index 0000000000..e697dc2978
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.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.server.utils;
+
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.PbTableOffsets;
+import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that {@link ServerRpcMessageUtils} omits the tiering states when absent (PREPARE read
+ * yields empty, GET fill leaves the field unset). The present-state passthrough is covered
+ * end-to-end by {@code CommitLakeTableSnapshotITCase}.
+ */
+class ServerRpcMessageUtilsTieringStateTest {
+
+ @Test
+ void testAbsentTieringStates() {
+ // PREPARE read without tiering states -> empty.
+ PbTableOffsets pbTableOffsets = new PbTableOffsets();
+ pbTableOffsets.setTableId(2L);
+ pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t");
+ pbTableOffsets.addBucketOffset().setBucketId(0).setLogEndOffset(100L);
+ assertThat(ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets).getTieringStates())
+ .isEmpty();
+
+ // GET fill without tiering states -> response omits them.
+ Map bucketOffsets = new HashMap<>();
+ bucketOffsets.put(new TableBucket(2L, 0), 100L);
+ GetLakeSnapshotResponse response =
+ ServerRpcMessageUtils.makeGetLakeSnapshotResponse(
+ 2L, new LakeTableSnapshot(9L, bucketOffsets));
+ assertThat(response.getTieringStatesCount()).isZero();
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java
index 369a53062d..c8d11cc7e8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.zk.data;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshotLegacyJsonSerde;
@@ -26,10 +27,12 @@
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link LakeTableSnapshotLegacyJsonSerde}. */
class LakeTableSnapshotLegacyJsonSerdeTest extends JsonSerdeTestBase {
@@ -104,4 +107,25 @@ void testBackwardCompatibility() {
LakeTableSnapshotLegacyJsonSerde.INSTANCE);
assertThat(snapshot3.getSnapshotId()).isEqualTo(3);
}
+
+ @Test
+ void testSerializeRejectsTieringStates() {
+ // Tiering states are a version-2 only feature; the legacy v1 format must reject them
+ // rather than silently drop them.
+ Map offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ LakeTableSnapshot withState =
+ new LakeTableSnapshot(
+ 1L,
+ offsets,
+ Collections.singletonList(
+ new TieringStateEntry(
+ "some-key", 1, "{}".getBytes(StandardCharsets.UTF_8))));
+ assertThatThrownBy(
+ () ->
+ JsonSerdeUtils.writeValueAsBytes(
+ withState, LakeTableSnapshotLegacyJsonSerde.INSTANCE))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("does not support tiering states");
+ }
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
index e3b0f78694..52fc9c40fc 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
@@ -24,6 +24,9 @@
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.fs.local.LocalFileSystem;
import org.apache.fluss.lake.committer.LakeCommitResult;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState;
+import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState;
+import org.apache.fluss.lake.committer.TieringStateEntry;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
@@ -42,7 +45,9 @@
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -253,6 +258,147 @@ void testRegisterLakeTableSnapshotWithRetention(@TempDir Path tempDir) throws Ex
.containsExactly(4L, 5L, 6L);
}
+ /**
+ * Verifies the tiering states merge by key (upsert): a request entry replaces the previous
+ * entry with the same key, entries of other keys (even unrecognized ones) are inherited, and
+ * bucket offsets are still merged.
+ */
+ @Test
+ void testMergeUpsertsTieringStateByKey(@TempDir Path tempDir) throws Exception {
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString());
+ long tableId = 100L;
+ TablePath tablePath = TablePath.of("test_db", "markdone_merge_test");
+ zookeeperClient.registerTable(tablePath, createTableReg(tableId));
+
+ // --- Previous snapshot: partitions 1, 2, 3 with a mark-done state and a foreign state ---
+ Map prevOffsets = new HashMap<>();
+ prevOffsets.put(new TableBucket(tableId, 1L, 0), 100L);
+ prevOffsets.put(new TableBucket(tableId, 2L, 0), 200L);
+ prevOffsets.put(new TableBucket(tableId, 3L, 0), 300L);
+ Map prevStates = new HashMap<>();
+ prevStates.put(1L, new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE));
+ prevStates.put(2L, new PartitionState(2000L, PartitionMarkDoneState.NOT_DONE));
+ prevStates.put(3L, new PartitionState(3000L, PartitionMarkDoneState.NOT_DONE));
+ TieringStateEntry foreignEntry =
+ new TieringStateEntry(
+ "future-key", 7, "{\"x\":1}".getBytes(StandardCharsets.UTF_8));
+ FsPath prevPath =
+ lakeTableHelper.storeLakeTableOffsetsFile(
+ tablePath,
+ new TableBucketOffsets(
+ tableId,
+ prevOffsets,
+ Arrays.asList(
+ new PartitionMarkDoneState(prevStates).toStateEntry(),
+ foreignEntry)));
+ lakeTableHelper.registerLakeTableSnapshotV2(
+ tableId, new LakeTable.LakeSnapshotMetadata(1L, prevPath, prevPath));
+ LakeTable previousLakeTable = zookeeperClient.getLakeTable(tableId).get();
+
+ // --- New offsets: advance partitions 1 & 3, carry only the mark-done state (partition 2
+ // now done). ---
+ Map newOffsets = new HashMap<>();
+ newOffsets.put(new TableBucket(tableId, 1L, 0), 150L);
+ newOffsets.put(new TableBucket(tableId, 3L, 0), 350L);
+ Map newStates = new HashMap<>();
+ newStates.put(1L, new PartitionState(1500L, PartitionMarkDoneState.NOT_DONE));
+ newStates.put(2L, new PartitionState(2000L, 2500L));
+ newStates.put(3L, new PartitionState(3500L, PartitionMarkDoneState.NOT_DONE));
+ PartitionMarkDoneState newState = new PartitionMarkDoneState(newStates);
+ TableBucketOffsets merged =
+ lakeTableHelper.mergeTableBucketOffsets(
+ previousLakeTable,
+ new TableBucketOffsets(
+ tableId,
+ newOffsets,
+ Collections.singletonList(newState.toStateEntry())));
+
+ // mark-done entry replaced by the new value; the foreign entry survives untouched
+ assertThat(merged.getTieringStates())
+ .containsExactlyInAnyOrder(newState.toStateEntry(), foreignEntry);
+ // bucket offsets merged (partition 2 kept from previous)
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L);
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 2L, 0), 200L);
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 3L, 0), 350L);
+ }
+
+ /**
+ * Verifies the upsert semantics for absent states: a request without tiering states leaves the
+ * previous entries untouched (a writer that sends nothing touches nothing), while bucket
+ * offsets are still merged.
+ */
+ @Test
+ void testMergePreservesTieringStatesWhenAbsent(@TempDir Path tempDir) throws Exception {
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString());
+ long tableId = 101L;
+ TablePath tablePath = TablePath.of("test_db", "markdone_keep_test");
+ zookeeperClient.registerTable(tablePath, createTableReg(tableId));
+
+ Map prevOffsets = new HashMap<>();
+ prevOffsets.put(new TableBucket(tableId, 1L, 0), 100L);
+ TieringStateEntry prevEntry =
+ new PartitionMarkDoneState(
+ Collections.singletonMap(
+ 1L,
+ new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE)))
+ .toStateEntry();
+ FsPath prevPath =
+ lakeTableHelper.storeLakeTableOffsetsFile(
+ tablePath,
+ new TableBucketOffsets(
+ tableId, prevOffsets, Collections.singletonList(prevEntry)));
+ lakeTableHelper.registerLakeTableSnapshotV2(
+ tableId, new LakeTable.LakeSnapshotMetadata(1L, prevPath, prevPath));
+ LakeTable previousLakeTable = zookeeperClient.getLakeTable(tableId).get();
+
+ Map newOffsets = new HashMap<>();
+ newOffsets.put(new TableBucket(tableId, 1L, 0), 150L);
+ // new request carries no tiering states -> previous entries preserved, offsets merged.
+ TableBucketOffsets merged =
+ lakeTableHelper.mergeTableBucketOffsets(
+ previousLakeTable, new TableBucketOffsets(tableId, newOffsets));
+
+ assertThat(merged.getTieringStates()).containsExactly(prevEntry);
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L);
+ }
+
+ /**
+ * A legacy (v1) commit on a state-bearing table drops the states (with a warning), not fails.
+ */
+ @Test
+ void testRegisterV1DropsExistingTieringStates(@TempDir Path tempDir) throws Exception {
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString());
+ long tableId = 102L;
+ TablePath tablePath = TablePath.of("test_db", "v1_drop_state_test");
+ zookeeperClient.registerTable(tablePath, createTableReg(tableId));
+
+ // previous v2 snapshot carrying a tiering state
+ Map offsets = new HashMap<>();
+ offsets.put(new TableBucket(tableId, 0), 100L);
+ TieringStateEntry entry =
+ new PartitionMarkDoneState(
+ Collections.singletonMap(
+ 1L,
+ new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE)))
+ .toStateEntry();
+ FsPath path =
+ lakeTableHelper.storeLakeTableOffsetsFile(
+ tablePath,
+ new TableBucketOffsets(tableId, offsets, Collections.singletonList(entry)));
+ lakeTableHelper.registerLakeTableSnapshotV2(
+ tableId, new LakeTable.LakeSnapshotMetadata(1L, path, path));
+
+ // a v1 commit drops the states without throwing
+ Map newOffsets = new HashMap<>();
+ newOffsets.put(new TableBucket(tableId, 0), 150L);
+ lakeTableHelper.registerLakeTableSnapshotV1(tableId, new LakeTableSnapshot(2L, newOffsets));
+
+ LakeTableSnapshot stored =
+ zookeeperClient.getLakeTable(tableId).get().getOrReadLatestTableSnapshot();
+ assertThat(stored.getTieringStates()).isEmpty();
+ assertThat(stored.getBucketLogEndOffset()).containsEntry(new TableBucket(tableId, 0), 150L);
+ }
+
/** Helper to store offset files and return the FsPath. */
private FsPath storeOffsetFile(
LakeTableHelper helper, TablePath path, long tableId, long offset) throws Exception {