You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2024/05/16 07:16:54 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-34537] Autoscaler JDBC Support HikariPool (#785)
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new eb191337 [FLINK-34537] Autoscaler JDBC Support HikariPool (#785)
eb191337 is described below
commit eb191337251410d9d40300a0050a17d5c2300d8b
Author: ConradJam <ja...@gmail.com>
AuthorDate: Thu May 16 15:16:49 2024 +0800
[FLINK-34537] Autoscaler JDBC Support HikariPool (#785)
* [FLINK-34537] Autoscaler JDBC Support HikariPool
* fix pom with dependency
* Remove unnecessary HikariCP dependency for flink-autoscaler-standalone module
* Polish the error message
* Address Yuepeng's comments
---------
Co-authored-by: Rui Fan <19...@gmail.com>
---
.../standalone/AutoscalerEventHandlerFactory.java | 14 +-----
.../standalone/AutoscalerStateStoreFactory.java | 14 +-----
.../standalone/utils/HikariJDBCUtil.java | 53 ++++++++++++++++++++++
.../src/main/resources/META-INF/NOTICE | 1 +
.../AutoscalerEventHandlerFactoryTest.java | 18 ++++----
.../AutoscalerStateStoreFactoryTest.java | 18 ++++----
flink-autoscaler/pom.xml | 12 +++++
pom.xml | 1 +
8 files changed, 89 insertions(+), 42 deletions(-)
diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactory.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactory.java
index cc69c453..5d063caf 100644
--- a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactory.java
+++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactory.java
@@ -22,20 +22,15 @@ import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
import org.apache.flink.autoscaler.event.LoggingEventHandler;
import org.apache.flink.autoscaler.jdbc.event.JdbcAutoScalerEventHandler;
import org.apache.flink.autoscaler.jdbc.event.JdbcEventInteractor;
+import org.apache.flink.autoscaler.standalone.utils.HikariJDBCUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.description.InlineElement;
-import java.sql.DriverManager;
-
import static org.apache.flink.autoscaler.standalone.AutoscalerEventHandlerFactory.EventHandlerType.JDBC;
import static org.apache.flink.autoscaler.standalone.AutoscalerEventHandlerFactory.EventHandlerType.LOGGING;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.EVENT_HANDLER_TYPE;
-import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_PASSWORD_ENV_VARIABLE;
-import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_URL;
-import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_USERNAME;
import static org.apache.flink.configuration.description.TextElement.text;
-import static org.apache.flink.util.Preconditions.checkArgument;
/** The factory of {@link AutoScalerEventHandler}. */
public class AutoscalerEventHandlerFactory {
@@ -77,12 +72,7 @@ public class AutoscalerEventHandlerFactory {
private static <KEY, Context extends JobAutoScalerContext<KEY>>
AutoScalerEventHandler<KEY, Context> createJdbcEventHandler(Configuration conf)
throws Exception {
- final var jdbcUrl = conf.get(JDBC_URL);
- checkArgument(jdbcUrl != null, "%s is required for jdbc event handler.", JDBC_URL.key());
- var user = conf.get(JDBC_USERNAME);
- var password = System.getenv().get(conf.get(JDBC_PASSWORD_ENV_VARIABLE));
-
- var conn = DriverManager.getConnection(jdbcUrl, user, password);
+ var conn = HikariJDBCUtil.getConnection(conf);
return new JdbcAutoScalerEventHandler<>(new JdbcEventInteractor(conn));
}
}
diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactory.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactory.java
index d5a62bdf..05283627 100644
--- a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactory.java
+++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactory.java
@@ -21,22 +21,17 @@ import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.autoscaler.jdbc.state.JdbcAutoScalerStateStore;
import org.apache.flink.autoscaler.jdbc.state.JdbcStateInteractor;
import org.apache.flink.autoscaler.jdbc.state.JdbcStateStore;
+import org.apache.flink.autoscaler.standalone.utils.HikariJDBCUtil;
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.description.InlineElement;
-import java.sql.DriverManager;
-
import static org.apache.flink.autoscaler.standalone.AutoscalerStateStoreFactory.StateStoreType.JDBC;
import static org.apache.flink.autoscaler.standalone.AutoscalerStateStoreFactory.StateStoreType.MEMORY;
-import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_PASSWORD_ENV_VARIABLE;
-import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_URL;
-import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_USERNAME;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.STATE_STORE_TYPE;
import static org.apache.flink.configuration.description.TextElement.text;
-import static org.apache.flink.util.Preconditions.checkArgument;
/** The factory of {@link AutoScalerStateStore}. */
public class AutoscalerStateStoreFactory {
@@ -79,12 +74,7 @@ public class AutoscalerStateStoreFactory {
private static <KEY, Context extends JobAutoScalerContext<KEY>>
AutoScalerStateStore<KEY, Context> createJdbcStateStore(Configuration conf)
throws Exception {
- final var jdbcUrl = conf.get(JDBC_URL);
- checkArgument(jdbcUrl != null, "%s is required for jdbc state store.", JDBC_URL.key());
- var user = conf.get(JDBC_USERNAME);
- var password = System.getenv().get(conf.get(JDBC_PASSWORD_ENV_VARIABLE));
-
- var conn = DriverManager.getConnection(jdbcUrl, user, password);
+ var conn = HikariJDBCUtil.getConnection(conf);
return new JdbcAutoScalerStateStore<>(new JdbcStateStore(new JdbcStateInteractor(conn)));
}
}
diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/HikariJDBCUtil.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/HikariJDBCUtil.java
new file mode 100644
index 00000000..65fb9642
--- /dev/null
+++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/HikariJDBCUtil.java
@@ -0,0 +1,53 @@
+/*
+ * 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.standalone.utils;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.StringUtils;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_PASSWORD_ENV_VARIABLE;
+import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_URL;
+import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_USERNAME;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Hikari JDBC common util. */
+public class HikariJDBCUtil {
+
+ public static final String JDBC_URL_REQUIRED_HINT =
+ String.format(
+ "%s is required when jdbc state store or jdbc event handler is used.",
+ JDBC_URL.key());
+
+ public static Connection getConnection(Configuration conf) throws SQLException {
+ final var jdbcUrl = conf.get(JDBC_URL);
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(jdbcUrl), JDBC_URL_REQUIRED_HINT);
+ var user = conf.get(JDBC_USERNAME);
+ var password = System.getenv().get(conf.get(JDBC_PASSWORD_ENV_VARIABLE));
+ HikariConfig hikariConfig = new HikariConfig();
+ hikariConfig.setJdbcUrl(jdbcUrl);
+ hikariConfig.setUsername(user);
+ hikariConfig.setPassword(password);
+ return new HikariDataSource(hikariConfig).getConnection();
+ }
+}
diff --git a/flink-autoscaler-standalone/src/main/resources/META-INF/NOTICE b/flink-autoscaler-standalone/src/main/resources/META-INF/NOTICE
index e17f3fd5..b85ac1e0 100644
--- a/flink-autoscaler-standalone/src/main/resources/META-INF/NOTICE
+++ b/flink-autoscaler-standalone/src/main/resources/META-INF/NOTICE
@@ -20,6 +20,7 @@ This project bundles the following dependencies under the Apache Software Licens
- org.apache.logging.log4j:log4j-api:2.17.1
- org.apache.logging.log4j:log4j-core:2.17.1
- org.apache.logging.log4j:log4j-1.2-api:2.17.1
+- com.zaxxer:HikariCP:5.1.0
This project bundles the following dependencies under the BSD License.
See bundled license files for details.
diff --git a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactoryTest.java b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactoryTest.java
index 4c76937b..2ab21aa1 100644
--- a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactoryTest.java
+++ b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerEventHandlerFactoryTest.java
@@ -19,17 +19,16 @@ package org.apache.flink.autoscaler.standalone;
import org.apache.flink.autoscaler.event.LoggingEventHandler;
import org.apache.flink.autoscaler.jdbc.event.JdbcAutoScalerEventHandler;
+import org.apache.flink.autoscaler.standalone.utils.HikariJDBCUtil;
import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-
import static org.apache.flink.autoscaler.standalone.AutoscalerEventHandlerFactory.EventHandlerType.JDBC;
import static org.apache.flink.autoscaler.standalone.AutoscalerEventHandlerFactory.EventHandlerType.LOGGING;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.EVENT_HANDLER_TYPE;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_URL;
+import static org.apache.flink.autoscaler.standalone.utils.HikariJDBCUtil.JDBC_URL_REQUIRED_HINT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -59,25 +58,26 @@ class AutoscalerEventHandlerFactoryTest {
conf.set(EVENT_HANDLER_TYPE, JDBC);
assertThatThrownBy(() -> AutoscalerEventHandlerFactory.create(conf))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("%s is required for jdbc event handler.", JDBC_URL.key());
+ .hasMessage(JDBC_URL_REQUIRED_HINT);
}
@Test
void testCreateJdbcEventHandler() throws Exception {
final var jdbcUrl = "jdbc:derby:memory:test";
- DriverManager.getConnection(String.format("%s;create=true", jdbcUrl)).close();
-
// Test for create JDBC Event Handler.
final var conf = new Configuration();
conf.set(EVENT_HANDLER_TYPE, JDBC);
- conf.set(JDBC_URL, jdbcUrl);
+ conf.set(JDBC_URL, String.format("%s;create=true", jdbcUrl));
+ HikariJDBCUtil.getConnection(conf).close();
var eventHandler = AutoscalerEventHandlerFactory.create(conf);
assertThat(eventHandler).isInstanceOf(JdbcAutoScalerEventHandler.class);
try {
- DriverManager.getConnection(String.format("%s;shutdown=true", jdbcUrl)).close();
- } catch (SQLException ignored) {
+ conf.set(JDBC_URL, String.format("%s;shutdown=true", jdbcUrl));
+ HikariJDBCUtil.getConnection(conf).close();
+ } catch (RuntimeException ignored) {
+ // database shutdown ignored exception
}
}
}
diff --git a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactoryTest.java b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactoryTest.java
index 0939ea06..7ef3c4ff 100644
--- a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactoryTest.java
+++ b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/AutoscalerStateStoreFactoryTest.java
@@ -18,18 +18,17 @@
package org.apache.flink.autoscaler.standalone;
import org.apache.flink.autoscaler.jdbc.state.JdbcAutoScalerStateStore;
+import org.apache.flink.autoscaler.standalone.utils.HikariJDBCUtil;
import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-
import static org.apache.flink.autoscaler.standalone.AutoscalerStateStoreFactory.StateStoreType.JDBC;
import static org.apache.flink.autoscaler.standalone.AutoscalerStateStoreFactory.StateStoreType.MEMORY;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.JDBC_URL;
import static org.apache.flink.autoscaler.standalone.config.AutoscalerStandaloneOptions.STATE_STORE_TYPE;
+import static org.apache.flink.autoscaler.standalone.utils.HikariJDBCUtil.JDBC_URL_REQUIRED_HINT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -58,25 +57,26 @@ class AutoscalerStateStoreFactoryTest {
conf.set(STATE_STORE_TYPE, JDBC);
assertThatThrownBy(() -> AutoscalerStateStoreFactory.create(conf))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("%s is required for jdbc state store.", JDBC_URL.key());
+ .hasMessage(JDBC_URL_REQUIRED_HINT);
}
@Test
void testCreateJdbcStateStore() throws Exception {
final var jdbcUrl = "jdbc:derby:memory:test";
- DriverManager.getConnection(String.format("%s;create=true", jdbcUrl)).close();
-
// Test for create JDBC State store.
final var conf = new Configuration();
conf.set(STATE_STORE_TYPE, JDBC);
- conf.set(JDBC_URL, jdbcUrl);
+ conf.set(JDBC_URL, String.format("%s;create=true", jdbcUrl));
+ HikariJDBCUtil.getConnection(conf).close();
var stateStore = AutoscalerStateStoreFactory.create(conf);
assertThat(stateStore).isInstanceOf(JdbcAutoScalerStateStore.class);
try {
- DriverManager.getConnection(String.format("%s;shutdown=true", jdbcUrl)).close();
- } catch (SQLException ignored) {
+ conf.set(JDBC_URL, String.format("%s;shutdown=true", jdbcUrl));
+ HikariJDBCUtil.getConnection(conf).close();
+ } catch (RuntimeException ignored) {
+ // database shutdown ignored exception
}
}
}
diff --git a/flink-autoscaler/pom.xml b/flink-autoscaler/pom.xml
index 14c34806..cde2353e 100644
--- a/flink-autoscaler/pom.xml
+++ b/flink-autoscaler/pom.xml
@@ -57,6 +57,18 @@ under the License.
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>${quartz.version}</version>
+ <exclusions>
+ <exclusion>
+ <artifactId>HikariCP-java7</artifactId>
+ <groupId>com.zaxxer</groupId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.zaxxer</groupId>
+ <artifactId>HikariCP</artifactId>
+ <version>${hikari.version}</version>
</dependency>
<dependency>
diff --git a/pom.xml b/pom.xml
index 0dc4d302..56e8d6bc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -99,6 +99,7 @@ under the License.
<assertj.version>3.21.0</assertj.version>
<quartz.version>2.3.2</quartz.version>
+ <hikari.version>5.1.0</hikari.version>
<flink-kubernetes-operator.surefire.baseArgLine>-XX:+IgnoreUnrecognizedVMOptions ${surefire.module.config}</flink-kubernetes-operator.surefire.baseArgLine>