[FLINK-8982][E2E Tests] E2E Test for queryable state with TM restart

Add javadoc and remove dead code

Move from precommit to nightly tests

Add parameters for state backend and tmp dir

Remove dead code and add option for num iterations

Add utility functions to queryable state

Refactor test_queryable_state

Minor fixed and improvements to common.sh

Some fixes

Add a test case where a TM gets restarted

Some code cleanup

K comments

Better naming for queryable state producer and consumer

Address some PR comments

Wait for checkpoints instead of sleep

cleanup log files

iterm

flink-end-to-end-tests/flink-queryable-state-test/pom.xml

@@ -0,0 +1,134 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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:https://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.
+ -->
+<project xmlns="http:https://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http:https://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http:https://maven.apache.org/POM/4.0.0 http:https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>1.6-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-queryable-state-test_${scala.binary.version}</artifactId>
+ <name>flink-queryable-state-test</name>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <version>2.4</version>
+
+ <executions>
+ <execution>
+ <id>QsStateProducer</id>
+ <phase>package</phase>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ <configuration>
+ <classifier>QsStateProducer</classifier>
+ <archive>
+ <manifestEntries>
+ <program-class>
+ org.apache.flink.streaming.tests.queryablestate.QsStateProducer
+ </program-class>
+ </manifestEntries>
+ </archive>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+
+ <execution>
+ <id>QsStateClient</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadeTestJar>false</shadeTestJar>
+ <shadedArtifactAttached>false</shadedArtifactAttached>
+ <createDependencyReducedPom>false</createDependencyReducedPom>
+ <transformers>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>
+ org.apache.flink.streaming.tests.queryablestate.QsStateClient
+ </mainClass>
+ </transformer>
+ </transformers>
+ <finalName>QsStateClient</finalName>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!--simplify the name of the testing JARs for referring to them in the end-to-end test scripts-->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <version>1.7</version>
+ <executions>
+ <execution>
+ <id>rename</id>
+ <phase>package</phase>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <configuration>
+ <target>
+ <copy
+ file="${project.basedir}/target/flink-queryable-state-test_${scala.binary.version}-${project.version}-QsStateProducer.jar"
+ tofile="${project.basedir}/target/QsStateProducer.jar"/>
+ </target>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>

flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/Email.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:https://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.queryablestate;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * Toy email resentation.
+ */
+public class Email {
+
+ private EmailId emailId;
+ private Instant timestamp;
+ private String foo;
+ private LabelSurrogate label;
+
+ public Email(EmailId emailId, Instant timestamp, String foo, LabelSurrogate label) {
+ this.emailId = emailId;
+ this.timestamp = timestamp;
+ this.foo = foo;
+ this.label = label;
+ }
+
+ public EmailId getEmailId() {
+ return emailId;
+ }
+
+ public void setEmailId(EmailId emailId) {
+ this.emailId = emailId;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Instant timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ public String getFoo() {
+ return foo;
+ }
+
+ public void setFoo(String foo) {
+ this.foo = foo;
+ }
+
+ public LabelSurrogate getLabel() {
+ return label;
+ }
+
+ public void setLabel(LabelSurrogate label) {
+ this.label = label;
+ }
+
+ public String getDate() {
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.of("UTC"));
+ return formatter.format(timestamp);
+ }
+}

flink-end-to-end-tests/flink-queryable-state-test/src/main/java/org/apache/flink/streaming/tests/queryablestate/EmailId.java

@@ -0,0 +1,73 @@
+/*
+ * 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:https://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.queryablestate;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * POJO representing an EmailId.
+ */
+public class EmailId implements Serializable {
+
+ private static final long serialVersionUID = -5001464312464872467L;
+
+ private String emailId;
+
+ public EmailId() {
+
+ }
+
+ public EmailId(String emailId) {
+ this.emailId = Objects.requireNonNull(emailId);
+ }
+
+ public void setEmailId(String emailId) {
+ this.emailId = emailId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ EmailId emailId1 = (EmailId) o;
+
+ return Objects.equals(emailId, emailId1.emailId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(emailId);
+ }
+
+ public String getEmailId() {
+ return emailId;
+ }
+
+ @Override
+ public String toString() {
+ return "EmailId{" +
+ "emailId='" + emailId + '\'' +
+ '}';
+ }
+}