You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by azagrebin <gi...@git.apache.org> on 2018/07/18 08:37:48 UTC
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
GitHub user azagrebin opened a pull request:
https://github.com/apache/flink/pull/6361
[FLINK-9858] [tests] State TTL End-to-End Test
## What is the purpose of the change
This PR introduces end to end test for state TTL feature, heap and rocksdb backends.
## Brief change log
- add state ttl e2e test module to e2e tests
- add script to run it: **test_stream_state_ttl.sh**
- add heap and rocksdb variants of it to **run-nightly-tests.sh**
## Verifying this change
run **run-nightly-tests.sh** (can be only **test_stream_state_ttl.sh** lines)
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/azagrebin/flink FLINK-9858
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6361.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6361
----
commit d59c266ee15d1aaf086461f91b6f4f23a5d784fb
Author: Andrey Zagrebin <az...@...>
Date: 2018-07-13T17:27:35Z
[FLINK-9858] [tests] State TTL End-to-End Test
----
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203377098
--- Diff: flink-end-to-end-tests/test-scripts/common.sh ---
@@ -240,6 +240,15 @@ function start_cluster {
done
}
+function start_taskmanagers {
--- End diff --
I think you could just reuse function `tm_watchdog` for this purpose.
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203363708
--- Diff: flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/DataStreamStateTTLTestProgram.java ---
@@ -0,0 +1,77 @@
+/*
+ * 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.streaming.tests;
+
+import org.apache.flink.api.common.state.StateTtlConfiguration;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
+import org.apache.flink.streaming.tests.verify.TtlUpdateContext;
+
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.SEQUENCE_GENERATOR_SRC_KEYSPACE;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.SEQUENCE_GENERATOR_SRC_SLEEP_AFTER_ELEMENTS;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.SEQUENCE_GENERATOR_SRC_SLEEP_TIME;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.setupEnvironment;
+
+/**
+ * A test job for State TTL feature.
+ */
+public class DataStreamStateTTLTestProgram {
+ private static final ConfigOption<Long> STATE_TTL_VERIFIER_TTL_MILLI = ConfigOptions
--- End diff --
(also for the ones we take from the allround job.)
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/6361
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203355448
--- Diff: flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/AbstractTtlStateVerifier.java ---
@@ -0,0 +1,132 @@
+/*
+ * 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.streaming.tests.verify;
+
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfiguration;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+
+import javax.annotation.Nonnull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+abstract class AbstractTtlStateVerifier<D extends StateDescriptor<S, SV>, S extends State, SV, UV, GV>
+ implements TtlStateVerifier<UV, GV> {
+ static final Random RANDOM = new Random();
+
+ @Nonnull
+ final D stateDesc;
+
+ AbstractTtlStateVerifier(@Nonnull D stateDesc) {
+ this.stateDesc = stateDesc;
+ }
+
+ @Nonnull
+ static String randomString() {
+ StringBuilder sb = new StringBuilder();
+ IntStream.range(0, RANDOM.nextInt(14) + 2).forEach(i -> sb.append(randomChar()));
+ return sb.toString();
+ }
+
+ private static char randomChar() {
+ char d = (char) ('0' + RANDOM.nextInt(9));
+ char l = (char) ('a' + RANDOM.nextInt(25));
+ return RANDOM.nextBoolean() ? d : l;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ @Nonnull
+ public State createState(@Nonnull FunctionInitializationContext context, @Nonnull StateTtlConfiguration ttlConfig) {
+ stateDesc.enableTimeToLive(ttlConfig);
+ return createState(context);
+ }
+
+ abstract State createState(FunctionInitializationContext context);
+
+ @SuppressWarnings("unchecked")
+ @Override
+ @Nonnull
+ public TypeSerializer<UV> getUpdateSerializer() {
+ return (TypeSerializer<UV>) stateDesc.getSerializer();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public GV get(@Nonnull State state) throws Exception {
+ return getInternal((S) state);
+ }
+
+ abstract GV getInternal(@Nonnull S state) throws Exception;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void update(@Nonnull State state, Object update) throws Exception {
+ updateInternal((S) state, (UV) update);
+ }
+
+ abstract void updateInternal(@Nonnull S state, UV update) throws Exception;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean verify(@Nonnull TtlVerificationContext<?, ?> verificationContextRaw, @Nonnull Time precision) {
+ TtlVerificationContext<UV, GV> verificationContext = (TtlVerificationContext<UV, GV>) verificationContextRaw;
+ if (!isWithinPrecision(verificationContext, precision)) {
+ return true;
+ }
+ List<TtlValue<UV>> updates = new ArrayList<>(verificationContext.getPrevUpdates());
+ long currentTimestamp = verificationContext.getUpdateContext().getTimestamp();
+ GV prevValue = expected(updates, currentTimestamp);
+ GV valueBeforeUpdate = verificationContext.getUpdateContext().getValueBeforeUpdate();
+ TtlValue<UV> update = verificationContext.getUpdateContext().getUpdateWithTs();
+ GV updatedValue = verificationContext.getUpdateContext().getUpdatedValue();
+ updates.add(update);
+ GV expectedValue = expected(updates, currentTimestamp);
+ return valuesEqual(valueBeforeUpdate, prevValue) && valuesEqual(updatedValue, expectedValue);
+ }
+
+ private boolean isWithinPrecision(TtlVerificationContext<UV, GV> verificationContext, Time precision) {
+ List<TtlValue<UV>> prevUpdates = verificationContext.getPrevUpdates();
+ if (prevUpdates.isEmpty()) {
+ return true;
+ }
+ long ts = verificationContext.getUpdateContext().getTimestamp();
+ long ttl = stateDesc.getTtlConfig().getTtl().toMilliseconds();
+ return prevUpdates.stream().allMatch(u -> {
+ long delta = ts - u.getUpdateTimestamp() - ttl;
+ return precision.toMilliseconds() < delta;
+ });
+ }
+
+ private boolean valuesEqual(GV v1, GV v2) {
+ return (v1 == null && v2 == null) || (v1 != null && v1.equals(v2));
--- End diff --
return `Objects.equals(v1, v2)`?
---
[GitHub] flink issue #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/6361
Very good work, LGTM 👍 Merging.
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203361200
--- Diff: flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/AbstractTtlStateVerifier.java ---
@@ -0,0 +1,132 @@
+/*
+ * 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.streaming.tests.verify;
+
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfiguration;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+
+import javax.annotation.Nonnull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+abstract class AbstractTtlStateVerifier<D extends StateDescriptor<S, SV>, S extends State, SV, UV, GV>
+ implements TtlStateVerifier<UV, GV> {
+ static final Random RANDOM = new Random();
+
+ @Nonnull
+ final D stateDesc;
+
+ AbstractTtlStateVerifier(@Nonnull D stateDesc) {
+ this.stateDesc = stateDesc;
+ }
+
+ @Nonnull
+ static String randomString() {
+ StringBuilder sb = new StringBuilder();
+ IntStream.range(0, RANDOM.nextInt(14) + 2).forEach(i -> sb.append(randomChar()));
+ return sb.toString();
+ }
+
+ private static char randomChar() {
+ char d = (char) ('0' + RANDOM.nextInt(9));
+ char l = (char) ('a' + RANDOM.nextInt(25));
+ return RANDOM.nextBoolean() ? d : l;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ @Nonnull
+ public State createState(@Nonnull FunctionInitializationContext context, @Nonnull StateTtlConfiguration ttlConfig) {
+ stateDesc.enableTimeToLive(ttlConfig);
+ return createState(context);
+ }
+
+ abstract State createState(FunctionInitializationContext context);
+
+ @SuppressWarnings("unchecked")
+ @Override
+ @Nonnull
+ public TypeSerializer<UV> getUpdateSerializer() {
+ return (TypeSerializer<UV>) stateDesc.getSerializer();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public GV get(@Nonnull State state) throws Exception {
+ return getInternal((S) state);
+ }
+
+ abstract GV getInternal(@Nonnull S state) throws Exception;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void update(@Nonnull State state, Object update) throws Exception {
+ updateInternal((S) state, (UV) update);
+ }
+
+ abstract void updateInternal(@Nonnull S state, UV update) throws Exception;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public boolean verify(@Nonnull TtlVerificationContext<?, ?> verificationContextRaw, @Nonnull Time precision) {
--- End diff --
I think you can get rid of handling wildcards `<?, ?>` and casting by introducing the following method in `TtlVerifyFunction`:
```
private <A, B> void verify(TtlStateVerifier<A, B> verifier, TtlVerificationContext<A, B> verificationContext) {
verifier.verify(verificationContext, precision);
}
```
and call it
```
verify(TtlStateVerifier.VERIFIERS_BY_NAME.get(value.getVerifierId()), new TtlVerificationContext<>(prevValues, value));
```
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203372004
--- Diff: flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/TtlUpdateContext.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.flink.streaming.tests.verify;
+
+import javax.annotation.Nonnull;
+
+import java.io.Serializable;
+
+/** Contains relevant for state TTL update data. */
+public class TtlUpdateContext<UV, GV> implements Serializable {
+ private final int key;
+
+ @Nonnull
+ private final String verifierId;
+
+ private final GV valueBeforeUpdate;
+ private final UV update;
+ private final GV updatedValue;
+
+ private final long timestamp;
+
+ public TtlUpdateContext(int key, @Nonnull String verifierId, GV valueBeforeUpdate, UV update, GV updatedValue) {
+ this(key, verifierId, valueBeforeUpdate, update, updatedValue, System.currentTimeMillis());
--- End diff --
This is recording the timestamp with some imprecision from the actual time that the TTL state saw. On the test server, this time difference can be significant and lead to flaky tests, even with your hardcoded imprecision. Instead, we could record the timestamp `t1` before accessing the TTL, and timestamp `t2` after accessing the TTL. All state with ttl < `t1` must clearly be expired, all with ttl > `t2` must clearly be there, and cases that fall between `t1` and `t2` could be ignored because we cannot clearly decide if they saw a timestamp before or after their expiration when accessed.
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203353112
--- Diff: flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/AbstractTtlStateVerifier.java ---
@@ -0,0 +1,132 @@
+/*
+ * 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.streaming.tests.verify;
+
+import org.apache.flink.api.common.state.State;
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfiguration;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+
+import javax.annotation.Nonnull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+abstract class AbstractTtlStateVerifier<D extends StateDescriptor<S, SV>, S extends State, SV, UV, GV>
+ implements TtlStateVerifier<UV, GV> {
+ static final Random RANDOM = new Random();
+
+ @Nonnull
+ final D stateDesc;
+
+ AbstractTtlStateVerifier(@Nonnull D stateDesc) {
+ this.stateDesc = stateDesc;
+ }
+
+ @Nonnull
+ static String randomString() {
--- End diff --
You could use the existing `StringUtil.getRandomString(...)` instead of your own random string generator.
---
[GitHub] flink pull request #6361: [FLINK-9858] [tests] State TTL End-to-End Test
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/6361#discussion_r203362731
--- Diff: flink-end-to-end-tests/flink-stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/DataStreamStateTTLTestProgram.java ---
@@ -0,0 +1,77 @@
+/*
+ * 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.streaming.tests;
+
+import org.apache.flink.api.common.state.StateTtlConfiguration;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
+import org.apache.flink.streaming.tests.verify.TtlUpdateContext;
+
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.SEQUENCE_GENERATOR_SRC_KEYSPACE;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.SEQUENCE_GENERATOR_SRC_SLEEP_AFTER_ELEMENTS;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.SEQUENCE_GENERATOR_SRC_SLEEP_TIME;
+import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.setupEnvironment;
+
+/**
+ * A test job for State TTL feature.
+ */
+public class DataStreamStateTTLTestProgram {
+ private static final ConfigOption<Long> STATE_TTL_VERIFIER_TTL_MILLI = ConfigOptions
--- End diff --
Please add documentation for the config parameters to the class-level javadoc
---