This is an automated email from the ASF dual-hosted git repository. hangxiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7699a0d25b4ec06e260dd0d78d814f1a3b33f574 Author: Hangxiang Yu <[email protected]> AuthorDate: Wed Apr 17 10:42:30 2024 +0800 [FLINK-34987][state] Introduce new StateDescriptor for Async State API --- .../flink/runtime/state/v2/StateDescriptor.java | 155 +++++++++++++++++++++ .../runtime/state/v2/StateDescriptorTest.java | 134 ++++++++++++++++++ 2 files changed, 289 insertions(+) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/StateDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/StateDescriptor.java new file mode 100644 index 00000000000..1ddc92c034d --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/v2/StateDescriptor.java @@ -0,0 +1,155 @@ +/* + * 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.runtime.state.v2; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.api.common.state.StateTtlConfig; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; + +import javax.annotation.Nonnull; + +import java.io.Serializable; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Base class for state descriptors. A {@code StateDescriptor} is used for creating partitioned + * State in stateful operations internally. + * + * @param <T> The type of the value of the state object described by this state descriptor. + */ +@Internal +public abstract class StateDescriptor<T> implements Serializable { + + private static final long serialVersionUID = 1L; + + /** An enumeration of the types of supported states. */ + public enum Type { + VALUE, + LIST, + REDUCING, + FOLDING, + AGGREGATING, + MAP + } + + /** ID that uniquely identifies state created from this StateDescriptor. */ + @Nonnull private final String stateId; + + /** The serializer for the type. */ + @Nonnull private final TypeSerializer<T> typeSerializer; + + /** The configuration of state time-to-live(TTL), it is disabled by default. */ + @Nonnull private StateTtlConfig ttlConfig = StateTtlConfig.DISABLED; + + // ------------------------------------------------------------------------ + + /** + * Create a new {@code StateDescriptor} with the given stateId and the given type information. + * + * @param stateId The stateId of the {@code StateDescriptor}. + * @param typeInfo The type information for the values in the state. + */ + protected StateDescriptor(@Nonnull String stateId, TypeInformation<T> typeInfo) { + this(stateId, typeInfo, new SerializerConfigImpl()); + } + + /** + * Create a new {@code StateDescriptor} with the given stateId and the given type information. + * + * @param stateId The stateId of the {@code StateDescriptor}. + * @param typeInfo The type information for the values in the state. + * @param serializerConfig The serializer related config used to generate {@code + * TypeSerializer}. + */ + protected StateDescriptor( + @Nonnull String stateId, + @Nonnull TypeInformation<T> typeInfo, + SerializerConfig serializerConfig) { + this.stateId = checkNotNull(stateId, "stateId must not be null"); + checkNotNull(typeInfo, "type information must not be null"); + this.typeSerializer = typeInfo.createSerializer(serializerConfig); + } + + // ------------------------------------------------------------------------ + + /** + * Configures optional activation of state time-to-live (TTL). + * + * <p>State user value will expire, become unavailable and be cleaned up in storage depending on + * configured {@link StateTtlConfig}. + * + * @param ttlConfig configuration of state TTL + */ + public void enableTimeToLive(StateTtlConfig ttlConfig) { + this.ttlConfig = checkNotNull(ttlConfig); + } + + @Nonnull + public StateTtlConfig getTtlConfig() { + return ttlConfig; + } + + @Nonnull + public String getStateId() { + return stateId; + } + + @Nonnull + public TypeSerializer<T> getSerializer() { + return typeSerializer.duplicate(); + } + + // ------------------------------------------------------------------------ + + @Override + public final int hashCode() { + return stateId.hashCode() + 31 * getClass().hashCode(); + } + + @Override + public final boolean equals(Object o) { + if (o == this) { + return true; + } else if (o != null && o.getClass() == this.getClass()) { + final StateDescriptor<?> that = (StateDescriptor<?>) o; + return this.stateId.equals(that.stateId); + } else { + return false; + } + } + + @Override + public String toString() { + return getClass().getSimpleName() + + "{stateId=" + + stateId + + ", typeSerializer=" + + typeSerializer + + ", ttlConfig=" + + ttlConfig + + '}'; + } + + /** Return the specific {@code Type} of described state. */ + public abstract Type getType(); +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/StateDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/StateDescriptorTest.java new file mode 100644 index 00000000000..75a69009223 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/v2/StateDescriptorTest.java @@ -0,0 +1,134 @@ +/* + * 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.runtime.state.v2; + +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.api.common.state.StateTtlConfig; +import org.apache.flink.api.common.typeinfo.BasicTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.typeutils.GenericTypeInfo; +import org.apache.flink.core.testutils.CommonTestUtils; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for the common/shared functionality of {@link StateDescriptor}. */ +class StateDescriptorTest { + + /** + * FLINK-6775, tests that the returned serializer is duplicated. This allows to share the state + * descriptor across threads. + */ + @Test + void testSerializerDuplication() { + // we need a serializer that actually duplicates for testing (a stateful one) + // we use Kryo here, because it meets these conditions + TestStateDescriptor<String> descr = + new TestStateDescriptor<>("foobar", new GenericTypeInfo<>(String.class)); + + TypeSerializer<String> serializerA = descr.getSerializer(); + TypeSerializer<String> serializerB = descr.getSerializer(); + + // check that the retrieved serializers are not the same + assertThat(serializerB).isNotSameAs(serializerA); + } + + @Test + void testHashCodeAndEquals() throws Exception { + final String name = "testName"; + + TestStateDescriptor<String> original = + new TestStateDescriptor<>(name, BasicTypeInfo.STRING_TYPE_INFO); + TestStateDescriptor<String> same = + new TestStateDescriptor<>(name, BasicTypeInfo.STRING_TYPE_INFO); + TestStateDescriptor<String> sameBySerializer = + new TestStateDescriptor<>(name, BasicTypeInfo.STRING_TYPE_INFO); + + // test that hashCode() works on state descriptors with initialized and uninitialized + // serializers + assertThat(same).hasSameHashCodeAs(original); + assertThat(sameBySerializer).hasSameHashCodeAs(original); + + assertThat(same).isEqualTo(original); + assertThat(sameBySerializer).isEqualTo(original); + + // equality with a clone + TestStateDescriptor<String> clone = CommonTestUtils.createCopySerializable(original); + assertThat(clone).isEqualTo(original); + } + + @Test + void testEqualsSameNameAndTypeDifferentClass() { + final String name = "test name"; + + final TestStateDescriptor<String> descr1 = + new TestStateDescriptor<>(name, BasicTypeInfo.STRING_TYPE_INFO); + final OtherTestStateDescriptor<String> descr2 = + new OtherTestStateDescriptor<>(name, BasicTypeInfo.STRING_TYPE_INFO); + + assertThat(descr2).isNotEqualTo(descr1); + } + + @Test + void testStateTTlConfig() { + TestStateDescriptor<Integer> stateDescriptor = + new TestStateDescriptor<>("test-state", BasicTypeInfo.INT_TYPE_INFO); + stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Duration.ofMinutes(60)).build()); + assertThat(stateDescriptor.getTtlConfig().isEnabled()).isTrue(); + + stateDescriptor.enableTimeToLive(StateTtlConfig.DISABLED); + assertThat(stateDescriptor.getTtlConfig().isEnabled()).isFalse(); + } + + // ------------------------------------------------------------------------ + // Mock implementations and test types + // ------------------------------------------------------------------------ + + private static class TestStateDescriptor<T> extends StateDescriptor<T> { + + private static final long serialVersionUID = 1L; + + TestStateDescriptor(String name, TypeInformation<T> typeInfo) { + super(name, typeInfo, new SerializerConfigImpl()); + } + + @Override + public Type getType() { + return Type.VALUE; + } + } + + private static class OtherTestStateDescriptor<T> extends StateDescriptor<T> { + + private static final long serialVersionUID = 1L; + + OtherTestStateDescriptor(String name, TypeInformation<T> typeInfo) { + super(name, typeInfo, new SerializerConfigImpl()); + } + + @Override + public Type getType() { + return Type.VALUE; + } + } +}
