From acdb370e55103d76e0b8f877ec886e4d16496520 Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Wed, 15 Jul 2026 14:36:00 +0800 Subject: [PATCH 1/4] [lake] Carry opaque table-level tiering state in lake offsets file and RPC --- .../fluss/client/metadata/LakeSnapshot.java | 29 ++++ .../client/utils/ClientRpcMessageUtils.java | 6 +- .../client/metadata/LakeSnapshotTest.java | 63 ++++++++ .../lake/committer/LakeTieringTableState.java | 142 ++++++++++++++++++ .../json/LakeTieringTableStateJsonSerde.java | 90 +++++++++++ .../fluss/utils/json/TableBucketOffsets.java | 52 ++++++- .../json/TableBucketOffsetsJsonSerde.java | 35 ++++- .../LakeTieringTableStateJsonSerdeTest.java | 82 ++++++++++ .../json/TableBucketOffsetsJsonSerdeTest.java | 56 ++++++- .../apache/fluss/rpc/protocol/ApiKeys.java | 6 +- fluss-rpc/src/main/proto/FlussApi.proto | 4 + .../server/utils/ServerRpcMessageUtils.java | 12 +- .../fluss/server/zk/data/lake/LakeTable.java | 11 +- .../server/zk/data/lake/LakeTableHelper.java | 18 ++- .../zk/data/lake/LakeTableSnapshot.java | 28 +++- ...ServerRpcMessageUtilsTieringStateTest.java | 89 +++++++++++ .../zk/data/lake/LakeTableHelperTest.java | 86 +++++++++++ 17 files changed, 785 insertions(+), 24 deletions(-) create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java create mode 100644 fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java index 89c7864106..4c87ea8536 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java @@ -18,14 +18,21 @@ package org.apache.fluss.client.metadata; import org.apache.fluss.annotation.PublicEvolving; +import org.apache.fluss.lake.committer.LakeTieringTableState; import org.apache.fluss.metadata.TableBucket; +import javax.annotation.Nullable; + +import java.util.Arrays; import java.util.Collections; import java.util.Map; /** * A class representing the lake snapshot information of a table. It contains: *
  • The snapshot id and the log offset for each bucket. + *
  • The opaque table-level tiering state (see {@link LakeTieringTableState}) for partitioned + * tables, parsed lazily via {@link #getLakeTieringTableState()}. It is {@code null} for + * non-partitioned tables or when talking to an old coordinator that does not report it. * * @since 0.3 */ @@ -37,9 +44,21 @@ public class LakeSnapshot { // the specific log offset of the snapshot private final Map tableBucketsOffset; + // opaque tiering-state JSON bytes; null when absent. parsed lazily (see + // getLakeTieringTableState). + @Nullable private final byte[] lakeTieringTableStateJson; + public LakeSnapshot(long snapshotId, Map tableBucketsOffset) { + this(snapshotId, tableBucketsOffset, null); + } + + public LakeSnapshot( + long snapshotId, + Map tableBucketsOffset, + @Nullable byte[] lakeTieringTableStateJson) { this.snapshotId = snapshotId; this.tableBucketsOffset = tableBucketsOffset; + this.lakeTieringTableStateJson = lakeTieringTableStateJson; } public long getSnapshotId() { @@ -50,6 +69,14 @@ public Map getTableBucketsOffset() { return Collections.unmodifiableMap(tableBucketsOffset); } + /** Parses and returns the tiering state (lazily), or {@code null} if absent. */ + @Nullable + public LakeTieringTableState getLakeTieringTableState() { + return lakeTieringTableStateJson == null + ? null + : LakeTieringTableState.fromJsonBytes(lakeTieringTableStateJson); + } + @Override public String toString() { return "LakeSnapshot{" @@ -57,6 +84,8 @@ public String toString() { + snapshotId + ", tableBucketsOffset=" + tableBucketsOffset + + ", lakeTieringTableStateJson=" + + Arrays.toString(lakeTieringTableStateJson) + '}'; } } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java index 70f940c3cd..62f4740597 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java @@ -286,7 +286,11 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLakeSnapshotResponse respo new TableBucket(tableId, partitionId, pbLakeSnapshotForBucket.getBucketId()); tableBucketsOffset.put(tableBucket, pbLakeSnapshotForBucket.getLogOffset()); } - return new LakeSnapshot(snapshotId, tableBucketsOffset); + + // pass through the opaque tiering-state bytes (parsed lazily by LakeSnapshot). + byte[] lakeTieringTableStateJson = + response.hasTieringStateJson() ? response.getTieringStateJson() : null; + return new LakeSnapshot(snapshotId, tableBucketsOffset, lakeTieringTableStateJson); } public static List toFsPathAndFileName( diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java new file mode 100644 index 0000000000..97522b85b1 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java @@ -0,0 +1,63 @@ +/* + * 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.client.metadata; + +import org.apache.fluss.lake.committer.LakeTieringTableState; +import org.apache.fluss.metadata.TableBucket; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test for {@link LakeSnapshot}, in particular the lazy parsing of the tiering state. */ +class LakeSnapshotTest { + + @Test + void testLazyParse() { + // absent -> null. + assertThat(new LakeSnapshot(1L, Collections.emptyMap()).getLakeTieringTableState()) + .isNull(); + + // present -> parsed lazily. + byte[] json = + new LakeTieringTableState(true, Collections.singletonMap(5L, 1000L)).toJsonBytes(); + LakeTieringTableState state = + new LakeSnapshot(1L, Collections.emptyMap(), json).getLakeTieringTableState(); + assertThat(state).isNotNull(); + assertThat(state.isPartitionDoneInitialized()).isTrue(); + assertThat(state.getPartitionUpdateTimes()).containsEntry(5L, 1000L); + } + + @Test + void testCorruptTieringStateDoesNotBlockBucketOffsets() { + TableBucket bucket = new TableBucket(1L, 0); + Map offsets = Collections.singletonMap(bucket, 100L); + LakeSnapshot snapshot = + new LakeSnapshot(1L, offsets, "{not-json".getBytes(StandardCharsets.UTF_8)); + + // bucket offsets remain accessible even though the tiering state is corrupt. + assertThat(snapshot.getTableBucketsOffset()).containsEntry(bucket, 100L); + // only the explicit state accessor surfaces the parse failure. + assertThatThrownBy(snapshot::getLakeTieringTableState).isInstanceOf(RuntimeException.class); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java new file mode 100644 index 0000000000..542a487faf --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java @@ -0,0 +1,142 @@ +/* + * 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.LakeTieringTableStateJsonSerde; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +/** + * Table-level tiering state carried by the lake offsets file as an opaque {@code tiering_state} + * payload (transported as {@code bytes tiering_state_json} in RPC); owned by the lake side and + * passed through by the offsets serde without parsing. + * + *

    Partition mark-done is the first consumer (delete-on-done model): + * + *

      + *
    • {@code partitionDoneInitialized}: table-level one-shot flag, {@code true} once the + * full-table cold-start back-fill has completed; disambiguates "all done" from "first run". + *
    • {@code partitionUpdateTimes}: last update time of only the not-yet-done partitions, keyed + * by partitionId. A partition is removed once marked done; no {@code doneTime} is persisted. + *
    + * + *

    This is a flat, versioned model. Evolve it by adding fields and bumping {@link + * #CURRENT_VERSION} (e.g. a future watermark mode). To stay safe when a newer and an older tiering + * job run against the same table across rounds, a consumer that reads a {@code version} higher than + * its own {@link #CURRENT_VERSION} MUST NOT overwrite the persisted state (leave it untouched, i.e. + * do not send {@code tiering_state} back), so it never drops fields written by a newer build. + * + * @since 0.9 + */ +@PublicEvolving +public class LakeTieringTableState { + + /** The tiering-state format version 1. */ + public static final int VERSION_1 = 1; + + /** The current tiering-state format version written by this build. */ + public static final int CURRENT_VERSION = VERSION_1; + + private final int version; + private final boolean partitionDoneInitialized; + private final Map partitionUpdateTimes; + + public LakeTieringTableState( + boolean partitionDoneInitialized, Map partitionUpdateTimes) { + this(CURRENT_VERSION, partitionDoneInitialized, partitionUpdateTimes); + } + + public LakeTieringTableState( + int version, boolean partitionDoneInitialized, Map partitionUpdateTimes) { + this.version = version; + this.partitionDoneInitialized = partitionDoneInitialized; + this.partitionUpdateTimes = + partitionUpdateTimes == null + ? Collections.emptyMap() + : new HashMap<>(partitionUpdateTimes); + } + + /** Returns the tiering-state format version. */ + public int getVersion() { + return version; + } + + /** Returns whether the full-table cold-start back-fill has completed. */ + public boolean isPartitionDoneInitialized() { + return partitionDoneInitialized; + } + + /** Returns the last update time of the not-yet-done partitions, keyed by partitionId. */ + public Map getPartitionUpdateTimes() { + return Collections.unmodifiableMap(partitionUpdateTimes); + } + + /** + * Serialize to a JSON byte array. + * + * @see LakeTieringTableStateJsonSerde + */ + public byte[] toJsonBytes() { + return JsonSerdeUtils.writeValueAsBytes(this, LakeTieringTableStateJsonSerde.INSTANCE); + } + + /** + * Deserialize from a JSON byte array. + * + * @see LakeTieringTableStateJsonSerde + */ + public static LakeTieringTableState fromJsonBytes(byte[] json) { + return JsonSerdeUtils.readValue(json, LakeTieringTableStateJsonSerde.INSTANCE); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + LakeTieringTableState that = (LakeTieringTableState) o; + return version == that.version + && partitionDoneInitialized == that.partitionDoneInitialized + && Objects.equals(partitionUpdateTimes, that.partitionUpdateTimes); + } + + @Override + public int hashCode() { + return Objects.hash(version, partitionDoneInitialized, partitionUpdateTimes); + } + + @Override + public String toString() { + return "LakeTieringTableState{" + + "version=" + + version + + ", partitionDoneInitialized=" + + partitionDoneInitialized + + ", partitionUpdateTimes=" + + partitionUpdateTimes + + '}'; + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java new file mode 100644 index 0000000000..ea9be581c3 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java @@ -0,0 +1,90 @@ +/* + * 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.LakeTieringTableState; +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; + +/** + * Json serializer and deserializer for {@link LakeTieringTableState}, e.g. {@code + * {"version":1,"partition_done_initialized":true,"partition_update_times":{"5":1704153550000}}}. + * + *

    Field-level compatibility: unknown fields are ignored and missing fields fall back to + * defaults, and {@code version} is NOT strictly gated on read, so a higher-version payload still + * degrades to the fields this build understands. Writers must honor the version rule documented on + * {@link LakeTieringTableState} (do not overwrite a state whose version is higher than {@link + * LakeTieringTableState#CURRENT_VERSION}). + */ +public class LakeTieringTableStateJsonSerde + implements JsonSerializer, JsonDeserializer { + + public static final LakeTieringTableStateJsonSerde INSTANCE = + new LakeTieringTableStateJsonSerde(); + + private static final String VERSION_KEY = "version"; + private static final String PARTITION_DONE_INITIALIZED_KEY = "partition_done_initialized"; + private static final String PARTITION_UPDATE_TIMES_KEY = "partition_update_times"; + + @Override + public void serialize(LakeTieringTableState state, JsonGenerator generator) throws IOException { + generator.writeStartObject(); + generator.writeNumberField(VERSION_KEY, state.getVersion()); + generator.writeBooleanField( + PARTITION_DONE_INITIALIZED_KEY, state.isPartitionDoneInitialized()); + generator.writeObjectFieldStart(PARTITION_UPDATE_TIMES_KEY); + for (Map.Entry entry : state.getPartitionUpdateTimes().entrySet()) { + generator.writeNumberField(String.valueOf(entry.getKey()), entry.getValue()); + } + generator.writeEndObject(); + generator.writeEndObject(); + } + + @Override + public LakeTieringTableState deserialize(JsonNode node) { + // no strict version gating on read: read best-effort and ignore unknown fields. + int version = + node.has(VERSION_KEY) + ? node.get(VERSION_KEY).asInt() + : LakeTieringTableState.VERSION_1; + + // VERSION_1 fields, read for any version so a higher (unsupported) version degrades to the + // fields this build knows. When evolving, bump CURRENT_VERSION and read the new fields here + // guarded by `if (version >= VERSION_2)`. + boolean partitionDoneInitialized = + node.has(PARTITION_DONE_INITIALIZED_KEY) + && node.get(PARTITION_DONE_INITIALIZED_KEY).asBoolean(); + + Map partitionUpdateTimes = new HashMap<>(); + JsonNode updateTimesNode = node.get(PARTITION_UPDATE_TIMES_KEY); + if (updateTimesNode != null && updateTimesNode.isObject()) { + Iterator> fields = updateTimesNode.fields(); + while (fields.hasNext()) { + Map.Entry field = fields.next(); + partitionUpdateTimes.put(Long.valueOf(field.getKey()), field.getValue().asLong()); + } + } + + return new LakeTieringTableState(version, partitionDoneInitialized, partitionUpdateTimes); + } +} 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..0585f1ba84 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 @@ -20,17 +20,24 @@ import org.apache.fluss.metadata.TableBucket; +import javax.annotation.Nullable; + +import java.util.Arrays; 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 an opaque table-level tiering state. * *

    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 an opaque table-level tiering-state payload (owned by the lake side, e.g. + * {@code LakeTieringTableState}), passed through without parsing and used by e.g. partition + * mark-done. + * *

    The offsets map contains entries for each bucket that has a valid offset. Missing buckets are * not included in the map. * @@ -48,14 +55,34 @@ public class TableBucketOffsets { private final Map offsets; /** - * Creates a new {@link TableBucketOffsets} instance. + * Opaque table-level tiering state (owned by the lake side, e.g. {@code + * LakeTieringTableState}), serialized as JSON bytes and passed through by the serde without + * parsing; {@code null} when absent. + */ + @Nullable private final byte[] tieringStateJson; + + /** + * Creates a new {@link TableBucketOffsets} instance without tiering state. * * @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, null); + } + + /** + * Creates a new {@link TableBucketOffsets} instance with an opaque tiering-state payload. + * + * @param tableId the table ID that all buckets belong to + * @param offsets the mapping from {@link TableBucket} to their offsets + * @param tieringStateJson the opaque tiering-state JSON bytes (nullable, passed through) + */ + public TableBucketOffsets( + long tableId, Map offsets, @Nullable byte[] tieringStateJson) { this.tableId = tableId; this.offsets = offsets; + this.tieringStateJson = tieringStateJson; } /** @@ -76,6 +103,12 @@ public Map getOffsets() { return offsets; } + /** Returns the opaque tiering-state JSON bytes, or {@code null} if absent. */ + @Nullable + public byte[] getTieringStateJson() { + return tieringStateJson; + } + /** * Serialize to a JSON byte array. * @@ -103,16 +136,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) + && Arrays.equals(tieringStateJson, that.tieringStateJson); } @Override public int hashCode() { - return Objects.hash(tableId, offsets); + return 31 * Objects.hash(tableId, offsets) + Arrays.hashCode(tieringStateJson); } @Override public String toString() { - return "TableBucketOffsets{" + "tableId=" + tableId + ", offsets=" + offsets + '}'; + return "TableBucketOffsets{" + + "tableId=" + + tableId + + ", offsets=" + + offsets + + ", tieringStateJson=" + + Arrays.toString(tieringStateJson) + + '}'; } } 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..648144e09c 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 @@ -23,6 +23,7 @@ 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.Comparator; import java.util.HashMap; @@ -70,8 +71,10 @@ 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_STATE_KEY = "tiering_state"; - 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 +97,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 +152,23 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener } } + // embed the opaque tiering-state payload (optional); require a JSON object, content not + // parsed. + byte[] tieringStateJson = tableBucketOffsets.getTieringStateJson(); + if (tieringStateJson != null) { + JsonNode tieringStateNode; + try { + tieringStateNode = JsonSerdeUtils.OBJECT_MAPPER_INSTANCE.readTree(tieringStateJson); + } catch (IOException e) { + throw new IllegalArgumentException("tiering_state payload is not valid JSON", e); + } + if (tieringStateNode == null || !tieringStateNode.isObject()) { + throw new IllegalArgumentException("tiering_state payload must be a JSON object"); + } + generator.writeFieldName(TIERING_STATE_KEY); + generator.writeRawValue(tieringStateNode.toString()); + } + generator.writeEndObject(); } @@ -165,7 +185,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 +231,14 @@ public TableBucketOffsets deserialize(JsonNode node) { } } - return new TableBucketOffsets(tableId, offsets); + // read the opaque tiering-state payload (optional) as raw JSON bytes; absent yields null. + byte[] tieringStateJson = null; + JsonNode tieringStateNode = node.get(TIERING_STATE_KEY); + if (tieringStateNode != null && !tieringStateNode.isNull()) { + tieringStateJson = tieringStateNode.toString().getBytes(StandardCharsets.UTF_8); + } + + return new TableBucketOffsets(tableId, offsets, tieringStateJson); } private void serializeBucketLogEndOffset( diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java new file mode 100644 index 0000000000..4e4c00f3bc --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java @@ -0,0 +1,82 @@ +/* + * 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.LakeTieringTableState; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link LakeTieringTableStateJsonSerde}. */ +class LakeTieringTableStateJsonSerdeTest { + + @Test + void testSerializeFormat() { + Map updateTimes = new HashMap<>(); + updateTimes.put(5L, 1704153550000L); + LakeTieringTableState state = new LakeTieringTableState(true, updateTimes); + + String json = new String(state.toJsonBytes(), StandardCharsets.UTF_8); + assertThat(json) + .isEqualTo( + "{\"version\":1,\"partition_done_initialized\":true," + + "\"partition_update_times\":{\"5\":1704153550000}}"); + } + + @Test + void testRoundTripAndDefaults() { + // populated state round-trips exactly. + Map updateTimes = new HashMap<>(); + updateTimes.put(5L, 1704153550000L); + updateTimes.put(8L, 1704157200000L); + LakeTieringTableState state = new LakeTieringTableState(true, updateTimes); + LakeTieringTableState result = LakeTieringTableState.fromJsonBytes(state.toJsonBytes()); + assertThat(result).isEqualTo(state); + assertThat(result.getVersion()).isEqualTo(LakeTieringTableState.VERSION_1); + assertThat(result.isPartitionDoneInitialized()).isTrue(); + assertThat(result.getPartitionUpdateTimes()) + .containsEntry(5L, 1704153550000L) + .containsEntry(8L, 1704157200000L); + + // missing fields (e.g. "{}") fall back to defaults. + LakeTieringTableState defaults = + LakeTieringTableState.fromJsonBytes("{}".getBytes(StandardCharsets.UTF_8)); + assertThat(defaults.getVersion()).isEqualTo(LakeTieringTableState.VERSION_1); + assertThat(defaults.isPartitionDoneInitialized()).isFalse(); + assertThat(defaults.getPartitionUpdateTimes()).isEmpty(); + } + + @Test + void testHigherVersionToleratedIgnoringUnknownFields() { + // a higher (unsupported) version and unknown fields must not raise; the version is kept and + // the known fields degrade cleanly. + String json = + "{\"version\":99,\"partition_done_initialized\":true," + + "\"partition_update_times\":{\"5\":1000},\"future_field\":\"x\"}"; + LakeTieringTableState state = + LakeTieringTableState.fromJsonBytes(json.getBytes(StandardCharsets.UTF_8)); + assertThat(state.getVersion()).isEqualTo(99); + assertThat(state.isPartitionDoneInitialized()).isTrue(); + assertThat(state.getPartitionUpdateTimes()).containsEntry(5L, 1000L); + } +} 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..c984ab98d5 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,18 @@ package org.apache.fluss.utils.json; +import org.apache.fluss.lake.committer.LakeTieringTableState; 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 +80,26 @@ protected TableBucketOffsets[] createObjects() { TableBucketOffsets tableBucketOffsets5 = new TableBucketOffsets(tableId, bucketLogEndOffset); + // Test case 6: Partition table with an opaque tiering-state payload + 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); + Map partitionUpdateTimes = new HashMap<>(); + partitionUpdateTimes.put(5L, 1704153550000L); + byte[] tieringStateJson = + new LakeTieringTableState(true, partitionUpdateTimes).toJsonBytes(); + TableBucketOffsets tableBucketOffsets6 = + new TableBucketOffsets(tableId, bucketLogEndOffset, tieringStateJson); + return new TableBucketOffsets[] { tableBucketOffsets1, tableBucketOffsets2, tableBucketOffsets3, tableBucketOffsets4, tableBucketOffsets5, + tableBucketOffsets6, }; } @@ -102,7 +122,41 @@ 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 an opaque tiering-state payload + "{\"version\":1,\"table_id\":8,\"partition_offsets\":[{\"partition_id\":4,\"bucket_offsets\":[420]},{\"partition_id\":5,\"bucket_offsets\":[100,230]}],\"tiering_state\":{\"version\":1,\"partition_done_initialized\":true,\"partition_update_times\":{\"5\":1704153550000}}}" }; } + + /** + * 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 state, exact JSON and round-trip) is covered by {@link + * #createObjects()} / {@link #expectedJsons()}. + */ + @Test + void testSerializeRejectsBadTieringStateJson() { + Map offsets = new HashMap<>(); + offsets.put(new TableBucket(1L, 1L, 0), 100L); + + assertThatThrownBy( + () -> + new TableBucketOffsets( + 1L, + offsets, + "{not-json".getBytes(StandardCharsets.UTF_8)) + .toJsonBytes()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("not valid JSON"); + + assertThatThrownBy( + () -> + new TableBucketOffsets( + 1L, + offsets, + "[1,2,3]".getBytes(StandardCharsets.UTF_8)) + .toJsonBytes()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must be a JSON object"); + } } 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 b03e1ec63a..abf43fdd57 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 @@ -69,7 +69,8 @@ public enum ApiKeys { NOTIFY_KV_SNAPSHOT_OFFSET(1029, 0, 0, PRIVATE), COMMIT_LAKE_TABLE_SNAPSHOT(1030, 0, 0, PRIVATE), NOTIFY_LAKE_TABLE_OFFSET(1031, 0, 0, PRIVATE), - GET_LAKE_SNAPSHOT(1032, 0, 0, PUBLIC), + // Version 1: response may carry the opaque tiering_state_json (for capability detection). + GET_LAKE_SNAPSHOT(1032, 0, 1, PUBLIC), LIMIT_SCAN(1033, 0, 0, PUBLIC), // Version 0: Uses lake's encoder for prefix key encoding (legacy behavior). @@ -94,7 +95,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 the opaque tiering_state_json (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 57e96092d4..fd76044dec 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -476,6 +476,8 @@ message GetLakeSnapshotResponse { required int64 table_id = 1; required int64 snapshotId = 2; repeated PbLakeSnapshotForBucket bucket_snapshots = 3; + // Opaque table-level tiering state (see LakeTieringTableState), passed through as JSON bytes. + optional bytes tiering_state_json = 4; } message GetFileSystemSecurityTokenRequest { @@ -1284,6 +1286,8 @@ message PbTableOffsets { required int64 table_id = 1; required PbTablePath table_path = 2; repeated PbBucketOffset bucket_offsets = 3; + // Opaque table-level tiering state (see LakeTieringTableState), passed through as JSON bytes. + optional bytes tiering_state_json = 4; } message PbBucketOffset { 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 1e5e914c0d..38ea6f2f72 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 @@ -1822,7 +1822,11 @@ public static TableBucketOffsets toTableBucketOffsets(PbTableOffsets pbTableOffs pbBucketOffset.getBucketId()); bucketOffsets.put(tableBucket, pbBucketOffset.getLogEndOffset()); } - return new TableBucketOffsets(tableId, bucketOffsets); + + // pass through the opaque tiering-state payload + byte[] tieringStateJson = + pbTableOffsets.hasTieringStateJson() ? pbTableOffsets.getTieringStateJson() : null; + return new TableBucketOffsets(tableId, bucketOffsets, tieringStateJson); } /** @@ -1960,6 +1964,12 @@ public static GetLakeSnapshotResponse makeGetLakeSnapshotResponse( } } + // pass through the opaque tiering-state payload + if (lakeTableSnapshot.getTieringStateJson() != null) { + getLakeTableSnapshotResponse.setTieringStateJson( + lakeTableSnapshot.getTieringStateJson()); + } + 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..d2659862b1 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; @@ -203,9 +201,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.getTieringStateJson()); } } 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..60c328e1c1 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 @@ -171,13 +171,25 @@ 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); + + // Unlike the per-bucket offsets above (partial merge), the tiering state is opaque here and + // cannot be field-merged, so it uses whole-value snapshot (PUT) semantics: + // - present (including an empty object) overwrites the whole state; + // - absent keeps the previous state (a legacy/no-op writer never sends it, so keeping the + // previous value is required for rolling-upgrade safety). + byte[] tieringStateJson = + newTableBucketOffsets.getTieringStateJson() != null + ? newTableBucketOffsets.getTieringStateJson() + : previousSnapshot.getTieringStateJson(); + + return new TableBucketOffsets( + newTableBucketOffsets.getTableId(), bucketLogEndOffset, tieringStateJson); } 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..01fe1fd49c 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 @@ -20,6 +20,9 @@ import org.apache.fluss.metadata.TableBucket; +import javax.annotation.Nullable; + +import java.util.Arrays; 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; + // opaque tiering-state JSON bytes; null when absent. + @Nullable private final byte[] tieringStateJson; + public LakeTableSnapshot(long snapshotId, Map bucketLogEndOffset) { + this(snapshotId, bucketLogEndOffset, null); + } + + public LakeTableSnapshot( + long snapshotId, + Map bucketLogEndOffset, + @Nullable byte[] tieringStateJson) { this.snapshotId = snapshotId; this.bucketLogEndOffset = bucketLogEndOffset; + this.tieringStateJson = tieringStateJson; } public long getSnapshotId() { @@ -52,6 +66,12 @@ public Map getBucketLogEndOffset() { return bucketLogEndOffset; } + /** Returns the opaque tiering-state JSON bytes, or {@code null} if absent. */ + @Nullable + public byte[] getTieringStateJson() { + return tieringStateJson; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -62,12 +82,14 @@ public boolean equals(Object o) { } LakeTableSnapshot that = (LakeTableSnapshot) o; return snapshotId == that.snapshotId - && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset); + && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset) + && Arrays.equals(tieringStateJson, that.tieringStateJson); } @Override public int hashCode() { - return Objects.hash(snapshotId, bucketLogEndOffset); + return 31 * Objects.hash(snapshotId, bucketLogEndOffset) + + Arrays.hashCode(tieringStateJson); } @Override @@ -77,6 +99,8 @@ public String toString() { + snapshotId + ", bucketLogEndOffset=" + bucketLogEndOffset + + ", tieringStateJson=" + + Arrays.toString(tieringStateJson) + '}'; } } 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..06a297b5d5 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java @@ -0,0 +1,89 @@ +/* + * 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.lake.committer.LakeTieringTableState; +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.apache.fluss.utils.json.TableBucketOffsets; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for the opaque tiering-state passthrough in {@link ServerRpcMessageUtils}, covering the + * PREPARE read path ({@link ServerRpcMessageUtils#toTableBucketOffsets}) and the GET fill path + * ({@link ServerRpcMessageUtils#makeGetLakeSnapshotResponse}). + */ +class ServerRpcMessageUtilsTieringStateTest { + + @Test + void testTieringStatePassthroughRoundTrip() { + // PREPARE read (toTableBucketOffsets) -> persist -> GET fill (makeGetLakeSnapshotResponse). + long tableId = 1L; + byte[] tieringStateJson = + new LakeTieringTableState(true, Collections.singletonMap(5L, 1704153550000L)) + .toJsonBytes(); + + PbTableOffsets pbTableOffsets = new PbTableOffsets(); + pbTableOffsets.setTableId(tableId); + pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t"); + pbTableOffsets.addBucketOffset().setPartitionId(5L).setBucketId(0).setLogEndOffset(100L); + pbTableOffsets.setTieringStateJson(tieringStateJson); + + TableBucketOffsets offsets = ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets); + assertThat(offsets.getOffsets()).containsEntry(new TableBucket(tableId, 5L, 0), 100L); + assertThat(offsets.getTieringStateJson()).isEqualTo(tieringStateJson); + + LakeTableSnapshot snapshot = + new LakeTableSnapshot(9L, offsets.getOffsets(), offsets.getTieringStateJson()); + GetLakeSnapshotResponse response = + ServerRpcMessageUtils.makeGetLakeSnapshotResponse(tableId, snapshot); + assertThat(response.getTableId()).isEqualTo(tableId); + assertThat(response.getSnapshotId()).isEqualTo(9L); + assertThat(response.hasTieringStateJson()).isTrue(); + assertThat(response.getTieringStateJson()).isEqualTo(tieringStateJson); + } + + @Test + void testAbsentTieringState() { + // PREPARE read without tiering state -> null. + PbTableOffsets pbTableOffsets = new PbTableOffsets(); + pbTableOffsets.setTableId(2L); + pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t"); + pbTableOffsets.addBucketOffset().setBucketId(0).setLogEndOffset(100L); + assertThat(ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets).getTieringStateJson()) + .isNull(); + + // GET fill with a null tiering state -> response omits it. + Map bucketOffsets = new HashMap<>(); + bucketOffsets.put(new TableBucket(2L, 0), 100L); + GetLakeSnapshotResponse response = + ServerRpcMessageUtils.makeGetLakeSnapshotResponse( + 2L, new LakeTableSnapshot(9L, bucketOffsets, null)); + assertThat(response.hasTieringStateJson()).isFalse(); + } +} 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..0d5ec3ea59 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,7 @@ 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.LakeTieringTableState; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; @@ -253,6 +254,91 @@ void testRegisterLakeTableSnapshotWithRetention(@TempDir Path tempDir) throws Ex .containsExactly(4L, 5L, 6L); } + /** + * Verifies the opaque tiering state uses overwrite semantics on merge: a new request carrying + * tiering_state replaces the previous one, while bucket offsets are still merged. + */ + @Test + void testMergeTieringStateOverwrite(@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 tiering 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 prevUpdateTimes = new HashMap<>(); + prevUpdateTimes.put(1L, 1000L); + prevUpdateTimes.put(2L, 2000L); + prevUpdateTimes.put(3L, 3000L); + byte[] prevState = new LakeTieringTableState(true, prevUpdateTimes).toJsonBytes(); + FsPath prevPath = + lakeTableHelper.storeLakeTableOffsetsFile( + tablePath, new TableBucketOffsets(tableId, prevOffsets, prevState)); + lakeTableHelper.registerLakeTableSnapshotV2( + tableId, new LakeTable.LakeSnapshotMetadata(1L, prevPath, prevPath)); + LakeTable previousLakeTable = zookeeperClient.getLakeTable(tableId).get(); + + // --- New offsets: advance partitions 1 & 3, carry a full new tiering state (partition 2 + // done -> dropped). --- + Map newOffsets = new HashMap<>(); + newOffsets.put(new TableBucket(tableId, 1L, 0), 150L); + newOffsets.put(new TableBucket(tableId, 3L, 0), 350L); + Map newUpdateTimes = new HashMap<>(); + newUpdateTimes.put(1L, 1500L); + newUpdateTimes.put(3L, 3500L); + byte[] newState = new LakeTieringTableState(true, newUpdateTimes).toJsonBytes(); + TableBucketOffsets merged = + lakeTableHelper.mergeTableBucketOffsets( + previousLakeTable, new TableBucketOffsets(tableId, newOffsets, newState)); + + // tiering state overwritten by the new value + assertThat(merged.getTieringStateJson()).isEqualTo(newState); + LakeTieringTableState mergedState = + LakeTieringTableState.fromJsonBytes(merged.getTieringStateJson()); + assertThat(mergedState.getPartitionUpdateTimes()).containsOnlyKeys(1L, 3L); + assertThat(mergedState.getPartitionUpdateTimes()).containsEntry(1L, 1500L); + assertThat(mergedState.getPartitionUpdateTimes()).containsEntry(3L, 3500L); + // 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 a new request without tiering state keeps the previous one on merge. */ + @Test + void testMergeKeepsPreviousTieringStateWhenAbsent(@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); + byte[] prevState = + new LakeTieringTableState(true, java.util.Collections.singletonMap(1L, 1000L)) + .toJsonBytes(); + FsPath prevPath = + lakeTableHelper.storeLakeTableOffsetsFile( + tablePath, new TableBucketOffsets(tableId, prevOffsets, prevState)); + 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 state (null) + TableBucketOffsets merged = + lakeTableHelper.mergeTableBucketOffsets( + previousLakeTable, new TableBucketOffsets(tableId, newOffsets, null)); + + assertThat(merged.getTieringStateJson()).isEqualTo(prevState); + assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L); + } + /** Helper to store offset files and return the FsPath. */ private FsPath storeOffsetFile( LakeTableHelper helper, TablePath path, long tableId, long offset) throws Exception { From 3fec3cfd747ee099514e6e346d6d0e27ef0f1dbe Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Wed, 15 Jul 2026 21:19:42 +0800 Subject: [PATCH 2/4] [lake] Tidy tiering-state model and offsets json serde (idiomatic hashCode, writeTree, readTree helper) --- .../lake/committer/LakeTieringTableState.java | 15 --------------- .../apache/fluss/utils/json/JsonSerdeUtils.java | 8 ++++++++ .../fluss/utils/json/TableBucketOffsets.java | 2 +- .../utils/json/TableBucketOffsetsJsonSerde.java | 4 ++-- .../server/zk/data/lake/LakeTableSnapshot.java | 3 +-- 5 files changed, 12 insertions(+), 20 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java index 542a487faf..1528cdbdce 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java @@ -51,10 +51,8 @@ @PublicEvolving public class LakeTieringTableState { - /** The tiering-state format version 1. */ public static final int VERSION_1 = 1; - /** The current tiering-state format version written by this build. */ public static final int CURRENT_VERSION = VERSION_1; private final int version; @@ -76,35 +74,22 @@ public LakeTieringTableState( : new HashMap<>(partitionUpdateTimes); } - /** Returns the tiering-state format version. */ public int getVersion() { return version; } - /** Returns whether the full-table cold-start back-fill has completed. */ public boolean isPartitionDoneInitialized() { return partitionDoneInitialized; } - /** Returns the last update time of the not-yet-done partitions, keyed by partitionId. */ public Map getPartitionUpdateTimes() { return Collections.unmodifiableMap(partitionUpdateTimes); } - /** - * Serialize to a JSON byte array. - * - * @see LakeTieringTableStateJsonSerde - */ public byte[] toJsonBytes() { return JsonSerdeUtils.writeValueAsBytes(this, LakeTieringTableStateJsonSerde.INSTANCE); } - /** - * Deserialize from a JSON byte array. - * - * @see LakeTieringTableStateJsonSerde - */ public static LakeTieringTableState fromJsonBytes(byte[] json) { return JsonSerdeUtils.readValue(json, LakeTieringTableStateJsonSerde.INSTANCE); } 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/TableBucketOffsets.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java index 0585f1ba84..5aea2a84ae 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 @@ -143,7 +143,7 @@ public boolean equals(Object o) { @Override public int hashCode() { - return 31 * Objects.hash(tableId, offsets) + Arrays.hashCode(tieringStateJson); + return Objects.hash(tableId, offsets, Arrays.hashCode(tieringStateJson)); } @Override 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 648144e09c..c72861fd1b 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 @@ -158,7 +158,7 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener if (tieringStateJson != null) { JsonNode tieringStateNode; try { - tieringStateNode = JsonSerdeUtils.OBJECT_MAPPER_INSTANCE.readTree(tieringStateJson); + tieringStateNode = JsonSerdeUtils.readTree(tieringStateJson); } catch (IOException e) { throw new IllegalArgumentException("tiering_state payload is not valid JSON", e); } @@ -166,7 +166,7 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener throw new IllegalArgumentException("tiering_state payload must be a JSON object"); } generator.writeFieldName(TIERING_STATE_KEY); - generator.writeRawValue(tieringStateNode.toString()); + generator.writeTree(tieringStateNode); } generator.writeEndObject(); 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 01fe1fd49c..9fc6894574 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 @@ -88,8 +88,7 @@ public boolean equals(Object o) { @Override public int hashCode() { - return 31 * Objects.hash(snapshotId, bucketLogEndOffset) - + Arrays.hashCode(tieringStateJson); + return Objects.hash(snapshotId, bucketLogEndOffset, Arrays.hashCode(tieringStateJson)); } @Override From 20ccb0c76d69190d28f1bc1e4aa4ea0e118a4787 Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Sat, 18 Jul 2026 09:49:09 +0800 Subject: [PATCH 3/4] [lake] Address review: deterministic reads, strict serde, v1 rejects state, commit epoch fencing --- .../lake/committer/LakeTieringTableState.java | 4 +- .../json/LakeTieringTableStateJsonSerde.java | 97 ++++++++++--- .../LakeTieringTableStateJsonSerdeTest.java | 46 +++++- .../apache/fluss/rpc/protocol/ApiKeys.java | 6 +- fluss-rpc/src/main/proto/FlussApi.proto | 5 + .../CoordinatorEventProcessor.java | 5 + .../coordinator/LakeTableTieringManager.java | 12 ++ .../entity/CommitLakeTableSnapshotsData.java | 18 ++- .../server/utils/ServerRpcMessageUtils.java | 10 +- .../fluss/server/zk/data/lake/LakeTable.java | 9 +- .../server/zk/data/lake/LakeTableHelper.java | 7 +- .../LakeTableSnapshotLegacyJsonSerde.java | 6 + .../CommitLakeTableSnapshotITCase.java | 135 ++++++++++++++++++ ...ServerRpcMessageUtilsTieringStateTest.java | 37 +---- .../LakeTableSnapshotLegacyJsonSerdeTest.java | 19 +++ 15 files changed, 343 insertions(+), 73 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java index 1528cdbdce..1677561707 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java @@ -43,8 +43,8 @@ *

    This is a flat, versioned model. Evolve it by adding fields and bumping {@link * #CURRENT_VERSION} (e.g. a future watermark mode). To stay safe when a newer and an older tiering * job run against the same table across rounds, a consumer that reads a {@code version} higher than - * its own {@link #CURRENT_VERSION} MUST NOT overwrite the persisted state (leave it untouched, i.e. - * do not send {@code tiering_state} back), so it never drops fields written by a newer build. + * its own {@link #CURRENT_VERSION} must treat it as read-only so it never drops fields written by a + * newer build; the serializer enforces this by refusing to write such a state. * * @since 0.9 */ diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java index ea9be581c3..34abdaa200 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java @@ -22,6 +22,7 @@ import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; import java.io.IOException; +import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.Map; @@ -30,11 +31,10 @@ * Json serializer and deserializer for {@link LakeTieringTableState}, e.g. {@code * {"version":1,"partition_done_initialized":true,"partition_update_times":{"5":1704153550000}}}. * - *

    Field-level compatibility: unknown fields are ignored and missing fields fall back to - * defaults, and {@code version} is NOT strictly gated on read, so a higher-version payload still - * degrades to the fields this build understands. Writers must honor the version rule documented on - * {@link LakeTieringTableState} (do not overwrite a state whose version is higher than {@link - * LakeTieringTableState#CURRENT_VERSION}). + *

    Validation is strict for known fields, tolerant for unknown ones. {@code version} is mandatory + * and positive; a corrupt known field fails fast. A higher-than-supported {@code version} is read + * as version-only (so an older build detects it and goes read-only), and serialization refuses it + * so an older build can never overwrite a state written by a newer build. */ public class LakeTieringTableStateJsonSerde implements JsonSerializer, JsonDeserializer { @@ -48,6 +48,16 @@ public class LakeTieringTableStateJsonSerde @Override public void serialize(LakeTieringTableState state, JsonGenerator generator) throws IOException { + // Refuse to write a version this build does not understand (it would drop unparsed newer + // fields); higher-version states are read-only. + if (state.getVersion() > LakeTieringTableState.CURRENT_VERSION) { + throw new IllegalStateException( + "Refusing to serialize tiering state with unsupported version " + + state.getVersion() + + " > " + + LakeTieringTableState.CURRENT_VERSION + + " (read-only)."); + } generator.writeStartObject(); generator.writeNumberField(VERSION_KEY, state.getVersion()); generator.writeBooleanField( @@ -62,26 +72,77 @@ public void serialize(LakeTieringTableState state, JsonGenerator generator) thro @Override public LakeTieringTableState deserialize(JsonNode node) { - // no strict version gating on read: read best-effort and ignore unknown fields. - int version = - node.has(VERSION_KEY) - ? node.get(VERSION_KEY).asInt() - : LakeTieringTableState.VERSION_1; + if (node == null || !node.isObject()) { + throw new IllegalArgumentException( + "Corrupt tiering state: expected a JSON object but got " + node); + } + + // version is mandatory and a positive integer; it gates the rest of the parsing. + JsonNode versionNode = node.get(VERSION_KEY); + if (versionNode == null || !versionNode.isInt() || versionNode.asInt() <= 0) { + throw new IllegalArgumentException( + "Corrupt tiering state: '" + + VERSION_KEY + + "' must be a positive integer but got " + + versionNode); + } + int version = versionNode.asInt(); - // VERSION_1 fields, read for any version so a higher (unsupported) version degrades to the - // fields this build knows. When evolving, bump CURRENT_VERSION and read the new fields here - // guarded by `if (version >= VERSION_2)`. - boolean partitionDoneInitialized = - node.has(PARTITION_DONE_INITIALIZED_KEY) - && node.get(PARTITION_DONE_INITIALIZED_KEY).asBoolean(); + // Higher-than-supported version: read version-only (skip other fields) so an older build + // detects it and stays read-only. + if (version > LakeTieringTableState.CURRENT_VERSION) { + return new LakeTieringTableState(version, false, Collections.emptyMap()); + } + + // partition_done_initialized: optional, must be a boolean when present. + JsonNode initializedNode = node.get(PARTITION_DONE_INITIALIZED_KEY); + if (initializedNode != null && !initializedNode.isBoolean()) { + throw new IllegalArgumentException( + "Corrupt tiering state: '" + + PARTITION_DONE_INITIALIZED_KEY + + "' must be a boolean but got " + + initializedNode); + } + boolean partitionDoneInitialized = initializedNode != null && initializedNode.asBoolean(); + // partition_update_times: optional object; positive-id keys, non-negative values. Map partitionUpdateTimes = new HashMap<>(); JsonNode updateTimesNode = node.get(PARTITION_UPDATE_TIMES_KEY); - if (updateTimesNode != null && updateTimesNode.isObject()) { + if (updateTimesNode != null && !updateTimesNode.isNull()) { + if (!updateTimesNode.isObject()) { + throw new IllegalArgumentException( + "Corrupt tiering state: '" + + PARTITION_UPDATE_TIMES_KEY + + "' must be a JSON object but got " + + updateTimesNode); + } Iterator> fields = updateTimesNode.fields(); while (fields.hasNext()) { Map.Entry field = fields.next(); - partitionUpdateTimes.put(Long.valueOf(field.getKey()), field.getValue().asLong()); + long partitionId; + try { + partitionId = Long.parseLong(field.getKey()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Corrupt tiering state: invalid partitionId key '" + + field.getKey() + + "' in " + + PARTITION_UPDATE_TIMES_KEY); + } + if (partitionId <= 0) { + throw new IllegalArgumentException( + "Corrupt tiering state: partitionId must be positive but got " + + partitionId); + } + JsonNode valueNode = field.getValue(); + if (!valueNode.canConvertToLong() || valueNode.asLong() < 0) { + throw new IllegalArgumentException( + "Corrupt tiering state: update time for partition " + + partitionId + + " must be a non-negative integer but got " + + valueNode); + } + partitionUpdateTimes.put(partitionId, valueNode.asLong()); } } diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java index 4e4c00f3bc..4e6539930a 100644 --- a/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java @@ -26,6 +26,7 @@ import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link LakeTieringTableStateJsonSerde}. */ class LakeTieringTableStateJsonSerdeTest { @@ -58,25 +59,56 @@ void testRoundTripAndDefaults() { .containsEntry(5L, 1704153550000L) .containsEntry(8L, 1704157200000L); - // missing fields (e.g. "{}") fall back to defaults. + // with a version present, missing fields fall back to defaults. LakeTieringTableState defaults = - LakeTieringTableState.fromJsonBytes("{}".getBytes(StandardCharsets.UTF_8)); + LakeTieringTableState.fromJsonBytes( + "{\"version\":1}".getBytes(StandardCharsets.UTF_8)); assertThat(defaults.getVersion()).isEqualTo(LakeTieringTableState.VERSION_1); assertThat(defaults.isPartitionDoneInitialized()).isFalse(); assertThat(defaults.getPartitionUpdateTimes()).isEmpty(); } @Test - void testHigherVersionToleratedIgnoringUnknownFields() { - // a higher (unsupported) version and unknown fields must not raise; the version is kept and - // the known fields degrade cleanly. + void testStrictValidationRejectsCorruptFields() { + // version is mandatory and must be a positive integer; for a known version, corrupt known + // fields must fail fast rather than be silently coerced. + String[] corrupt = { + "{}", + "{\"version\":0}", + "{\"version\":\"x\"}", + "{\"version\":1,\"partition_done_initialized\":\"x\"}", + "{\"version\":1,\"partition_update_times\":5}", + "{\"version\":1,\"partition_update_times\":{\"x\":1}}", + "{\"version\":1,\"partition_update_times\":{\"0\":1}}", + "{\"version\":1,\"partition_update_times\":{\"5\":-1}}" + }; + for (String bad : corrupt) { + assertThatThrownBy( + () -> + LakeTieringTableState.fromJsonBytes( + bad.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IllegalArgumentException.class); + } + } + + @Test + void testHigherVersionIsReadableButNotWritable() { + // Reading a higher (unsupported) version and unknown fields must not raise; only the + // version + // is read (higher-version fields skipped) so an older build can detect it and degrade to + // read-only. String json = "{\"version\":99,\"partition_done_initialized\":true," + "\"partition_update_times\":{\"5\":1000},\"future_field\":\"x\"}"; LakeTieringTableState state = LakeTieringTableState.fromJsonBytes(json.getBytes(StandardCharsets.UTF_8)); assertThat(state.getVersion()).isEqualTo(99); - assertThat(state.isPartitionDoneInitialized()).isTrue(); - assertThat(state.getPartitionUpdateTimes()).containsEntry(5L, 1000L); + // higher-version fields are NOT interpreted. + assertThat(state.isPartitionDoneInitialized()).isFalse(); + assertThat(state.getPartitionUpdateTimes()).isEmpty(); + // Writing it back is refused: an older build must not overwrite a newer-version state. + assertThatThrownBy(state::toJsonBytes) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("read-only"); } } 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 abf43fdd57..d1fc649c70 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 @@ -67,10 +67,10 @@ 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), - // Version 1: response may carry the opaque tiering_state_json (for capability detection). - GET_LAKE_SNAPSHOT(1032, 0, 1, PUBLIC), + GET_LAKE_SNAPSHOT(1032, 0, 0, PUBLIC), LIMIT_SCAN(1033, 0, 0, PUBLIC), // Version 0: Uses lake's encoder for prefix key encoding (legacy behavior). diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index fd76044dec..bcb1c74be7 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -1247,6 +1247,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; @@ -1254,6 +1256,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 { 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 18c4429aff..8d063d61f6 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 @@ -2273,6 +2273,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 e281381de5..355d6acecb 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 @@ -511,6 +511,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 38ea6f2f72..c0d292dd17 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 @@ -1775,7 +1775,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) @@ -1796,6 +1797,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( @@ -1803,7 +1808,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(); 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 d2659862b1..19500a2f06 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 @@ -51,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 @@ -106,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; } 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 60c328e1c1..3464e52b33 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 @@ -68,7 +68,12 @@ public void registerLakeTableSnapshotV1(long tableId, LakeTableSnapshot lakeTabl optPreviousLakeTable.get(), new TableBucketOffsets( tableId, lakeTableSnapshot.getBucketLogEndOffset())); - lakeTableSnapshot = new LakeTableSnapshot(tableId, tableBucketOffsets.getOffsets()); + // State is v2-only: carry it through so the legacy (v1) serde fails fast if present. + lakeTableSnapshot = + new LakeTableSnapshot( + tableId, + tableBucketOffsets.getOffsets(), + tableBucketOffsets.getTieringStateJson()); } zkClient.upsertLakeTable( tableId, new LakeTable(lakeTableSnapshot), optPreviousLakeTable.isPresent()); 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..7d2700efb5 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 state is a v2-only feature; the legacy v1 format cannot carry it (fail fast + // rather than silently drop it). + if (lakeTableSnapshot.getTieringStateJson() != null) { + throw new IllegalStateException( + "The legacy (v1) lake table format does not support tiering state."); + } 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..8bad9e6b8d 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,23 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.committer.LakeTieringTableState; 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.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 +47,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 +146,131 @@ 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; + LakeTieringTableState state1 = + new LakeTieringTableState(true, Collections.singletonMap(1L, 1000L)); + 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. It appends another entry with the same id; reads resolve to the latest. + LakeTieringTableState state2 = + new LakeTieringTableState(true, Collections.singletonMap(1L, 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(LakeTieringTableState.fromJsonBytes(latest.getTieringStateJson())) + .isEqualTo(state2); + assertThat(readState(coordinatorGateway, byIdRequest(snapshotId))).isEqualTo(state2); + } + + private static String prepareOffsetsFile( + CoordinatorGateway gateway, + long tableId, + Map offsets, + LakeTieringTableState 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()); + pbTableOffsets.setTieringStateJson(state.toJsonBytes()); + for (Map.Entry entry : offsets.entrySet()) { + PbBucketOffset pbBucketOffset = pbTableOffsets.addBucketOffset(); + pbBucketOffset.setBucketId(entry.getKey().getBucket()); + pbBucketOffset.setLogEndOffset(entry.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 LakeTieringTableState readState( + CoordinatorGateway gateway, GetLakeSnapshotRequest request) throws Exception { + GetLakeSnapshotResponse response = gateway.getLakeSnapshot(request).get(); + assertThat(response.hasTieringStateJson()).isTrue(); + return LakeTieringTableState.fromJsonBytes(response.getTieringStateJson()); + } + + 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 index 06a297b5d5..07af40e449 100644 --- 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 @@ -18,56 +18,25 @@ package org.apache.fluss.server.utils; -import org.apache.fluss.lake.committer.LakeTieringTableState; 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.apache.fluss.utils.json.TableBucketOffsets; import org.junit.jupiter.api.Test; -import java.util.Collections; import java.util.HashMap; import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; /** - * Tests for the opaque tiering-state passthrough in {@link ServerRpcMessageUtils}, covering the - * PREPARE read path ({@link ServerRpcMessageUtils#toTableBucketOffsets}) and the GET fill path - * ({@link ServerRpcMessageUtils#makeGetLakeSnapshotResponse}). + * Tests that {@link ServerRpcMessageUtils} omits the tiering state when absent (PREPARE read yields + * {@code null}, GET fill leaves the field unset). The present-state passthrough is covered + * end-to-end by {@code CommitLakeTableSnapshotITCase}. */ class ServerRpcMessageUtilsTieringStateTest { - @Test - void testTieringStatePassthroughRoundTrip() { - // PREPARE read (toTableBucketOffsets) -> persist -> GET fill (makeGetLakeSnapshotResponse). - long tableId = 1L; - byte[] tieringStateJson = - new LakeTieringTableState(true, Collections.singletonMap(5L, 1704153550000L)) - .toJsonBytes(); - - PbTableOffsets pbTableOffsets = new PbTableOffsets(); - pbTableOffsets.setTableId(tableId); - pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t"); - pbTableOffsets.addBucketOffset().setPartitionId(5L).setBucketId(0).setLogEndOffset(100L); - pbTableOffsets.setTieringStateJson(tieringStateJson); - - TableBucketOffsets offsets = ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets); - assertThat(offsets.getOffsets()).containsEntry(new TableBucket(tableId, 5L, 0), 100L); - assertThat(offsets.getTieringStateJson()).isEqualTo(tieringStateJson); - - LakeTableSnapshot snapshot = - new LakeTableSnapshot(9L, offsets.getOffsets(), offsets.getTieringStateJson()); - GetLakeSnapshotResponse response = - ServerRpcMessageUtils.makeGetLakeSnapshotResponse(tableId, snapshot); - assertThat(response.getTableId()).isEqualTo(tableId); - assertThat(response.getSnapshotId()).isEqualTo(9L); - assertThat(response.hasTieringStateJson()).isTrue(); - assertThat(response.getTieringStateJson()).isEqualTo(tieringStateJson); - } - @Test void testAbsentTieringState() { // PREPARE read without tiering state -> null. 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..e7fca60b5d 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 @@ -30,6 +30,7 @@ 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 +105,22 @@ void testBackwardCompatibility() { LakeTableSnapshotLegacyJsonSerde.INSTANCE); assertThat(snapshot3.getSnapshotId()).isEqualTo(3); } + + @Test + void testSerializeRejectsTieringState() { + // Tiering state is a version-2 only feature; the legacy v1 format must reject it rather + // than + // silently drop it. + Map offsets = new HashMap<>(); + offsets.put(new TableBucket(1L, 0), 100L); + LakeTableSnapshot withState = + new LakeTableSnapshot( + 1L, offsets, "{\"version\":1}".getBytes(StandardCharsets.UTF_8)); + assertThatThrownBy( + () -> + JsonSerdeUtils.writeValueAsBytes( + withState, LakeTableSnapshotLegacyJsonSerde.INSTANCE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("does not support tiering state"); + } } From 59f7e20d8c61ca25b77f38b3886ba8699daa1b2c Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Sat, 18 Jul 2026 12:11:49 +0800 Subject: [PATCH 4/4] fix review --- .../fluss/client/metadata/LakeSnapshot.java | 14 +++++- .../client/metadata/LakeSnapshotTest.java | 24 +++++++--- .../lake/committer/LakeTieringTableState.java | 8 ++-- .../json/LakeTieringTableStateJsonSerde.java | 25 +++++----- .../LakeTieringTableStateJsonSerdeTest.java | 27 +++++------ .../server/zk/data/lake/LakeTableHelper.java | 39 ++++++++------- .../zk/data/lake/LakeTableHelperTest.java | 47 ++++++++++++++++--- 7 files changed, 120 insertions(+), 64 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java index 4c87ea8536..95f062b122 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java @@ -31,8 +31,9 @@ * A class representing the lake snapshot information of a table. It contains: *

  • The snapshot id and the log offset for each bucket. *
  • The opaque table-level tiering state (see {@link LakeTieringTableState}) for partitioned - * tables, parsed lazily via {@link #getLakeTieringTableState()}. It is {@code null} for - * non-partitioned tables or when talking to an old coordinator that does not report it. + * tables, exposed parsed via {@link #getLakeTieringTableState()} or raw via {@link + * #getRawTieringStateJson()}. It is {@code null} for non-partitioned tables or when talking to + * an old coordinator that does not report it. * * @since 0.3 */ @@ -77,6 +78,15 @@ public LakeTieringTableState getLakeTieringTableState() { : LakeTieringTableState.fromJsonBytes(lakeTieringTableStateJson); } + /** + * Returns the raw, unparsed tiering-state JSON bytes ({@code null} if absent), for passing a + * newer, unreadable state through unchanged. + */ + @Nullable + public byte[] getRawTieringStateJson() { + return lakeTieringTableStateJson; + } + @Override public String toString() { return "LakeSnapshot{" diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java index 97522b85b1..35d8b01f3a 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java @@ -34,18 +34,30 @@ class LakeSnapshotTest { @Test void testLazyParse() { - // absent -> null. - assertThat(new LakeSnapshot(1L, Collections.emptyMap()).getLakeTieringTableState()) - .isNull(); + // absent -> null (parsed and raw). + LakeSnapshot absent = new LakeSnapshot(1L, Collections.emptyMap()); + assertThat(absent.getLakeTieringTableState()).isNull(); + assertThat(absent.getRawTieringStateJson()).isNull(); - // present -> parsed lazily. + // present -> parsed lazily; raw bytes also exposed. byte[] json = new LakeTieringTableState(true, Collections.singletonMap(5L, 1000L)).toJsonBytes(); - LakeTieringTableState state = - new LakeSnapshot(1L, Collections.emptyMap(), json).getLakeTieringTableState(); + LakeSnapshot present = new LakeSnapshot(1L, Collections.emptyMap(), json); + LakeTieringTableState state = present.getLakeTieringTableState(); assertThat(state).isNotNull(); assertThat(state.isPartitionDoneInitialized()).isTrue(); assertThat(state.getPartitionUpdateTimes()).containsEntry(5L, 1000L); + assertThat(present.getRawTieringStateJson()).isEqualTo(json); + } + + @Test + void testHigherVersionExposesRawBytesForPassthrough() { + byte[] json = "{\"version\":99}".getBytes(StandardCharsets.UTF_8); + LakeSnapshot snapshot = new LakeSnapshot(1L, Collections.emptyMap(), json); + // unreadable here: parsing fails so the caller passes the raw bytes through unchanged. + assertThatThrownBy(snapshot::getLakeTieringTableState) + .isInstanceOf(IllegalArgumentException.class); + assertThat(snapshot.getRawTieringStateJson()).isEqualTo(json); } @Test diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java index 1677561707..656dc550fa 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java @@ -41,10 +41,10 @@ * * *

    This is a flat, versioned model. Evolve it by adding fields and bumping {@link - * #CURRENT_VERSION} (e.g. a future watermark mode). To stay safe when a newer and an older tiering - * job run against the same table across rounds, a consumer that reads a {@code version} higher than - * its own {@link #CURRENT_VERSION} must treat it as read-only so it never drops fields written by a - * newer build; the serializer enforces this by refusing to write such a state. + * #CURRENT_VERSION}. When a newer and an older tiering job run against the same table across + * rounds, a build that reads a {@code version} higher than its own {@link #CURRENT_VERSION} cannot + * interpret it: {@link #fromJsonBytes} raises {@link IllegalArgumentException}, and the caller must + * pass the raw bytes through unchanged so the newer build's state is never dropped. * * @since 0.9 */ diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java index 34abdaa200..7594b96b5d 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java @@ -22,7 +22,6 @@ import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; import java.io.IOException; -import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.Map; @@ -32,9 +31,9 @@ * {"version":1,"partition_done_initialized":true,"partition_update_times":{"5":1704153550000}}}. * *

    Validation is strict for known fields, tolerant for unknown ones. {@code version} is mandatory - * and positive; a corrupt known field fails fast. A higher-than-supported {@code version} is read - * as version-only (so an older build detects it and goes read-only), and serialization refuses it - * so an older build can never overwrite a state written by a newer build. + * and positive; a corrupt known field fails fast. A higher-than-supported {@code version} is + * neither read nor written; the caller keeps the raw bytes and passes them through unchanged so a + * newer build's state is never dropped. */ public class LakeTieringTableStateJsonSerde implements JsonSerializer, JsonDeserializer { @@ -48,15 +47,15 @@ public class LakeTieringTableStateJsonSerde @Override public void serialize(LakeTieringTableState state, JsonGenerator generator) throws IOException { - // Refuse to write a version this build does not understand (it would drop unparsed newer - // fields); higher-version states are read-only. + // Defense: this build never constructs a higher-version state to write (it passes the raw + // bytes through instead), so this guards against misuse of the version-taking constructor. if (state.getVersion() > LakeTieringTableState.CURRENT_VERSION) { throw new IllegalStateException( - "Refusing to serialize tiering state with unsupported version " + "Refusing to serialize unsupported tiering state version " + state.getVersion() + " > " + LakeTieringTableState.CURRENT_VERSION - + " (read-only)."); + + "."); } generator.writeStartObject(); generator.writeNumberField(VERSION_KEY, state.getVersion()); @@ -88,10 +87,14 @@ public LakeTieringTableState deserialize(JsonNode node) { } int version = versionNode.asInt(); - // Higher-than-supported version: read version-only (skip other fields) so an older build - // detects it and stays read-only. + // A newer version cannot be interpreted here; the caller must pass the raw bytes through. if (version > LakeTieringTableState.CURRENT_VERSION) { - return new LakeTieringTableState(version, false, Collections.emptyMap()); + throw new IllegalArgumentException( + "Unsupported tiering state version " + + version + + " > " + + LakeTieringTableState.CURRENT_VERSION + + "; pass the raw bytes through unchanged."); } // partition_done_initialized: optional, must be a boolean when present. diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java index 4e6539930a..3eb15e8f8c 100644 --- a/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java @@ -92,23 +92,20 @@ void testStrictValidationRejectsCorruptFields() { } @Test - void testHigherVersionIsReadableButNotWritable() { - // Reading a higher (unsupported) version and unknown fields must not raise; only the - // version - // is read (higher-version fields skipped) so an older build can detect it and degrade to - // read-only. + void testHigherVersionIsNeitherReadableNorWritable() { + // Reading a newer, unreadable version fails fast so the caller passes the raw bytes through + // instead of interpreting (and possibly dropping) it. String json = "{\"version\":99,\"partition_done_initialized\":true," + "\"partition_update_times\":{\"5\":1000},\"future_field\":\"x\"}"; - LakeTieringTableState state = - LakeTieringTableState.fromJsonBytes(json.getBytes(StandardCharsets.UTF_8)); - assertThat(state.getVersion()).isEqualTo(99); - // higher-version fields are NOT interpreted. - assertThat(state.isPartitionDoneInitialized()).isFalse(); - assertThat(state.getPartitionUpdateTimes()).isEmpty(); - // Writing it back is refused: an older build must not overwrite a newer-version state. - assertThatThrownBy(state::toJsonBytes) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("read-only"); + assertThatThrownBy( + () -> + LakeTieringTableState.fromJsonBytes( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IllegalArgumentException.class); + + // Writing a higher-version object is likewise refused (defense against misuse). + LakeTieringTableState higher = new LakeTieringTableState(99, true, new HashMap<>()); + assertThatThrownBy(higher::toJsonBytes).isInstanceOf(IllegalStateException.class); } } 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 3464e52b33..a6c69f556c 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 @@ -63,17 +63,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())); - // State is v2-only: carry it through so the legacy (v1) serde fails fast if present. - lakeTableSnapshot = - new LakeTableSnapshot( - tableId, - tableBucketOffsets.getOffsets(), - tableBucketOffsets.getTieringStateJson()); + LakeTableSnapshot previousSnapshot = + optPreviousLakeTable.get().getOrReadLatestTableSnapshot(); + // The legacy (v1) format cannot carry tiering state. This path is not expected to run + // on a state-bearing table; if it does (e.g. an old committer), drop the state with a + // warning instead of failing the commit. + if (previousSnapshot.getTieringStateJson() != null) { + LOG.warn( + "Dropping tiering state for table {} on a legacy (v1) lake commit; " + + "the v1 format cannot store it.", + tableId); + } + Map bucketLogEndOffset = + new HashMap<>(previousSnapshot.getBucketLogEndOffset()); + bucketLogEndOffset.putAll(lakeTableSnapshot.getBucketLogEndOffset()); + lakeTableSnapshot = new LakeTableSnapshot(tableId, bucketLogEndOffset); } zkClient.upsertLakeTable( tableId, new LakeTable(lakeTableSnapshot), optPreviousLakeTable.isPresent()); @@ -183,15 +187,10 @@ public TableBucketOffsets mergeTableBucketOffsets( new HashMap<>(previousSnapshot.getBucketLogEndOffset()); bucketLogEndOffset.putAll(newTableBucketOffsets.getOffsets()); - // Unlike the per-bucket offsets above (partial merge), the tiering state is opaque here and - // cannot be field-merged, so it uses whole-value snapshot (PUT) semantics: - // - present (including an empty object) overwrites the whole state; - // - absent keeps the previous state (a legacy/no-op writer never sends it, so keeping the - // previous value is required for rolling-upgrade safety). - byte[] tieringStateJson = - newTableBucketOffsets.getTieringStateJson() != null - ? newTableBucketOffsets.getTieringStateJson() - : previousSnapshot.getTieringStateJson(); + // The tiering state is opaque and cannot be field-merged; the committer always sends the + // complete state, so it is overwritten wholesale (absent means "no state"). Byte pass- + // through on the committer side keeps a newer, unreadable state from being dropped. + byte[] tieringStateJson = newTableBucketOffsets.getTieringStateJson(); return new TableBucketOffsets( newTableBucketOffsets.getTableId(), bucketLogEndOffset, tieringStateJson); 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 0d5ec3ea59..83434b38c8 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 @@ -308,9 +308,13 @@ void testMergeTieringStateOverwrite(@TempDir Path tempDir) throws Exception { assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 3L, 0), 350L); } - /** Verifies a new request without tiering state keeps the previous one on merge. */ + /** + * Verifies the opaque tiering state uses pure overwrite semantics: a new request without + * tiering state drops it (the committer sends the complete state each round), while bucket + * offsets are still merged. + */ @Test - void testMergeKeepsPreviousTieringStateWhenAbsent(@TempDir Path tempDir) throws Exception { + void testMergeDropsTieringStateWhenAbsent(@TempDir Path tempDir) throws Exception { LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString()); long tableId = 101L; TablePath tablePath = TablePath.of("test_db", "markdone_keep_test"); @@ -319,8 +323,7 @@ void testMergeKeepsPreviousTieringStateWhenAbsent(@TempDir Path tempDir) throws Map prevOffsets = new HashMap<>(); prevOffsets.put(new TableBucket(tableId, 1L, 0), 100L); byte[] prevState = - new LakeTieringTableState(true, java.util.Collections.singletonMap(1L, 1000L)) - .toJsonBytes(); + new LakeTieringTableState(true, Collections.singletonMap(1L, 1000L)).toJsonBytes(); FsPath prevPath = lakeTableHelper.storeLakeTableOffsetsFile( tablePath, new TableBucketOffsets(tableId, prevOffsets, prevState)); @@ -330,15 +333,47 @@ void testMergeKeepsPreviousTieringStateWhenAbsent(@TempDir Path tempDir) throws Map newOffsets = new HashMap<>(); newOffsets.put(new TableBucket(tableId, 1L, 0), 150L); - // new request carries no tiering state (null) + // new request carries no tiering state (null) -> dropped, offsets still merged. TableBucketOffsets merged = lakeTableHelper.mergeTableBucketOffsets( previousLakeTable, new TableBucketOffsets(tableId, newOffsets, null)); - assertThat(merged.getTieringStateJson()).isEqualTo(prevState); + assertThat(merged.getTieringStateJson()).isNull(); assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L); } + /** + * A legacy (v1) commit on a state-bearing table drops the state (with a warning), not fails. + */ + @Test + void testRegisterV1DropsExistingTieringState(@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 tiering state + Map offsets = new HashMap<>(); + offsets.put(new TableBucket(tableId, 0), 100L); + byte[] state = + new LakeTieringTableState(true, Collections.singletonMap(1L, 1000L)).toJsonBytes(); + FsPath path = + lakeTableHelper.storeLakeTableOffsetsFile( + tablePath, new TableBucketOffsets(tableId, offsets, state)); + lakeTableHelper.registerLakeTableSnapshotV2( + tableId, new LakeTable.LakeSnapshotMetadata(1L, path, path)); + + // a v1 commit drops the state 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.getTieringStateJson()).isNull(); + 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 {