This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit c3e2f30c6a7fa6110027adc21ad62c414021912a Author: Aleksandr Savonin <[email protected]> AuthorDate: Mon Sep 14 23:16:07 2026 +0200 [FLINK-40584][state/rocksdb] Check MapState iterator validity before advancing Cache refill can seek past the last entry after MapState.remove() deletes the resume key. That removal does not update the cached deletion flag, so advancing the exhausted iterator violates the RocksDB API contract. --- .../flink/state/rocksdb/RocksDBMapState.java | 14 ++- .../flink/state/rocksdb/RocksIteratorWrapper.java | 1 + .../flink/state/rocksdb/RocksDBMapStateTest.java | 103 +++++++++++++++++++++ 3 files changed, 114 insertions(+), 4 deletions(-) diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java index c811fc8caac..6282000afad 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java @@ -18,6 +18,7 @@ package org.apache.flink.state.rocksdb; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.state.MapState; import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; @@ -70,6 +71,9 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U private static final Logger LOG = LoggerFactory.getLogger(RocksDBMapState.class); + /** Maximum number of entries cached by a map state iterator. */ + @VisibleForTesting static final int ITERATOR_CACHE_SIZE = 128; + /** Serializer for the keys and values. */ private TypeSerializer<UK> userKeySerializer; @@ -562,8 +566,6 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U /** An auxiliary utility to scan all entries under the given key. */ private abstract class RocksDBMapIterator<T> implements Iterator<T> { - private static final int CACHE_SIZE_LIMIT = 128; - /** The db where data resides. */ private final RocksDB db; @@ -675,8 +677,12 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U /* * If the entry pointing to the current position is not removed, it will be the first entry in the * new iterating. Skip it to avoid redundant access in such cases. + * + * Removing the current entry through MapState does not update its cached 'deleted' flag. + * A resumed seek can therefore return an invalid iterator even if 'deleted' is false. + * RocksDB requires a valid iterator before calling next(). */ - if (currentEntry != null && !currentEntry.deleted) { + if (currentEntry != null && !currentEntry.deleted && iterator.isValid()) { iterator.next(); } @@ -687,7 +693,7 @@ class RocksDBMapState<K, N, UK, UV> extends AbstractRocksDBState<K, N, Map<UK, U break; } - if (cacheEntries.size() >= CACHE_SIZE_LIMIT) { + if (cacheEntries.size() >= ITERATOR_CACHE_SIZE) { break; } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksIteratorWrapper.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksIteratorWrapper.java index 3422bf7ca35..c03a82ef50b 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksIteratorWrapper.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksIteratorWrapper.java @@ -94,6 +94,7 @@ public class RocksIteratorWrapper implements RocksIteratorInterface, Closeable { @Override public void next() { + assert isValid() : "Iterator must be valid before calling next()"; iterator.next(); } diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMapStateTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMapStateTest.java new file mode 100644 index 00000000000..fb62664af42 --- /dev/null +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBMapStateTest.java @@ -0,0 +1,103 @@ +/* + * 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.flink.state.rocksdb; + +import org.apache.flink.api.common.state.MapState; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.api.common.typeutils.base.IntSerializer; +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.util.IOUtils; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.Collections; +import java.util.Iterator; +import java.util.Map; + +import static org.apache.flink.state.rocksdb.RocksDBMapState.ITERATOR_CACHE_SIZE; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests iteration of RocksDB map state. */ +class RocksDBMapStateTest { + + private static final MapStateDescriptor<Integer, Integer> STATE_DESCRIPTOR = + new MapStateDescriptor<>("map", IntSerializer.INSTANCE, IntSerializer.INSTANCE); + + @TempDir private Path temporaryDirectory; + + private RocksDBKeyedStateBackend<Integer> backend; + + @BeforeEach + void setUp() throws Exception { + backend = + RocksDBTestUtils.builderForTestDefaults( + temporaryDirectory.toFile(), + IntSerializer.INSTANCE, + 1, + new KeyGroupRange(0, 0), + Collections.emptyList()) + .build(); + } + + @AfterEach + void tearDown() throws Exception { + if (backend != null) { + IOUtils.closeQuietly(backend); + backend.dispose(); + } + } + + /** Isolates the validity guard without requiring prefix bounds or neighboring tombstones. */ + @Test + void testIteratorCacheReloadDoesNotAdvanceAfterExhaustedSeek() throws Exception { + assertRocksIteratorAssertionsEnabled(); + final MapState<Integer, Integer> state = mapState(0, 0); + + // One entry beyond the cache ensures that exhaustion requires another seek. + for (int i = 0; i <= ITERATOR_CACHE_SIZE; i++) { + state.put(i, i); + } + final Iterator<Map.Entry<Integer, Integer>> iterator = state.iterator(); + for (int i = 0; i < ITERATOR_CACHE_SIZE; i++) { + assertThat(iterator.next().getKey()).isEqualTo(i); + } + + // MapState.remove() leaves the cached entry's 'deleted' flag false. Remove both the + // resume entry and the only later entry so the refill seek is exhausted. + state.remove(ITERATOR_CACHE_SIZE - 1); + state.remove(ITERATOR_CACHE_SIZE); + + assertThat(iterator).isExhausted(); + } + + private static void assertRocksIteratorAssertionsEnabled() { + assertThat(RocksIteratorWrapper.class.desiredAssertionStatus()) + .as("This test requires Java assertions for RocksIteratorWrapper") + .isTrue(); + } + + private MapState<Integer, Integer> mapState(int key, int namespace) throws Exception { + backend.setCurrentKey(key); + return backend.getPartitionedState(namespace, IntSerializer.INSTANCE, STATE_DESCRIPTOR); + } +}
