Finish the AbstractJDBCStateInteractorITCase

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/AbstractJDBCStateInteractorITCase.java

@@ -0,0 +1,49 @@
+package org.apache.flink.autoscaler.jdbc.state;
+
+import org.apache.flink.autoscaler.jdbc.testutils.databases.DatabaseTest;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.autoscaler.jdbc.state.StateType.COLLECTED_METRICS;
+import static org.apache.flink.autoscaler.jdbc.state.StateType.SCALING_HISTORY;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The abstract IT case for {@link JDBCStateInteractor}. */
+public abstract class AbstractJDBCStateInteractorITCase implements DatabaseTest {
+
+    @Test
+    void testAllOperations() throws Exception {
+        var jobKey = "jobKey";
+        var value1 = "value1";
+        var value2 = "value2";
+        var value3 = "value3";
+        var jdbcStateInteractor = new JDBCStateInteractor(getConnection());
+        assertThat(jdbcStateInteractor.queryData(jobKey)).isEmpty();
+
+        // Test for creating data.
+        jdbcStateInteractor.createData(
+                jobKey,
+                List.of(COLLECTED_METRICS, SCALING_HISTORY),
+                Map.of(COLLECTED_METRICS, value1, SCALING_HISTORY, value2));
+        assertThat(jdbcStateInteractor.queryData(jobKey))
+                .isEqualTo(Map.of(COLLECTED_METRICS, value1, SCALING_HISTORY, value2));
+
+        // Test for updating data.
+        jdbcStateInteractor.updateData(
+                jobKey,
+                List.of(COLLECTED_METRICS),
+                Map.of(COLLECTED_METRICS, value3, SCALING_HISTORY, value2));
+        assertThat(jdbcStateInteractor.queryData(jobKey))
+                .isEqualTo(Map.of(COLLECTED_METRICS, value3, SCALING_HISTORY, value2));
+
+        // Test for deleting data.
+        jdbcStateInteractor.deleteData(jobKey, List.of(COLLECTED_METRICS));
+        assertThat(jdbcStateInteractor.queryData(jobKey))
+                .isEqualTo(Map.of(SCALING_HISTORY, value2));
+        jdbcStateInteractor.deleteData(jobKey, List.of(SCALING_HISTORY));
+        assertThat(jdbcStateInteractor.queryData(jobKey)).isEmpty();
+    }
+}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/DerbyJDBCStateInteractorITCase.java

@@ -0,0 +1,26 @@
+/*
+ * 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.autoscaler.jdbc.state.database;
+
+import org.apache.flink.autoscaler.jdbc.state.AbstractJDBCStateInteractorITCase;
+import org.apache.flink.autoscaler.jdbc.state.JDBCStateInteractor;
+import org.apache.flink.autoscaler.jdbc.testutils.databases.derby.DerbyTestBase;
+
+/** Test {@link JDBCStateInteractor} via Derby database. */
+public class DerbyJDBCStateInteractorITCase extends AbstractJDBCStateInteractorITCase
+        implements DerbyTestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/MySQL56JDBCStateInteractorITCase.java

@@ -0,0 +1,26 @@
+/*
+ * 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.autoscaler.jdbc.state.database;
+
+import org.apache.flink.autoscaler.jdbc.state.AbstractJDBCStateInteractorITCase;
+import org.apache.flink.autoscaler.jdbc.state.JDBCStateInteractor;
+import org.apache.flink.autoscaler.jdbc.testutils.databases.mysql.MySQL56TestBase;
+
+/** Test {@link JDBCStateInteractor} via MySQL 5.6.x. */
+public class MySQL56JDBCStateInteractorITCase extends AbstractJDBCStateInteractorITCase
+        implements MySQL56TestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/MySQL56JDBCStoreITCase.java

@@ -21,5 +21,5 @@
 import org.apache.flink.autoscaler.jdbc.state.JDBCStore;
 import org.apache.flink.autoscaler.jdbc.testutils.databases.mysql.MySQL56TestBase;
 
-/** Test {@link JDBCStore} via MySQL 5.6. */
+/** Test {@link JDBCStore} via MySQL 5.6.x. */
 public class MySQL56JDBCStoreITCase extends AbstractJDBCStoreITCase implements MySQL56TestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/MySQL57JDBCStateInteractorITCase.java

@@ -0,0 +1,26 @@
+/*
+ * 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.autoscaler.jdbc.state.database;
+
+import org.apache.flink.autoscaler.jdbc.state.AbstractJDBCStateInteractorITCase;
+import org.apache.flink.autoscaler.jdbc.state.JDBCStateInteractor;
+import org.apache.flink.autoscaler.jdbc.testutils.databases.mysql.MySQL57TestBase;
+
+/** Test {@link JDBCStateInteractor} via MySQL 5.7.x. */
+public class MySQL57JDBCStateInteractorITCase extends AbstractJDBCStateInteractorITCase
+        implements MySQL57TestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/MySQL57JDBCStoreITCase.java

@@ -21,5 +21,5 @@
 import org.apache.flink.autoscaler.jdbc.state.JDBCStore;
 import org.apache.flink.autoscaler.jdbc.testutils.databases.mysql.MySQL57TestBase;
 
-/** Test {@link JDBCStore} via MySQL 5.7. */
+/** Test {@link JDBCStore} via MySQL 5.7.x. */
 public class MySQL57JDBCStoreITCase extends AbstractJDBCStoreITCase implements MySQL57TestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/MySQL8JDBCStateInteractorITCase.java

@@ -0,0 +1,26 @@
+/*
+ * 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.autoscaler.jdbc.state.database;
+
+import org.apache.flink.autoscaler.jdbc.state.AbstractJDBCStateInteractorITCase;
+import org.apache.flink.autoscaler.jdbc.state.JDBCStateInteractor;
+import org.apache.flink.autoscaler.jdbc.testutils.databases.mysql.MySQL8TestBase;
+
+/** Test {@link JDBCStateInteractor} via MySQL 8.x. */
+public class MySQL8JDBCStateInteractorITCase extends AbstractJDBCStateInteractorITCase
+        implements MySQL8TestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/PostgreSQLJDBCStateInteractorITCase.java

@@ -0,0 +1,26 @@
+/*
+ * 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.autoscaler.jdbc.state.database;
+
+import org.apache.flink.autoscaler.jdbc.state.AbstractJDBCStateInteractorITCase;
+import org.apache.flink.autoscaler.jdbc.state.JDBCStateInteractor;
+import org.apache.flink.autoscaler.jdbc.testutils.databases.postgres.PostgreSQLTestBase;
+
+/** Test {@link JDBCStateInteractor} via Postgre SQL. */
+public class PostgreSQLJDBCStateInteractorITCase extends AbstractJDBCStateInteractorITCase
+        implements PostgreSQLTestBase {}

flink-autoscaler-plugin-jdbc/src/test/java/org/apache/flink/autoscaler/jdbc/state/database/PostgreSQLJDBCStoreITCase.java

@@ -18,8 +18,9 @@
 package org.apache.flink.autoscaler.jdbc.state.database;
 
 import org.apache.flink.autoscaler.jdbc.state.AbstractJDBCStoreITCase;
+import org.apache.flink.autoscaler.jdbc.state.JDBCStore;
 import org.apache.flink.autoscaler.jdbc.testutils.databases.postgres.PostgreSQLTestBase;
 
-/** Test for PostgreSQL 15.1. */
+/** Test {@link JDBCStore} via Postgre SQL. */
 public class PostgreSQLJDBCStoreITCase extends AbstractJDBCStoreITCase
         implements PostgreSQLTestBase {}