From 55ac55cc30ab5a8baf371438fbf23b3146c2a2e9 Mon Sep 17 00:00:00 2001 From: och5351 Date: Sun, 19 Jul 2026 21:09:10 +0900 Subject: [PATCH 1/2] [Feature][Common] Migrate conf package from Scala to Java --- .gitignore | 1 + .../streampark/common/conf/CommonConfig.java | 175 +++++++++++++ .../streampark/common/conf/ConfigKeys.java | 230 +++++++++++++++++ .../streampark/common/conf/ConfigOption.java | 145 +++++++++++ .../streampark/common/conf/FlinkVersion.java | 244 ++++++++++++++++++ .../common/conf/InternalConfigHolder.java | 142 ++++++++++ .../common/conf/InternalOption.java | 80 ++++++ .../common/conf/InternalOptionSpec.java} | 38 +-- .../common/conf/K8sFlinkConfig.java | 117 +++++++++ .../streampark/common/conf/SparkVersion.java | 167 ++++++++++++ .../streampark/common/conf/Workspace.java | 209 +++++++++++++++ .../streampark/common/util/TypeCastUtils.java | 45 ++++ .../streampark/common/conf/CommonConfig.scala | 117 --------- .../streampark/common/conf/ConfigKeys.scala | 204 --------------- .../streampark/common/conf/ConfigOption.scala | 82 ------ .../streampark/common/conf/FlinkVersion.scala | 161 ------------ .../common/conf/InternalConfigHolder.scala | 176 ------------- .../common/conf/K8sFlinkConfig.scala | 81 ------ .../streampark/common/conf/SparkVersion.scala | 113 -------- .../streampark/common/conf/Workspace.scala | 123 --------- .../console/core/runner/EnvInitializer.java | 2 +- 21 files changed, 1577 insertions(+), 1075 deletions(-) create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/CommonConfig.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigKeys.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigOption.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/FlinkVersion.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/InternalConfigHolder.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOption.java rename streampark-common/src/main/{scala/org/apache/streampark/common/conf/InternalOption.scala => java/org/apache/streampark/common/conf/InternalOptionSpec.java} (59%) create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/K8sFlinkConfig.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/SparkVersion.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/conf/Workspace.java create mode 100644 streampark-common/src/main/java/org/apache/streampark/common/util/TypeCastUtils.java delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/CommonConfig.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigKeys.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigOption.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalConfigHolder.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/K8sFlinkConfig.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/SparkVersion.scala delete mode 100644 streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala diff --git a/.gitignore b/.gitignore index a460e3033a..a05973bc70 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,4 @@ hs_err_pid* **/node dist/ dist-license-check/ +.flattened-pom.xml diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/CommonConfig.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/CommonConfig.java new file mode 100644 index 0000000000..ccbb297a6b --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/CommonConfig.java @@ -0,0 +1,175 @@ +/* + * 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.streampark.common.conf; + +public final class CommonConfig { + + private CommonConfig() { + } + + private static final InternalOption STREAMPARK_WORKSPACE_LOCAL = + new InternalOption<>("streampark.workspace.local", "/streampark", String.class); + + public static InternalOption STREAMPARK_WORKSPACE_LOCAL() { + return STREAMPARK_WORKSPACE_LOCAL; + } + + private static final InternalOption STREAMPARK_WORKSPACE_REMOTE = + new InternalOption<>("streampark.workspace.remote", "/streampark", String.class); + + public static InternalOption STREAMPARK_WORKSPACE_REMOTE() { + return STREAMPARK_WORKSPACE_REMOTE; + } + + private static final InternalOption STREAMPARK_HADOOP_USER_NAME = + new InternalOption<>("streampark.hadoop-user-name", "hdfs", String.class); + + public static InternalOption STREAMPARK_HADOOP_USER_NAME() { + return STREAMPARK_HADOOP_USER_NAME; + } + + private static final InternalOption STREAMPARK_PROXY_YARN_URL = + new InternalOption<>( + "streampark.proxy.yarn-url", "", String.class, "proxy yarn url. ex: knox proxy or other"); + + public static InternalOption STREAMPARK_PROXY_YARN_URL() { + return STREAMPARK_PROXY_YARN_URL; + } + + private static final InternalOption STREAMPARK_YARN_AUTH = + new InternalOption<>( + "streampark.yarn.http-auth", + "", + String.class, + "yarn http auth type. ex: simple, kerberos"); + + public static InternalOption STREAMPARK_YARN_AUTH() { + return STREAMPARK_YARN_AUTH; + } + + private static final InternalOption DOCKER_HOST = + new InternalOption<>( + "streampark.docker.http-client.docker-host", + "", + String.class, + "docker host for DockerHttpClient"); + + public static InternalOption DOCKER_HOST() { + return DOCKER_HOST; + } + + private static final InternalOption DOCKER_MAX_CONNECTIONS = + new InternalOption<>( + "streampark.docker.http-client.max-connections", + 100, + Integer.class, + "instantiating max connections for DockerHttpClient"); + + public static InternalOption DOCKER_MAX_CONNECTIONS() { + return DOCKER_MAX_CONNECTIONS; + } + + private static final InternalOption DOCKER_CONNECTION_TIMEOUT_SEC = + new InternalOption<>( + "streampark.docker.http-client.connection-timeout-sec", + 100L, + Long.class, + "instantiating connection timeout for DockerHttpClient"); + + public static InternalOption DOCKER_CONNECTION_TIMEOUT_SEC() { + return DOCKER_CONNECTION_TIMEOUT_SEC; + } + + private static final InternalOption DOCKER_RESPONSE_TIMEOUT_SEC = + new InternalOption<>( + "streampark.docker.http-client.response-timeout-sec", + 120L, + Long.class, + "instantiating connection timeout for DockerHttpClient"); + + public static InternalOption DOCKER_RESPONSE_TIMEOUT_SEC() { + return DOCKER_RESPONSE_TIMEOUT_SEC; + } + + private static final InternalOption MAVEN_SETTINGS_PATH = + new InternalOption<>( + "streampark.maven.settings", null, String.class, "maven settings.xml full path"); + + public static InternalOption MAVEN_SETTINGS_PATH() { + return MAVEN_SETTINGS_PATH; + } + + private static final InternalOption MAVEN_REMOTE_URL = + new InternalOption<>( + "streampark.maven.central.repository", + "https://repo1.maven.org/maven2/", + String.class, + "maven repository used for built-in compilation"); + + public static InternalOption MAVEN_REMOTE_URL() { + return MAVEN_REMOTE_URL; + } + + private static final InternalOption MAVEN_AUTH_USER = + new InternalOption<>( + "streampark.maven.auth.user", + null, + String.class, + "maven repository used for built-in compilation"); + + public static InternalOption MAVEN_AUTH_USER() { + return MAVEN_AUTH_USER; + } + + private static final InternalOption MAVEN_AUTH_PASSWORD = + new InternalOption<>( + "streampark.maven.auth.password", + null, + String.class, + "maven repository used for built-in compilation"); + + public static InternalOption MAVEN_AUTH_PASSWORD() { + return MAVEN_AUTH_PASSWORD; + } + + private static final InternalOption KERBEROS_TTL = + new InternalOption<>("security.kerberos.ttl", "2h", String.class, "kerberos default ttl"); + + public static InternalOption KERBEROS_TTL() { + return KERBEROS_TTL; + } + + private static final InternalOption READ_LOG_MAX_SIZE = + new InternalOption<>( + "streampark.read-log.max-size", + "1mb", + String.class, + "The maximum size of the default read log"); + + public static InternalOption READ_LOG_MAX_SIZE() { + return READ_LOG_MAX_SIZE; + } + + private static final InternalOption SPRING_PROFILES_ACTIVE = + new InternalOption<>( + "spring.profiles.active", "h2", String.class, "Use the database type"); + + public static InternalOption SPRING_PROFILES_ACTIVE() { + return SPRING_PROFILES_ACTIVE; + } +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigKeys.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigKeys.java new file mode 100644 index 0000000000..9d4919eea9 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigKeys.java @@ -0,0 +1,230 @@ +/* + * 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.streampark.common.conf; + +public final class ConfigKeys { + + private ConfigKeys() { + } + + public static final String PARAM_PREFIX = "--"; + + public static final String KEY_APP_HOME = "app.home"; + + public static final String KEY_HOST = "host"; + + public static final String KEY_PORT = "port"; + + public static final String KEY_DB = "db"; + + public static final String KEY_USER = "user"; + + public static final String KEY_PASSWORD = "password"; + + public static final String KEY_TIMEOUT = "timeout"; + + public static final String KEY_SEMANTIC = "semantic"; + + public static final String KEY_KERBEROS = "kerberos"; + + public static final String KEY_KERBEROS_SERVICE_ACCOUNT = "kubernetes.service-account"; + + public static final String KEY_HADOOP_USER_NAME = "HADOOP_USER_NAME"; + + public static final String KEY_HADOOP_SECURITY_AUTHENTICATION = "hadoop.security.authentication"; + + public static final String KEY_SECURITY_KERBEROS_ENABLE = "security.kerberos.login.enable"; + + public static final String KEY_SECURITY_KERBEROS_DEBUG = "security.kerberos.login.debug"; + + public static final String KEY_SECURITY_KERBEROS_KEYTAB = "security.kerberos.login.keytab"; + + public static final String KEY_SECURITY_KERBEROS_PRINCIPAL = "security.kerberos.login.principal"; + + public static final String KEY_SECURITY_KERBEROS_KRB5_CONF = "security.kerberos.login.krb5"; + + public static final String KEY_SPARK_MAIN_CLASS = "spark.main.class"; + + public static final String KEY_SPARK_APP_NAME = "spark.app.name"; + + public static final String KEY_SPARK_BATCH_DURATION = "spark.batch.duration"; + + public static final String KEY_SPARK_DRIVER_CORES = "spark.driver.cores"; + + public static final String KEY_SPARK_DRIVER_MEMORY = "spark.driver.memory"; + + public static final String KEY_SPARK_EXECUTOR_INSTANCES = "spark.executor.instances"; + + public static final String KEY_SPARK_EXECUTOR_CORES = "spark.executor.cores"; + + public static final String KEY_SPARK_EXECUTOR_MEMORY = "spark.executor.memory"; + + public static final String KEY_SPARK_DYNAMIC_ALLOCATION_ENABLED = + "spark.dynamicAllocation.enabled"; + + public static final String KEY_SPARK_DYNAMIC_ALLOCATION_MAX_EXECUTORS = + "spark.dynamicAllocation.maxExecutors"; + + public static final String KEY_SPARK_YARN_QUEUE = "spark.yarn.queue"; + + public static final String KEY_SPARK_YARN_QUEUE_NAME = "yarnQueueName"; + + public static final String KEY_SPARK_YARN_QUEUE_LABEL = "yarnQueueLabel"; + + public static final String KEY_SPARK_YARN_AM_NODE_LABEL = "spark.yarn.am.nodeLabelExpression"; + + public static final String KEY_SPARK_YARN_EXECUTOR_NODE_LABEL = + "spark.yarn.executor.nodeLabelExpression"; + + public static String KEY_SPARK_SQL(String prefix) { + return (prefix != null ? prefix : "") + "sql"; + } + + public static String KEY_SPARK_SQL() { + return KEY_SPARK_SQL(null); + } + + public static String KEY_APP_CONF(String prefix) { + return (prefix != null ? prefix : "") + "conf"; + } + + public static String KEY_APP_CONF() { + return KEY_APP_CONF(null); + } + + public static String KEY_FLINK_CONF(String prefix) { + return (prefix != null ? prefix : "") + "flink.conf"; + } + + public static String KEY_FLINK_CONF() { + return KEY_FLINK_CONF(null); + } + + public static String KEY_APP_NAME(String prefix) { + return (prefix != null ? prefix : "") + "app.name"; + } + + public static String KEY_APP_NAME() { + return KEY_APP_NAME(null); + } + + public static String KEY_FLINK_SQL(String prefix) { + return (prefix != null ? prefix : "") + "sql"; + } + + public static String KEY_FLINK_SQL() { + return KEY_FLINK_SQL(null); + } + + public static String KEY_FLINK_PARALLELISM(String prefix) { + return (prefix != null ? prefix : "") + "parallelism.default"; + } + + public static String KEY_FLINK_PARALLELISM() { + return KEY_FLINK_PARALLELISM(null); + } + + public static final String KEY_FLINK_OPTION_PREFIX = "flink.option."; + + public static final String KEY_FLINK_PROPERTY_PREFIX = "flink.property."; + + public static final String KEY_FLINK_TABLE_PREFIX = "flink.table."; + + public static final String KEY_SPARK_PROPERTY_PREFIX = "spark."; + + public static final String KEY_APP_PREFIX = "app."; + + public static final String KEY_SQL_PREFIX = "sql."; + + public static final String KEY_FLINK_APP_NAME = "pipeline.name"; + + public static final String KEY_YARN_APP_ID = "yarn.application.id"; + + public static final String KEY_YARN_APP_NAME = "yarn.application.name"; + + public static final String KEY_YARN_APP_QUEUE = "yarn.application.queue"; + + public static final String KEY_YARN_APP_NODE_LABEL = "yarn.application.node-label"; + + public static final String KEY_K8S_IMAGE_PULL_POLICY = + "kubernetes.container.image.pull-policy"; + + public static final String FLINK_NATIVE_KUBERNETES_LABEL = "flink-native-kubernetes"; + + public static final String KEY_FLINK_TABLE_PLANNER = "flink.table.planner"; + + public static final String KEY_FLINK_TABLE_MODE = "flink.table.mode"; + + public static final String KEY_FLINK_TABLE_CATALOG = "flink.table.catalog"; + + public static final String KEY_FLINK_TABLE_DATABASE = "flink.table.database"; + + public static final String KAFKA_SINK_PREFIX = "kafka.sink."; + + public static final String KAFKA_SOURCE_PREFIX = "kafka.source."; + + public static final String KEY_KAFKA_TOPIC = "topic"; + + public static final String KEY_KAFKA_SEMANTIC = "semantic"; + + public static final String KEY_KAFKA_PATTERN = "pattern"; + + public static final String KEY_KAFKA_START_FROM = "start.from"; + + public static final String KEY_KAFKA_START_FROM_OFFSET = "offset"; + + public static final String KEY_KAFKA_START_FROM_TIMESTAMP = "timestamp"; + + public static final String KEY_ALIAS = "alias"; + + public static final String KEY_JDBC_PREFIX = "jdbc."; + + public static final String KEY_JDBC_DRIVER = "driverClassName"; + + public static final String KEY_JDBC_URL = "jdbcUrl"; + + public static final String KEY_JDBC_USER = "username"; + + public static final String KEY_JDBC_PASSWORD = "password"; + + public static final String KEY_JDBC_INSERT_BATCH = "batch.size"; + + public static final int DEFAULT_JDBC_INSERT_BATCH = 1; + + public static final String MONGO_PREFIX = "mongodb."; + + public static final String HBASE_PREFIX = "hbase."; + + public static final String KEY_HBASE_COMMIT_BATCH = "hbase.commit.batch"; + + public static final String KEY_HBASE_WRITE_SIZE = "hbase.client.write.size"; + + public static final int DEFAULT_HBASE_COMMIT_BATCH = 1000; + + public static final String KEY_HBASE_AUTH_USER = "hbase.auth.user"; + + public static final int DEFAULT_HBASE_WRITE_SIZE = 1024 * 1024 * 10; + + public static final String INFLUX_PREFIX = "influx."; + + public static final String KEY_FLINK_APPLICATION_MAIN_CLASS = "$internal.application.main"; + + public static final String KEY_FLINK_JM_PROCESS_MEMORY = "jobmanager.memory.process.size"; + + public static final String KEY_FLINK_TM_PROCESS_MEMORY = "taskmanager.memory.process.size"; +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigOption.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigOption.java new file mode 100644 index 0000000000..1bd3b5d519 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/ConfigOption.java @@ -0,0 +1,145 @@ +/* + * 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.streampark.common.conf; + +import org.apache.streampark.common.util.TypeCastUtils; + +import java.util.Properties; +import java.util.function.Function; + +/** + * @param type of the configuration value + */ +public final class ConfigOption { + + private final String key; + private final T defaultValue; + private final boolean required; + private final Class classType; + private final String description; + private final Function handle; + private final String prefix; + private final Properties prop; + private final String fullKey; + + private ConfigOption(Builder b) { + this.key = b.key; + this.defaultValue = b.defaultValue; + this.required = b.required; + this.classType = b.classType; + this.description = b.description != null ? b.description : ""; + this.handle = b.handle; + this.prefix = b.prefix != null ? b.prefix : ""; + this.prop = b.prop; + this.fullKey = (!this.prefix.isEmpty()) ? this.prefix + "." + this.key : this.key; + } + + public T get() { + if (handle != null) { + if (required) { + try { + return handle.apply(fullKey); + } catch (Exception e) { + throw error(e.getMessage()); + } + } else { + try { + return handle.apply(fullKey); + } catch (Exception e) { + return defaultValue; + } + } + } else { + if (required) { + Object v = prop.get(fullKey); + if (v == null) + throw error("Is require"); + return TypeCastUtils.cast(v.toString(), classType); + } else { + String v = prop.getProperty(fullKey); + if (v == null) + return defaultValue; + return TypeCastUtils.cast(v, classType); + } + } + } + + public IllegalArgumentException error(String message) { + return new IllegalArgumentException( + "[StreamPark] config error: key:" + fullKey + ", detail: " + message); + } + + public static Builder builder(Class classType) { + return new Builder<>(classType); + } + + public static final class Builder { + + private final Class classType; + private String key; + private T defaultValue; + private boolean required = false; + private String description; + private Function handle; + private String prefix; + private Properties prop; + + private Builder(Class classType) { + this.classType = classType; + } + + public Builder key(String key) { + this.key = key; + return this; + } + + public Builder defaultValue(T defaultValue) { + this.defaultValue = defaultValue; + return this; + } + + public Builder required(boolean required) { + this.required = required; + return this; + } + + public Builder description(String description) { + this.description = description; + return this; + } + + public Builder handle(Function handle) { + this.handle = handle; + return this; + } + + public Builder prefix(String prefix) { + this.prefix = prefix; + return this; + } + + public Builder prop(Properties prop) { + this.prop = prop; + return this; + } + + public ConfigOption build() { + return new ConfigOption<>(this); + } + } +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/FlinkVersion.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/FlinkVersion.java new file mode 100644 index 0000000000..df71446310 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/FlinkVersion.java @@ -0,0 +1,244 @@ +/* + * 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.streampark.common.conf; + +import org.apache.streampark.common.util.CommandUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.Serializable; +import java.net.MalformedURLException; +import java.net.URL; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** @param flinkHome actual flink home that must be a readable local path */ +public class FlinkVersion implements Serializable { + + private static final Logger log = LoggerFactory.getLogger(FlinkVersion.class); + + private static final Pattern FLINK_VER_PATTERN = Pattern.compile("^(\\d+\\.\\d+)(\\.)?.*$"); + private static final Pattern FLINK_VERSION_PATTERN = + Pattern.compile("^Version: (.*), Commit ID: (.*)$"); + private static final Pattern FLINK_SCALA_VERSION_PATTERN = + Pattern.compile("^flink-dist_(\\d\\.\\d+).*.jar$"); + private static final Pattern APACHE_FLINK_VERSION_PATTERN = Pattern.compile("(^\\d+\\.\\d+\\.\\d+)"); + private static final Pattern OTHER_FLINK_VERSION_PATTERN = Pattern.compile("(\\d+\\.\\d+)(-*)"); + + private final String flinkHome; + private final File flinkLib; + private final List flinkLibs; + private final File flinkDistJar; + private final String scalaVersion; + private final String version; + private final String majorVersion; + private final String fullVersion; + + public FlinkVersion(String flinkHome) { + this.flinkHome = flinkHome; + this.flinkLib = resolveFlinkLib(flinkHome); + this.flinkLibs = resolveFlinkLibs(this.flinkLib); + this.flinkDistJar = resolveFlinkDistJar(this.flinkLib); + this.scalaVersion = parseScalaVersion(this.flinkDistJar); + this.version = parseVersion(this.flinkLib, this.flinkDistJar); + this.majorVersion = parseMajorVersion(this.version); + this.fullVersion = this.version + "_" + this.scalaVersion; + } + + public String flinkHome() { + return flinkHome; + } + + public File flinkLib() { + return flinkLib; + } + + public List flinkLibs() { + return flinkLibs; + } + + public File flinkDistJar() { + return flinkDistJar; + } + + public String scalaVersion() { + return scalaVersion; + } + + public String version() { + return version; + } + + public String majorVersion() { + return majorVersion; + } + + public String fullVersion() { + return fullVersion; + } + + public boolean checkVersion() { + return checkVersion(true); + } + + public boolean checkVersion(boolean throwException) { + String[] parts = version().split("\\."); + try { + if (parts.length == 3) { + int major = Integer.parseInt(parts[0].trim()); + int minor = Integer.parseInt(parts[1].trim()); + if (major == 1 && minor >= 12 && minor <= 20) { + return true; + } + } + } catch (NumberFormatException e) { + // fall through + } + if (throwException) { + throw new UnsupportedOperationException("Unsupported flink version: " + version()); + } + return false; + } + + public boolean checkVersion(int sinceVersion) { + String[] parts = version().split("\\."); + try { + if (parts.length == 3) { + int major = Integer.parseInt(parts[0].trim()); + int minor = Integer.parseInt(parts[1].trim()); + return major == 1 && minor >= sinceVersion; + } + } catch (NumberFormatException e) { + // fall through + } + return false; + } + + @Override + public String toString() { + String shimsVersion = "streampark-flink-shims_flink-" + majorVersion(); + return "\n" + + "----------------------------------------- flink version -----------------------------------\n" + + " flinkHome : " + flinkHome() + "\n" + + " distJarName : " + flinkDistJar().getName() + "\n" + + " flinkVersion : " + version() + "\n" + + " majorVersion : " + majorVersion() + "\n" + + " scalaVersion : " + scalaVersion() + "\n" + + " shimsVersion : " + shimsVersion + "\n" + + "-------------------------------------------------------------------------------------------\n"; + } + + private static File resolveFlinkLib(String flinkHome) { + if (flinkHome == null) { + throw new IllegalArgumentException("[StreamPark] flinkHome must not be null."); + } + if (!new File(flinkHome).exists()) { + throw new IllegalArgumentException("[StreamPark] flinkHome must be exists."); + } + File lib = new File(flinkHome + "/lib"); + if (!lib.exists() || !lib.isDirectory()) { + throw new IllegalArgumentException( + "[StreamPark] " + flinkHome + "/lib must be exists and must be directory."); + } + return lib; + } + + private static List resolveFlinkLibs(File flinkLib) { + File[] files = flinkLib.listFiles(); + if (files == null) { + return Collections.emptyList(); + } + List urls = new ArrayList<>(files.length); + for (File f : files) { + try { + urls.add(f.toURI().toURL()); + } catch (MalformedURLException e) { + throw new RuntimeException(e); + } + } + return urls; + } + + private static File resolveFlinkDistJar(File flinkLib) { + File[] distJars = + flinkLib.listFiles(f -> f.getName().matches("flink-dist.*\\.jar")); + if (distJars == null || distJars.length == 0) { + throw new IllegalArgumentException( + "[StreamPark] can no found flink-dist jar in " + flinkLib); + } + if (distJars.length > 1) { + throw new IllegalArgumentException( + "[StreamPark] found multiple flink-dist jar in " + flinkLib); + } + return distJars[0]; + } + + private static String parseScalaVersion(File flinkDistJar) { + Matcher matcher = FLINK_SCALA_VERSION_PATTERN.matcher(flinkDistJar.getName()); + if (matcher.matches()) { + return matcher.group(1); + } + return "2.12"; + } + + private static String parseVersion(File flinkLib, File flinkDistJar) { + List cmd = + Collections.singletonList( + "java -classpath " + + flinkDistJar.getName() + + " org.apache.flink.client.cli.CliFrontend --version"); + StringBuilder buffer = new StringBuilder(); + String[] ref = {null}; + + CommandUtils.execute( + flinkLib.getAbsolutePath(), + cmd, + out -> { + buffer.append(out).append("\n"); + Matcher m = FLINK_VERSION_PATTERN.matcher(out); + if (m.find()) { + String v = m.group(1); + if (APACHE_FLINK_VERSION_PATTERN.matcher(v).find()) { + ref[0] = v; + } else if (OTHER_FLINK_VERSION_PATTERN.matcher(v).find()) { + ref[0] = v; + } + } + }); + + log.info(buffer.toString()); + if (ref[0] == null) { + throw new IllegalStateException( + "[StreamPark] parse flink version failed. " + buffer); + } + return ref[0]; + } + + private static String parseMajorVersion(String version) { + if (version == null) + return null; + Matcher matcher = FLINK_VER_PATTERN.matcher(version); + matcher.matches(); + return matcher.group(1); + } +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalConfigHolder.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalConfigHolder.java new file mode 100644 index 0000000000..a9ee0560aa --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalConfigHolder.java @@ -0,0 +1,142 @@ +/* + * 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.streampark.common.conf; + +import org.apache.streampark.common.constants.Constants; +import org.apache.streampark.common.util.SystemPropertyUtils; +import org.apache.streampark.common.util.TypeCastUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** Thread-safe configuration storage containers. */ +public final class InternalConfigHolder { + + private static final Logger log = LoggerFactory.getLogger(InternalConfigHolder.class); + + private static final int INITIAL_CAPACITY = 45; + + private static final ConcurrentHashMap confData = + new ConcurrentHashMap<>(INITIAL_CAPACITY); + + private static final ConcurrentHashMap> confOptions = + new ConcurrentHashMap<>(INITIAL_CAPACITY); + + private InternalConfigHolder() { + } + + public static void initConfigHub() { + CommonConfig.STREAMPARK_WORKSPACE_LOCAL(); + K8sFlinkConfig.INGRESS_CLASS(); + } + + static void register(@Nonnull InternalOption conf) { + confOptions.put(conf.getKey(), conf); + if (conf.getDefaultValue() != null) { + confData.put(conf.getKey(), conf.getDefaultValue()); + } + } + + @Nonnull + public static T get(@Nonnull InternalOption conf) { + Object value = confData.get(conf.getKey()); + if (value == null || value.equals(conf.getDefaultValue())) { + String v = SystemPropertyUtils.get(conf.getKey()); + if (v != null) { + if (!v.equals(value)) { + set(conf, v); + } + return TypeCastUtils.cast(v, conf.getClassType()); + } + return conf.getDefaultValue(); + } + return TypeCastUtils.cast(value.toString(), conf.getClassType()); + } + + @Nonnull + @SuppressWarnings("unchecked") + public static T get(@Nonnull String key) { + Object v = confData.get(key); + if (v != null) { + return (T) v; + } + InternalOption conf = confOptions.get(key); + if (conf != null) { + return (T) conf.getDefaultValue(); + } + InternalOptionSpec config = getConfig(key); + String sysProp = SystemPropertyUtils.get(key); + if (sysProp != null) { + return TypeCastUtils.cast(sysProp, config.getClassType()); + } + throw new IllegalArgumentException("Config key has not been registered: " + key); + } + + @Nullable + public static InternalOptionSpec getConfig(String key) { + return confOptions.get(key); + } + + @Nonnull + public static Set keys() { + Map> copy = new HashMap<>(confOptions); + return copy.keySet(); + } + + public static void set(@Nonnull InternalOptionSpec conf, Object value) { + if (!confOptions.containsKey(conf.getKey())) { + throw new IllegalArgumentException("config key has not been registered: " + conf.getKey()); + } + if (value == null) { + confData.remove(conf.getKey()); + return; + } + if (conf.getClassType() != value.getClass()) { + throw new IllegalArgumentException( + "config value type is not match of " + + conf.getKey() + + ", required: " + + conf.getClassType() + + ", actual: " + + value.getClass()); + } + SystemPropertyUtils.set(conf.getKey(), value.toString()); + confData.put(conf.getKey(), value); + } + + public static void log() { + Set configKeys = keys(); + StringBuilder sb = new StringBuilder(); + sb.append("Registered configs:\n"); + sb.append("ConfigHub collected configs: ").append(configKeys.size()).append("\n"); + for (String key : configKeys) { + String value = + key.contains("password") ? Constants.DEFAULT_DATAMASK_STRING : String.valueOf(get(key)); + sb.append(" ").append(key).append(" = ").append(value).append("\n"); + } + log.info(sb.toString()); + } +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOption.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOption.java new file mode 100644 index 0000000000..50a5497a26 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOption.java @@ -0,0 +1,80 @@ +/* + * 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.streampark.common.conf; + +/** + * Internal use of the system. + * + * @param the type of the configuration value. + * @param key key of configuration that consistent with the spring config. + * @param defaultValue default value of configuration that should not be null. + * @param classType the class type of value. please use java class type. + * @param description description of configuration. + */ +public final class InternalOption implements InternalOptionSpec { + + private final String key; + private final T defaultValue; + private final Class classType; + private final String description; + + public InternalOption(String key, T defaultValue, Class classType, String description) { + this.key = key; + this.defaultValue = defaultValue; + this.classType = classType; + this.description = description; + InternalConfigHolder.register(this); + } + + public InternalOption(String key, T defaultValue, Class classType) { + this(key, defaultValue, classType, ""); + } + + // Scala-compatible accessors (matching case class field names) + public String key() { + return key; + } + + public T defaultValue() { + return defaultValue; + } + + public Class classType() { + return classType; + } + + public String description() { + return description; + } + + public String getKey() { + return key; + } + + public T getDefaultValue() { + return defaultValue; + } + + public Class getClassType() { + return classType; + } + + public String getDescription() { + return description; + } +} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalOption.scala b/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOptionSpec.java similarity index 59% rename from streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalOption.scala rename to streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOptionSpec.java index 86816356ac..6ae81c46e9 100644 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalOption.scala +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/InternalOptionSpec.java @@ -15,24 +15,28 @@ * limitations under the License. */ -package org.apache.streampark.common.conf +package org.apache.streampark.common.conf; /** - * Internal use of the system - * - * @param key - * key of configuration that consistent with the spring config. - * @param defaultValue - * default value of configuration that should not be null. - * @param classType - * the class type of value. please use java class type. - * @param description - * description of configuration. + * Non-generic view of {@link InternalOption} used where the type parameter is unknown at compile + * time (e.g. when looking up options by key from a map). */ -case class InternalOption( - key: String, - defaultValue: Any, - classType: Class[_], - description: String = "") { - InternalConfigHolder.register(this) +public interface InternalOptionSpec { + + String getKey(); + + Object getDefaultValue(); + + Class getClassType(); + + String getDescription(); + + // Scala-compatible accessors + String key(); + + Object defaultValue(); + + Class classType(); + + String description(); } diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/K8sFlinkConfig.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/K8sFlinkConfig.java new file mode 100644 index 0000000000..28c242100c --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/K8sFlinkConfig.java @@ -0,0 +1,117 @@ +/* + * 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.streampark.common.conf; + +/** Flink kubernetes Configuration for v1 version */ +public final class K8sFlinkConfig { + + private K8sFlinkConfig() { + } + + @Deprecated + private static final InternalOption JOB_STATUS_TRACK_TASK_TIMEOUT_SEC = + new InternalOption<>( + "streampark.flink-k8s.tracking.polling-task-timeout-sec.job-status", + 120L, + Long.class, + "run timeout seconds of single flink-k8s metrics tracking task"); + + @Deprecated + public static InternalOption JOB_STATUS_TRACK_TASK_TIMEOUT_SEC() { + return JOB_STATUS_TRACK_TASK_TIMEOUT_SEC; + } + + @Deprecated + private static final InternalOption JOB_STATUS_TRACK_CACHE_TIMEOUT_SEC = + new InternalOption<>( + "streampark.flink-k8s.tracking.cache-timeout-sec.job-status", + 300, + Integer.class, + "status cache timeout seconds of single flink-k8s job status tracking task"); + + @Deprecated + public static InternalOption JOB_STATUS_TRACK_CACHE_TIMEOUT_SEC() { + return JOB_STATUS_TRACK_CACHE_TIMEOUT_SEC; + } + + @Deprecated + private static final InternalOption METRIC_TRACK_TASK_TIMEOUT_SEC = + new InternalOption<>( + "streampark.flink-k8s.tracking.polling-task-timeout-sec.cluster-metric", + 120L, + Long.class, + "run timeout seconds of single flink-k8s job status tracking task"); + + @Deprecated + public static InternalOption METRIC_TRACK_TASK_TIMEOUT_SEC() { + return METRIC_TRACK_TASK_TIMEOUT_SEC; + } + + @Deprecated + private static final InternalOption JOB_STATUS_TRACK_TASK_INTERVAL_SEC = + new InternalOption<>( + "streampark.flink-k8s.tracking.polling-interval-sec.job-status", + 5L, + Long.class, + "interval seconds between two single flink-k8s metrics tracking task"); + + @Deprecated + public static InternalOption JOB_STATUS_TRACK_TASK_INTERVAL_SEC() { + return JOB_STATUS_TRACK_TASK_INTERVAL_SEC; + } + + @Deprecated + private static final InternalOption METRIC_TRACK_TASK_INTERVAL_SEC = + new InternalOption<>( + "streampark.flink-k8s.tracking.polling-interval-sec.cluster-metric", + 5L, + Long.class, + "interval seconds between two single flink-k8s metrics tracking task"); + + @Deprecated + public static InternalOption METRIC_TRACK_TASK_INTERVAL_SEC() { + return METRIC_TRACK_TASK_INTERVAL_SEC; + } + + @Deprecated + private static final InternalOption SILENT_STATE_JOB_KEEP_TRACKING_SEC = + new InternalOption<>( + "streampark.flink-k8s.tracking.silent-state-keep-sec", + 60, + Integer.class, + "retained tracking time for SILENT state flink tasks"); + + @Deprecated + public static InternalOption SILENT_STATE_JOB_KEEP_TRACKING_SEC() { + return SILENT_STATE_JOB_KEEP_TRACKING_SEC; + } + + private static final InternalOption INGRESS_CLASS = + new InternalOption<>( + "streampark.flink-k8s.ingress.class", + "nginx", + String.class, + "Direct ingress to the ingress controller."); + + public static InternalOption INGRESS_CLASS() { + return INGRESS_CLASS; + } + + @Deprecated + public static final String DEFAULT_KUBERNETES_NAMESPACE = "default"; +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/SparkVersion.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/SparkVersion.java new file mode 100644 index 0000000000..122495c804 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/SparkVersion.java @@ -0,0 +1,167 @@ +/* + * 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.streampark.common.conf; + +import org.apache.streampark.common.util.CommandUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.Serializable; +import java.util.Collections; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** @param sparkHome actual spark home that must be a readable local path */ +public class SparkVersion implements Serializable { + + private static final Logger log = LoggerFactory.getLogger(SparkVersion.class); + + private static final Pattern SPARK_VER_PATTERN = Pattern.compile("^(\\d+\\.\\d+)(\\.)?.*$"); + private static final Pattern SPARK_VERSION_PATTERN = + Pattern.compile("\\s{2}version\\s(\\d+\\.\\d+\\.\\d+)"); + private static final Pattern SPARK_SCALA_VERSION_PATTERN = + Pattern.compile("Using\\sScala\\sversion\\s(\\d+\\.\\d+)"); + + private final String sparkHome; + private final String version; + private final String scalaVersion; + private final String majorVersion; + private final String fullVersion; + private final File sparkLib; + + public SparkVersion(String sparkHome) { + this.sparkHome = sparkHome; + String[] parsed = parseVersionAndScala(sparkHome); + this.version = parsed[0]; + this.scalaVersion = parsed[1]; + this.majorVersion = parseMajorVersion(this.version); + this.fullVersion = this.version + "_" + this.scalaVersion; + this.sparkLib = resolveSparkLib(sparkHome); + } + + public String sparkHome() { + return sparkHome; + } + + public String version() { + return version; + } + + public String scalaVersion() { + return scalaVersion; + } + + public String majorVersion() { + return majorVersion; + } + + public String fullVersion() { + return fullVersion; + } + + public File sparkLib() { + return sparkLib; + } + + public boolean checkVersion() { + return checkVersion(true); + } + + public boolean checkVersion(boolean throwException) { + String[] parts = version().split("\\."); + try { + if (parts.length == 3) { + int major = Integer.parseInt(parts[0].trim()); + if (major == 2 || major == 3) { + return true; + } + } + } catch (NumberFormatException e) { + // fall through + } + if (throwException) { + throw new UnsupportedOperationException("Unsupported spark version: " + version()); + } + return false; + } + + @Override + public String toString() { + return "\n" + + "----------------------------------------- spark version -----------------------------------\n" + + " sparkHome : " + sparkHome() + "\n" + + " sparkVersion : " + version() + "\n" + + " scalaVersion : " + scalaVersion() + "\n" + + "-------------------------------------------------------------------------------------------\n"; + } + + private static String[] parseVersionAndScala(String sparkHome) { + String[] refs = {null, null}; + StringBuilder buffer = new StringBuilder(); + java.util.List cmd = + Collections.singletonList( + "export SPARK_HOME=" + sparkHome + "&&" + sparkHome + "/bin/spark-submit --version"); + + CommandUtils.execute( + sparkHome, + cmd, + out -> { + buffer.append(out).append("\n"); + Matcher m = SPARK_VERSION_PATTERN.matcher(out); + if (m.find()) { + refs[0] = m.group(1); + } else { + Matcher m1 = SPARK_SCALA_VERSION_PATTERN.matcher(out); + if (m1.find()) { + refs[1] = m1.group(1); + } + } + }); + + log.info(buffer.toString()); + if (refs[0] == null || refs[1] == null) { + throw new IllegalStateException("[StreamPark] parse spark version failed. " + buffer); + } + return refs; + } + + private static String parseMajorVersion(String version) { + if (version == null) + return null; + Matcher matcher = SPARK_VER_PATTERN.matcher(version); + matcher.matches(); + return matcher.group(1); + } + + private static File resolveSparkLib(String sparkHome) { + if (sparkHome == null) { + throw new IllegalArgumentException("[StreamPark] sparkHome must not be null."); + } + if (!new File(sparkHome).exists()) { + throw new IllegalArgumentException("[StreamPark] sparkHome must be exists."); + } + File lib = new File(sparkHome + "/jars"); + if (!lib.exists() || !lib.isDirectory()) { + throw new IllegalArgumentException( + "[StreamPark] " + sparkHome + "/lib must be exists and must be directory."); + } + return lib; + } +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/conf/Workspace.java b/streampark-common/src/main/java/org/apache/streampark/common/conf/Workspace.java new file mode 100644 index 0000000000..fe995ebd91 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/conf/Workspace.java @@ -0,0 +1,209 @@ +/* + * 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.streampark.common.conf; + +import org.apache.streampark.common.enums.StorageType; +import org.apache.streampark.common.util.HdfsUtils; +import org.apache.streampark.common.util.SystemPropertyUtils; +import org.apache.streampark.common.util.TypeCastUtils; + +import java.net.URI; + +public final class Workspace { + + private static volatile Workspace localInstance; + private static volatile Workspace remoteInstance; + + private final String workspace; + private final String appPlugins; + private final String appClient; + private final String appShims; + private final String appUploads; + private final String appPython; + private final String appPythonVenv; + private final String appWorkspace; + private final String appFlink; + private final String appSpark; + private final String appBackups; + private final String appSavepoints; + private final String appJars; + + private Workspace(StorageType storageType) { + this.workspace = computeWorkspace(storageType); + this.appPlugins = workspace + "/plugins"; + this.appClient = workspace + "/client"; + this.appShims = workspace + "/shims"; + this.appUploads = workspace + "/uploads"; + this.appPython = workspace + "/python"; + this.appPythonVenv = appPython + "/venv.zip"; + this.appWorkspace = workspace + "/workspace"; + this.appFlink = workspace + "/flink"; + this.appSpark = workspace + "/spark"; + this.appBackups = workspace + "/backups"; + this.appSavepoints = workspace + "/savepoints"; + this.appJars = workspace + "/jars"; + } + + public static Workspace of(StorageType storageType) { + return new Workspace(storageType); + } + + public static Workspace local() { + if (localInstance == null) { + synchronized (Workspace.class) { + if (localInstance == null) { + localInstance = new Workspace(StorageType.LFS); + } + } + } + return localInstance; + } + + public static Workspace remote() { + if (remoteInstance == null) { + synchronized (Workspace.class) { + if (remoteInstance == null) { + remoteInstance = new Workspace(StorageType.HDFS); + } + } + } + return remoteInstance; + } + + /** local build path */ + public static String APP_LOCAL_DIST() { + return local().WORKSPACE() + "/dist"; + } + + /** dirPath of the maven local repository with built-in compilation process */ + public static String MAVEN_LOCAL_PATH() { + return local().WORKSPACE() + "/mvnrepo"; + } + + /** local sourceCode path.(for git...) */ + public static String PROJECT_LOCAL_PATH() { + return local().WORKSPACE() + "/project"; + } + + /** local log path. */ + public static String LOG_LOCAL_PATH() { + return local().WORKSPACE() + "/logs"; + } + + /** project build log path. */ + public static String PROJECT_BUILD_LOG_PATH() { + return LOG_LOCAL_PATH() + "/build_logs"; + } + + /** project archives path */ + public static String ARCHIVES_FILE_PATH() { + return remote().WORKSPACE() + "/historyserver/archive"; + } + + public String WORKSPACE() { + return workspace; + } + + public String APP_PLUGINS() { + return appPlugins; + } + + public String APP_CLIENT() { + return appClient; + } + + /** store flink multi version support jars */ + public String APP_SHIMS() { + return appShims; + } + + public String APP_UPLOADS() { + return appUploads; + } + + public String APP_PYTHON() { + return appPython; + } + + public String APP_PYTHON_VENV() { + return appPythonVenv; + } + + public String APP_WORKSPACE() { + return appWorkspace; + } + + public String APP_FLINK() { + return appFlink; + } + + public String APP_SPARK() { + return appSpark; + } + + public String APP_BACKUPS() { + return appBackups; + } + + public String APP_SAVEPOINTS() { + return appSavepoints; + } + + /** store global public jars */ + public String APP_JARS() { + return appJars; + } + + private T getConfigValue(InternalOption option) { + String s = SystemPropertyUtils.get(option.getKey()); + T v = InternalConfigHolder.get(option); + if (s != null && v.equals(option.getDefaultValue())) + return TypeCastUtils.cast(s, option.getClassType()); + return v; + } + + private String computeWorkspace(StorageType storageType) { + switch (storageType) { + case LFS: { + String path = getConfigValue(CommonConfig.STREAMPARK_WORKSPACE_LOCAL()); + if (path == null) { + throw new IllegalArgumentException( + "[StreamPark] streampark.workspace.local must not be null"); + } + return path; + } + case HDFS: { + String path = getConfigValue(CommonConfig.STREAMPARK_WORKSPACE_REMOTE()); + if (path.isEmpty()) { + return HdfsUtils.getDefaultFS() + + CommonConfig.STREAMPARK_WORKSPACE_REMOTE().getDefaultValue(); + } + String defaultFs = HdfsUtils.getDefaultFS(); + if (path.startsWith("hdfs://")) { + if (path.startsWith(defaultFs)) { + return path; + } + return defaultFs + URI.create(path).getPath(); + } + return defaultFs + path; + } + default: + throw new IllegalArgumentException("Unsupported storage type: " + storageType); + } + } +} diff --git a/streampark-common/src/main/java/org/apache/streampark/common/util/TypeCastUtils.java b/streampark-common/src/main/java/org/apache/streampark/common/util/TypeCastUtils.java new file mode 100644 index 0000000000..d30fb314c1 --- /dev/null +++ b/streampark-common/src/main/java/org/apache/streampark/common/util/TypeCastUtils.java @@ -0,0 +1,45 @@ +/* + * 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.streampark.common.util; + +public final class TypeCastUtils { + + private TypeCastUtils() { + } + + @SuppressWarnings("unchecked") + public static T cast(String value, Class classType) { + if (classType == String.class) + return (T) value; + if (classType == Integer.class || classType == int.class) + return (T) Integer.valueOf(value); + if (classType == Long.class || classType == long.class) + return (T) Long.valueOf(value); + if (classType == Boolean.class || classType == boolean.class) + return (T) Boolean.valueOf(value); + if (classType == Double.class || classType == double.class) + return (T) Double.valueOf(value); + if (classType == Float.class || classType == float.class) + return (T) Float.valueOf(value); + if (classType == Short.class || classType == short.class) + return (T) Short.valueOf(value); + if (classType == Byte.class || classType == byte.class) + return (T) Byte.valueOf(value); + throw new IllegalArgumentException("Unsupported type: " + classType); + } +} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/CommonConfig.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/CommonConfig.scala deleted file mode 100644 index c2d2540a77..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/CommonConfig.scala +++ /dev/null @@ -1,117 +0,0 @@ -/* - * 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.streampark.common.conf - -import java.lang.{Integer => JavaInt, Long => JavaLong} - -object CommonConfig { - - val STREAMPARK_WORKSPACE_LOCAL: InternalOption = InternalOption( - key = "streampark.workspace.local", - defaultValue = "/streampark", - classType = classOf[String]) - - val STREAMPARK_WORKSPACE_REMOTE: InternalOption = InternalOption( - key = "streampark.workspace.remote", - defaultValue = "/streampark", - classType = classOf[String]) - - val STREAMPARK_HADOOP_USER_NAME: InternalOption = InternalOption( - key = "streampark.hadoop-user-name", - defaultValue = "hdfs", - classType = classOf[String]) - - val STREAMPARK_PROXY_YARN_URL: InternalOption = InternalOption( - key = "streampark.proxy.yarn-url", - defaultValue = "", - classType = classOf[String], - description = "proxy yarn url. ex: knox proxy or other") - - val STREAMPARK_YARN_AUTH: InternalOption = InternalOption( - key = "streampark.yarn.http-auth", - defaultValue = "", - classType = classOf[String], - description = "yarn http auth type. ex: simple, kerberos") - - val DOCKER_HOST: InternalOption = InternalOption( - key = "streampark.docker.http-client.docker-host", - defaultValue = "", - classType = classOf[String], - description = "docker host for DockerHttpClient") - - val DOCKER_MAX_CONNECTIONS: InternalOption = InternalOption( - key = "streampark.docker.http-client.max-connections", - defaultValue = 100, - classType = classOf[JavaInt], - description = "instantiating max connections for DockerHttpClient") - - val DOCKER_CONNECTION_TIMEOUT_SEC: InternalOption = InternalOption( - key = "streampark.docker.http-client.connection-timeout-sec", - defaultValue = 100L, - classType = classOf[JavaLong], - description = "instantiating connection timeout for DockerHttpClient") - - val DOCKER_RESPONSE_TIMEOUT_SEC: InternalOption = InternalOption( - key = "streampark.docker.http-client.response-timeout-sec", - defaultValue = 120L, - classType = classOf[JavaLong], - description = "instantiating connection timeout for DockerHttpClient") - - val MAVEN_SETTINGS_PATH: InternalOption = InternalOption( - key = "streampark.maven.settings", - defaultValue = null, - classType = classOf[String], - description = "maven settings.xml full path") - - val MAVEN_REMOTE_URL: InternalOption = InternalOption( - key = "streampark.maven.central.repository", - defaultValue = "https://repo1.maven.org/maven2/", - classType = classOf[String], - description = "maven repository used for built-in compilation") - - val MAVEN_AUTH_USER: InternalOption = InternalOption( - key = "streampark.maven.auth.user", - defaultValue = null, - classType = classOf[String], - description = "maven repository used for built-in compilation") - - val MAVEN_AUTH_PASSWORD: InternalOption = InternalOption( - key = "streampark.maven.auth.password", - defaultValue = null, - classType = classOf[String], - description = "maven repository used for built-in compilation") - - val KERBEROS_TTL: InternalOption = InternalOption( - key = "security.kerberos.ttl", - defaultValue = "2h", - classType = classOf[String], - description = "kerberos default ttl") - - val READ_LOG_MAX_SIZE: InternalOption = InternalOption( - key = "streampark.read-log.max-size", - defaultValue = "1mb", - classType = classOf[String], - description = "The maximum size of the default read log") - - val SPRING_PROFILES_ACTIVE: InternalOption = InternalOption( - key = "spring.profiles.active", - defaultValue = "h2", - classType = classOf[String], - description = "Use the database type") - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigKeys.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigKeys.scala deleted file mode 100644 index 7f321ee92c..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigKeys.scala +++ /dev/null @@ -1,204 +0,0 @@ -/* - * 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.streampark.common.conf - -object ConfigKeys { - - val PARAM_PREFIX = "--" - - /** about parameter... */ - val KEY_APP_HOME = "app.home" - - val KEY_HOST = "host" - - val KEY_PORT = "port" - - val KEY_DB = "db" - - val KEY_USER = "user" - - val KEY_PASSWORD = "password" - - val KEY_TIMEOUT = "timeout" - - val KEY_SEMANTIC = "semantic" - - /** kerberos */ - val KEY_KERBEROS = "kerberos" - - val KEY_KERBEROS_SERVICE_ACCOUNT = "kubernetes.service-account" - - val KEY_HADOOP_USER_NAME = "HADOOP_USER_NAME" - - /** hadoop.security.authentication */ - val KEY_HADOOP_SECURITY_AUTHENTICATION = "hadoop.security.authentication" - - val KEY_SECURITY_KERBEROS_ENABLE = "security.kerberos.login.enable" - - val KEY_SECURITY_KERBEROS_DEBUG = "security.kerberos.login.debug" - - val KEY_SECURITY_KERBEROS_KEYTAB = "security.kerberos.login.keytab" - - val KEY_SECURITY_KERBEROS_PRINCIPAL = "security.kerberos.login.principal" - - val KEY_SECURITY_KERBEROS_KRB5_CONF = "security.kerberos.login.krb5" - - /** about spark */ - val KEY_SPARK_MAIN_CLASS = "spark.main.class" - - val KEY_SPARK_APP_NAME = "spark.app.name" - - val KEY_SPARK_BATCH_DURATION = "spark.batch.duration" - - val KEY_SPARK_DRIVER_CORES = "spark.driver.cores" - - val KEY_SPARK_DRIVER_MEMORY = "spark.driver.memory" - - val KEY_SPARK_EXECUTOR_INSTANCES = "spark.executor.instances" - - val KEY_SPARK_EXECUTOR_CORES = "spark.executor.cores" - - val KEY_SPARK_EXECUTOR_MEMORY = "spark.executor.memory" - - val KEY_SPARK_DYNAMIC_ALLOCATION_ENABLED = "spark.dynamicAllocation.enabled" - - val KEY_SPARK_DYNAMIC_ALLOCATION_MAX_EXECUTORS = "spark.dynamicAllocation.maxExecutors" - - val KEY_SPARK_YARN_QUEUE = "spark.yarn.queue" - - val KEY_SPARK_YARN_QUEUE_NAME = "yarnQueueName" - - val KEY_SPARK_YARN_QUEUE_LABEL = "yarnQueueLabel" - - val KEY_SPARK_YARN_AM_NODE_LABEL = "spark.yarn.am.nodeLabelExpression" - - val KEY_SPARK_YARN_EXECUTOR_NODE_LABEL = "spark.yarn.executor.nodeLabelExpression" - - def KEY_SPARK_SQL(prefix: String = null): String = - s"${Option(prefix).getOrElse("")}sql" - - /** about config flink */ - def KEY_APP_CONF(prefix: String = null): String = - s"${Option(prefix).getOrElse("")}conf" - - def KEY_FLINK_CONF(prefix: String = null): String = - s"${Option(prefix).getOrElse("")}flink.conf" - - def KEY_APP_NAME(prefix: String = null): String = - s"${Option(prefix).getOrElse("")}app.name" - - def KEY_FLINK_SQL(prefix: String = null): String = - s"${Option(prefix).getOrElse("")}sql" - - def KEY_FLINK_PARALLELISM(prefix: String = null): String = - s"${Option(prefix).getOrElse("")}parallelism.default" - - val KEY_FLINK_OPTION_PREFIX = "flink.option." - - val KEY_FLINK_PROPERTY_PREFIX = "flink.property." - - val KEY_FLINK_TABLE_PREFIX = "flink.table." - - val KEY_SPARK_PROPERTY_PREFIX = "spark." - - val KEY_APP_PREFIX = "app." - - val KEY_SQL_PREFIX = "sql." - - val KEY_FLINK_APP_NAME = "pipeline.name" - - val KEY_YARN_APP_ID = "yarn.application.id" - - val KEY_YARN_APP_NAME = "yarn.application.name" - - val KEY_YARN_APP_QUEUE = "yarn.application.queue" - - val KEY_YARN_APP_NODE_LABEL = "yarn.application.node-label" - - val KEY_K8S_IMAGE_PULL_POLICY = "kubernetes.container.image.pull-policy" - - val FLINK_NATIVE_KUBERNETES_LABEL = "flink-native-kubernetes" - - /** about flink table */ - val KEY_FLINK_TABLE_PLANNER = "flink.table.planner" - - val KEY_FLINK_TABLE_MODE = "flink.table.mode" - - val KEY_FLINK_TABLE_CATALOG = "flink.table.catalog" - - val KEY_FLINK_TABLE_DATABASE = "flink.table.database" - - /** about config Kafka */ - val KAFKA_SINK_PREFIX = "kafka.sink." - - val KAFKA_SOURCE_PREFIX = "kafka.source." - - val KEY_KAFKA_TOPIC = "topic" - - val KEY_KAFKA_SEMANTIC = "semantic" - - val KEY_KAFKA_PATTERN = "pattern" - - val KEY_KAFKA_START_FROM = "start.from" - - val KEY_KAFKA_START_FROM_OFFSET = "offset" - - val KEY_KAFKA_START_FROM_TIMESTAMP = "timestamp" - - val KEY_ALIAS = "alias" - - /** about config jdbc... */ - val KEY_JDBC_PREFIX = "jdbc." - - val KEY_JDBC_DRIVER = "driverClassName" - - val KEY_JDBC_URL = "jdbcUrl" - - val KEY_JDBC_USER = "username" - - val KEY_JDBC_PASSWORD = "password" - - val KEY_JDBC_INSERT_BATCH = "batch.size" - - val DEFAULT_JDBC_INSERT_BATCH = 1 - - val MONGO_PREFIX = "mongodb." - - /** about config HBase */ - val HBASE_PREFIX = "hbase." - - val KEY_HBASE_COMMIT_BATCH = "hbase.commit.batch" - - val KEY_HBASE_WRITE_SIZE = "hbase.client.write.size" - - val DEFAULT_HBASE_COMMIT_BATCH = 1000 - - val KEY_HBASE_AUTH_USER = "hbase.auth.user" - - val DEFAULT_HBASE_WRITE_SIZE: Int = 1024 * 1024 * 10 - - /** about influx */ - val INFLUX_PREFIX = "influx." - - val KEY_FLINK_APPLICATION_MAIN_CLASS = "$internal.application.main" - - val KEY_FLINK_JM_PROCESS_MEMORY = "jobmanager.memory.process.size" - - val KEY_FLINK_TM_PROCESS_MEMORY = "taskmanager.memory.process.size" - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigOption.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigOption.scala deleted file mode 100644 index 0ec3c84b9d..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigOption.scala +++ /dev/null @@ -1,82 +0,0 @@ -/* - * 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.streampark.common.conf - -import org.apache.streampark.common.util.Implicits._ - -import java.util.Properties - -import scala.util.{Failure, Success, Try} - -/** - * @param key - * key of configuration that consistent with the spring config. - * @param defaultValue - * default value of configuration that should not be null. - * @param classType - * the class type of value. please use java class type. - * @param required - * is required - * @param description - * description of configuration. - * @param handle - * Processing function of special parameters - */ -case class ConfigOption[T]( - key: String, - defaultValue: T = null, - required: Boolean, - classType: Class[_], - description: String = "", - handle: String => T = null)(implicit prefix: String = "", prop: Properties) { - - private[this] lazy val fullKey = - if (prefix != null && prefix.nonEmpty) s"$prefix.$key" else key - - def get(): T = handle match { - case null => - if (required) { - prop.get(fullKey) match { - case null => throw error("Is require") - case v => v.toString.cast[T](classType) - } - } else { - prop.getProperty(fullKey) match { - case null => defaultValue - case v => v.cast[T](classType) - } - } - case _ => - if (required) { - Try(handle(fullKey)) match { - case Success(v) => v - case Failure(e) => throw error(e.getMessage) - } - } else { - Try(handle(fullKey)) match { - case Success(v) => v - case Failure(_) => defaultValue - } - } - } - - def error(message: String): Exception = { - new IllegalArgumentException(s"[StreamPark] config error: key:$fullKey, detail: $message") - } - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala deleted file mode 100644 index 83a09a227c..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala +++ /dev/null @@ -1,161 +0,0 @@ -/* - * 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.streampark.common.conf - -import org.apache.streampark.common.util.{CommandUtils, Logger} -import org.apache.streampark.common.util.Implicits._ - -import java.io.File -import java.net.URL -import java.util.function.Consumer -import java.util.regex.Pattern - -import scala.collection.mutable - -/** @param flinkHome actual flink home that must be a readable local path */ -class FlinkVersion(val flinkHome: String) extends Serializable with Logger { - - private[this] lazy val FLINK_VER_PATTERN = Pattern.compile("^(\\d+\\.\\d+)(\\.)?.*$") - - private[this] lazy val FLINK_VERSION_PATTERN = Pattern.compile("^Version: (.*), Commit ID: (.*)$") - - private[this] lazy val FLINK_SCALA_VERSION_PATTERN = - Pattern.compile("^flink-dist_(\\d\\.\\d+).*.jar$") - - private[this] lazy val APACHE_FLINK_VERSION_PATTERN = Pattern.compile("(^\\d+\\.\\d+\\.\\d+)") - - private[this] lazy val OTHER_FLINK_VERSION_PATTERN = Pattern.compile("(\\d+\\.\\d+)(-*)") - - lazy val scalaVersion: String = { - val matcher = FLINK_SCALA_VERSION_PATTERN.matcher(flinkDistJar.getName) - if (matcher.matches()) { - matcher.group(1) - } else { - // flink 1.15 + on support scala 2.12 - "2.12" - } - } - - lazy val fullVersion: String = s"${version}_$scalaVersion" - - lazy val flinkLib: File = { - require(flinkHome != null, "[StreamPark] flinkHome must not be null.") - require(new File(flinkHome).exists(), "[StreamPark] flinkHome must be exists.") - val lib = new File(s"$flinkHome/lib") - require( - lib.exists() && lib.isDirectory, - s"[StreamPark] $flinkHome/lib must be exists and must be directory.") - lib - } - - lazy val flinkLibs: List[URL] = flinkLib.listFiles().map(_.toURI.toURL).toList - - lazy val version: String = { - val cmd = List( - s"java -classpath ${flinkDistJar.getName} org.apache.flink.client.cli.CliFrontend --version") - var flinkVersion: String = null - val buffer = new mutable.StringBuilder - CommandUtils.execute( - flinkLib.getAbsolutePath, - cmd, - new Consumer[String]() { - override def accept(out: String): Unit = { - buffer.append(out).append("\n") - val matcher = FLINK_VERSION_PATTERN.matcher(out) - if (matcher.find) { - val version = matcher.group(1) - val matcher1 = APACHE_FLINK_VERSION_PATTERN.matcher(version) - if (matcher1.find) { - flinkVersion = version - } else { - val matcher2 = OTHER_FLINK_VERSION_PATTERN.matcher(version) - if (matcher2.find) { - flinkVersion = version - } - } - } - } - }) - - logInfo(buffer.toString()) - if (flinkVersion == null) { - throw new IllegalStateException(s"[StreamPark] parse flink version failed. $buffer") - } - buffer.clear() - flinkVersion - } - - // flink major version, like "1.13", "1.14" - lazy val majorVersion: String = { - if (version == null) { - null - } else { - val matcher = FLINK_VER_PATTERN.matcher(version) - matcher.matches() - matcher.group(1) - } - } - - lazy val flinkDistJar: File = { - val distJar = flinkLib.listFiles().filter(_.getName.matches("flink-dist.*\\.jar")) - distJar match { - case x if x.isEmpty => - throw new IllegalArgumentException(s"[StreamPark] can no found flink-dist jar in $flinkLib") - case x if x.length > 1 => - throw new IllegalArgumentException( - s"[StreamPark] found multiple flink-dist jar in $flinkLib") - case _ => - } - distJar.head - } - - def checkVersion(throwException: Boolean = true): Boolean = { - version.split("\\.").map(_.trim.toInt) match { - case Array(1, v, _) if v >= 12 && v <= 20 => true - case _ => - if (throwException) { - throw new UnsupportedOperationException(s"Unsupported flink version: $version") - } else { - false - } - } - } - - def checkVersion(sinceVersion: Int): Boolean = { - version.split("\\.").map(_.trim.toInt) match { - case Array(1, v, _) if v >= sinceVersion => true - case _ => false - } - } - - // StreamPark flink shims version, like "streampark-flink-shims_flink-1.13" - private lazy val shimsVersion: String = s"streampark-flink-shims_flink-$majorVersion" - - override def toString: String = - s""" - |----------------------------------------- flink version ----------------------------------- - | flinkHome : $flinkHome - | distJarName : ${flinkDistJar.getName} - | flinkVersion : $version - | majorVersion : $majorVersion - | scalaVersion : $scalaVersion - | shimsVersion : $shimsVersion - |------------------------------------------------------------------------------------------- - |""".stripMargin - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalConfigHolder.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalConfigHolder.scala deleted file mode 100644 index 977baf6629..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/InternalConfigHolder.scala +++ /dev/null @@ -1,176 +0,0 @@ -/* - * 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.streampark.common.conf - -import org.apache.streampark.common.constants.Constants -import org.apache.streampark.common.util.{Logger, SystemPropertyUtils} -import org.apache.streampark.common.util.Implicits._ - -import javax.annotation.{Nonnull, Nullable} - -import java.util -import java.util.concurrent.ConcurrentHashMap - -/** - * Thread-safe configuration storage containers. All configurations will be automatically - * initialized from the spring configuration items of the same name. - */ -object InternalConfigHolder extends Logger { - - private val initialCapacity = 45 - - /** configuration values storage (key -> value) */ - private val confData = new ConcurrentHashMap[String, Any](initialCapacity) - - /** configuration key options storage (key -> ConfigOption) */ - private val confOptions = - new ConcurrentHashMap[String, InternalOption](initialCapacity) - - /** Initialize the ConfigHub. */ - def initConfigHub(): Unit = { - Seq(CommonConfig, K8sFlinkConfig) - } - - /** Register the ConfigOption */ - private[conf] def register(@Nonnull conf: InternalOption): Unit = { - confOptions.put(conf.key, conf) - if (conf.defaultValue != null) { - confData.put(conf.key, conf.defaultValue) - } - } - - /** - * Get configuration value via ConfigOption. - * - * When using this api, the type must be explicitly declared and the relevant type will be - * automatically converted to some extent. 1) in scala: val result: Long = - * ConfigHub.get(K8sFlinkConfig.sglMetricTrackTaskTimeoutSec) 2) in java: Long result = - * ConfigHub.get(K8sFlinkConfig.sglMetricTrackTaskTimeoutSec()); - * - * @return - * return the defaultValue of ConfigOption when the value has not been overwritten. - */ - @Nonnull - def get[T](@Nonnull conf: InternalOption): T = { - val value = confData.get(conf.key) - if (value == null || value == conf.defaultValue) { - val v = SystemPropertyUtils.get(conf.key) - if (v != null) { - if (v != value) { - set(conf, v) - } - v.cast[T](conf.classType) - } else { - conf.defaultValue.asInstanceOf[T] - } - } else { - value.toString.cast[T](conf.classType) - } - } - - /** - * Get configuration value via key. - * - * When using this api, the type must be explicitly declared and the relevant type will be - * automatically converted to some extent. 1) in scala: val result: Long = - * ConfigHub.get("streampark.flink-k8s.tracking.polling-task-timeout-sec.cluster-metric") 2) in - * java: Long result = - * ConfigHub.get("streampark.flink-k8s.tracking.polling-task-timeout-sec.cluster-metric"); - * - * @throws IllegalArgumentException - * when the key has not been registered to ConfigHub. - * @return - * return the defaultValue of ConfigOption when the value has not been overwritten. - */ - @throws[IllegalArgumentException] - @Nonnull - def get[T](@Nonnull key: String): T = { - confData.get(key) match { - case null => - confOptions.get(key) match { - case null => - val config = getConfig(key) - SystemPropertyUtils.get(key) match { - case v if v != null => v.cast[T](config.classType) - case _ => - throw new IllegalArgumentException(s"Config key has not been registered: $key") - } - case conf: InternalOption => conf.defaultValue.asInstanceOf[T] - } - case v => v.asInstanceOf[T] - } - } - - /** - * Get registered ConfigOption by key. - * - * @return - * nullable - */ - @Nullable - def getConfig(key: String): InternalOption = { - confOptions.get(key) - } - - /** Get keys of all registered ConfigOption. */ - @Nonnull - def keys(): JavaSet[String] = { - val map = new util.HashMap[String, InternalOption](confOptions.size()) - map.putAll(confOptions) - map.keySet() - } - - /** - * Overwritten configuration value. - * - * @param conf - * should not be null. - * @param value - * the type of value should be same as conf.classType. - * @throws IllegalArgumentException - * when the conf has not been registered, or the value type is not same as conf.classType. - */ - @throws[IllegalArgumentException] - def set(@Nonnull conf: InternalOption, value: Any): Unit = { - if (!confOptions.containsKey(conf.key)) { - throw new IllegalArgumentException(s"config key has not been registered: $conf") - } - value match { - case null => confData.remove(conf.key) - case v if conf.classType != v.getClass => - throw new IllegalArgumentException( - s"config value type is not match of ${conf.key}, required: ${conf.classType}, actual: ${v.getClass}") - case v => - SystemPropertyUtils.set(conf.key, v.toString) - confData.put(conf.key, v) - } - } - - /** log the current configuration info. */ - def log(): Unit = { - val configKeys = keys() - logInfo(s"""Registered configs: - |ConfigHub collected configs: ${configKeys.size} - | ${configKeys - .map(key => - s"$key = ${if (key.contains("password")) Constants.DEFAULT_DATAMASK_STRING - else get(key)}") - .mkString("\n ")}""".stripMargin) - } - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/K8sFlinkConfig.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/K8sFlinkConfig.scala deleted file mode 100644 index 79173ac219..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/K8sFlinkConfig.scala +++ /dev/null @@ -1,81 +0,0 @@ -/* - * 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.streampark.common.conf - -/** Flink kubernetes Configuration for v1 version */ -object K8sFlinkConfig { - - // ======= deprecated ======= - @deprecated - val jobStatusTrackTaskTimeoutSec: InternalOption = InternalOption( - key = "streampark.flink-k8s.tracking.polling-task-timeout-sec.job-status", - defaultValue = 120L, - classType = classOf[java.lang.Long], - description = "run timeout seconds of single flink-k8s metrics tracking task") - - @deprecated - val jobStatusTrackCacheTimeoutSec: InternalOption = InternalOption( - key = "streampark.flink-k8s.tracking.cache-timeout-sec.job-status", - defaultValue = 300, - classType = classOf[java.lang.Integer], - description = "status cache timeout seconds of single flink-k8s job status tracking task") - - @deprecated - val metricTrackTaskTimeoutSec: InternalOption = InternalOption( - key = "streampark.flink-k8s.tracking.polling-task-timeout-sec.cluster-metric", - defaultValue = 120L, - classType = classOf[java.lang.Long], - description = "run timeout seconds of single flink-k8s job status tracking task") - - @deprecated - val jobStatueTrackTaskIntervalSec: InternalOption = InternalOption( - key = "streampark.flink-k8s.tracking.polling-interval-sec.job-status", - defaultValue = 5L, - classType = classOf[java.lang.Long], - description = "interval seconds between two single flink-k8s metrics tracking task") - - @deprecated - val metricTrackTaskIntervalSec: InternalOption = InternalOption( - key = "streampark.flink-k8s.tracking.polling-interval-sec.cluster-metric", - defaultValue = 5L, - classType = classOf[java.lang.Long], - description = "interval seconds between two single flink-k8s metrics tracking task") - - @deprecated - val silentStateJobKeepTrackingSec: InternalOption = InternalOption( - key = "streampark.flink-k8s.tracking.silent-state-keep-sec", - defaultValue = 60, - classType = classOf[java.lang.Integer], - description = "retained tracking time for SILENT state flink tasks") - - /** - * If an ingress controller is specified in the configuration, the ingress class - * kubernetes.io/ingress.class must be specified when creating the ingress, since there are often - * multiple ingress controllers in a production environment. - */ - val ingressClass: InternalOption = InternalOption( - key = "streampark.flink-k8s.ingress.class", - defaultValue = "nginx", - classType = classOf[java.lang.String], - description = "Direct ingress to the ingress controller.") - - /** kubernetes default namespace */ - @deprecated - val DEFAULT_KUBERNETES_NAMESPACE = "default" - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/SparkVersion.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/SparkVersion.scala deleted file mode 100644 index cbbe113218..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/SparkVersion.scala +++ /dev/null @@ -1,113 +0,0 @@ -/* - * 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.streampark.common.conf - -import org.apache.streampark.common.util.{CommandUtils, Logger} -import org.apache.streampark.common.util.Implicits._ - -import java.io.File -import java.util.function.Consumer -import java.util.regex.Pattern - -import scala.collection.mutable - -/** @param sparkHome actual spark home that must be a readable local path */ -class SparkVersion(val sparkHome: String) extends Serializable with Logger { - - private[this] lazy val SPARK_VER_PATTERN = Pattern.compile("^(\\d+\\.\\d+)(\\.)?.*$") - - private[this] lazy val SPARK_VERSION_PATTERN = Pattern.compile("\\s{2}version\\s(\\d+\\.\\d+\\.\\d+)") - - private[this] lazy val SPARK_SCALA_VERSION_PATTERN = Pattern.compile("Using\\sScala\\sversion\\s(\\d+\\.\\d+)") - - val (version, scalaVersion) = { - var sparkVersion: String = null - var scalaVersion: String = null - val cmd = List(s"export SPARK_HOME=$sparkHome&&$sparkHome/bin/spark-submit --version") - val buffer = new mutable.StringBuilder - - CommandUtils.execute( - sparkHome, - cmd, - new Consumer[String]() { - override def accept(out: String): Unit = { - buffer.append(out).append("\n") - val matcher = SPARK_VERSION_PATTERN.matcher(out) - if (matcher.find) { - sparkVersion = matcher.group(1) - } else { - val matcher1 = SPARK_SCALA_VERSION_PATTERN.matcher(out) - if (matcher1.find) { - scalaVersion = matcher1.group(1) - } - } - } - }) - - logInfo(buffer.toString()) - if (sparkVersion == null || scalaVersion == null) { - throw new IllegalStateException(s"[StreamPark] parse spark version failed. $buffer") - } - buffer.clear() - (sparkVersion, scalaVersion) - } - - lazy val majorVersion: String = { - if (version == null) { - null - } else { - val matcher = SPARK_VER_PATTERN.matcher(version) - matcher.matches() - matcher.group(1) - } - } - - lazy val fullVersion: String = s"${version}_$scalaVersion" - - lazy val sparkLib: File = { - require(sparkHome != null, "[StreamPark] sparkHome must not be null.") - require(new File(sparkHome).exists(), "[StreamPark] sparkHome must be exists.") - val lib = new File(s"$sparkHome/jars") - require( - lib.exists() && lib.isDirectory, - s"[StreamPark] $sparkHome/lib must be exists and must be directory.") - lib - } - - def checkVersion(throwException: Boolean = true): Boolean = { - version.split("\\.").map(_.trim.toInt) match { - case Array(v, _, _) if v == 2 || v == 3 => true - case _ => - if (throwException) { - throw new UnsupportedOperationException(s"Unsupported spark version: $version") - } else { - false - } - } - } - - override def toString: String = - s""" - |----------------------------------------- spark version ----------------------------------- - | sparkHome : $sparkHome - | sparkVersion : $version - | scalaVersion : $scalaVersion - |------------------------------------------------------------------------------------------- - |""".stripMargin - -} diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala deleted file mode 100644 index cc9161f5e5..0000000000 --- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala +++ /dev/null @@ -1,123 +0,0 @@ -/* - * 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.streampark.common.conf - -import org.apache.streampark.common.enums.StorageType -import org.apache.streampark.common.util.{HdfsUtils, SystemPropertyUtils} -import org.apache.streampark.common.util.Implicits._ - -import java.net.URI - -object Workspace { - - def of(storageType: StorageType): Workspace = Workspace(storageType) - - lazy val local: Workspace = Workspace.of(StorageType.LFS) - - lazy val remote: Workspace = Workspace.of(StorageType.HDFS) - - private[this] lazy val localWorkspace = local.WORKSPACE - - /** local build path */ - lazy val APP_LOCAL_DIST = s"$localWorkspace/dist" - - /** dirPath of the maven local repository with built-in compilation process */ - lazy val MAVEN_LOCAL_PATH = s"$localWorkspace/mvnrepo" - - /** local sourceCode path.(for git...) */ - lazy val PROJECT_LOCAL_PATH = s"$localWorkspace/project" - - /** local log path. */ - lazy val LOG_LOCAL_PATH = s"$localWorkspace/logs" - - /** project build log path. */ - lazy val PROJECT_BUILD_LOG_PATH = s"$LOG_LOCAL_PATH/build_logs" - - /** project archives path */ - lazy val ARCHIVES_FILE_PATH = s"${remote.WORKSPACE}/historyserver/archive" - -} - -case class Workspace(storageType: StorageType) { - - private[this] def getConfigValue[T](option: InternalOption): T = { - val s = SystemPropertyUtils.get(option.key) - val v = InternalConfigHolder.get(option).asInstanceOf[T] - val d = option.defaultValue.asInstanceOf[T] - (s, v) match { - case (null, null) => d - case (null, b) => b - case (a, null) => a.cast[T](option.classType) - case (a, b) => if (b == d) a.cast[T](option.classType) else b - } - } - - lazy val WORKSPACE: String = { - storageType match { - case StorageType.LFS => - val path: String = getConfigValue[String](CommonConfig.STREAMPARK_WORKSPACE_LOCAL) - require(path != null, "[StreamPark] streampark.workspace.local must not be null") - path - case StorageType.HDFS => - val path: String = getConfigValue[String](CommonConfig.STREAMPARK_WORKSPACE_REMOTE) - path match { - case p if p.isEmpty => - s"${HdfsUtils.getDefaultFS}${CommonConfig.STREAMPARK_WORKSPACE_REMOTE.defaultValue}" - case p => - val defaultFs = HdfsUtils.getDefaultFS - if (p.startsWith("hdfs://")) { - if (p.startsWith(defaultFs)) { - p - } else { - val path = URI.create(p).getPath - s"${HdfsUtils.getDefaultFS}$path" - } - } else { - s"${HdfsUtils.getDefaultFS}$p" - } - } - } - } - - lazy val APP_PLUGINS = s"$WORKSPACE/plugins" - - lazy val APP_CLIENT = s"$WORKSPACE/client" - - /** store flink multi version support jars */ - lazy val APP_SHIMS = s"$WORKSPACE/shims" - - lazy val APP_UPLOADS = s"$WORKSPACE/uploads" - - lazy val APP_PYTHON = s"$WORKSPACE/python" - - lazy val APP_PYTHON_VENV = s"$APP_PYTHON/venv.zip" - - lazy val APP_WORKSPACE = s"$WORKSPACE/workspace" - - lazy val APP_FLINK = s"$WORKSPACE/flink" - - lazy val APP_SPARK = s"$WORKSPACE/spark" - - lazy val APP_BACKUPS = s"$WORKSPACE/backups" - - lazy val APP_SAVEPOINTS = s"$WORKSPACE/savepoints" - - /** store global public jars */ - lazy val APP_JARS = s"$WORKSPACE/jars" - -} diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java index c9addb4178..67ffd68d8e 100644 --- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java +++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java @@ -96,7 +96,7 @@ private void initConfig() { .filter(env::containsProperty) .forEach( key -> { - InternalOption config = InternalConfigHolder.getConfig(key); + InternalOption config = InternalConfigHolder.getConfig(key); AssertUtils.notNull(config); InternalConfigHolder.set(config, env.getProperty(key, config.classType())); }); From 91f2bbd2fb0f3ccff2683ee7f859899fe50fa930 Mon Sep 17 00:00:00 2001 From: och5351 Date: Sun, 19 Jul 2026 22:24:00 +0900 Subject: [PATCH 2/2] [Hotfix][service] Apply InternalOptionSpec in EnvInitializer.java --- .../apache/streampark/console/core/runner/EnvInitializer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java index 67ffd68d8e..fcdbb3716f 100644 --- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java +++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java @@ -20,7 +20,7 @@ import org.apache.streampark.common.conf.CommonConfig; import org.apache.streampark.common.conf.ConfigKeys; import org.apache.streampark.common.conf.InternalConfigHolder; -import org.apache.streampark.common.conf.InternalOption; +import org.apache.streampark.common.conf.InternalOptionSpec; import org.apache.streampark.common.conf.Workspace; import org.apache.streampark.common.enums.StorageType; import org.apache.streampark.common.fs.FsOperator; @@ -96,7 +96,7 @@ private void initConfig() { .filter(env::containsProperty) .forEach( key -> { - InternalOption config = InternalConfigHolder.getConfig(key); + InternalOptionSpec config = InternalConfigHolder.getConfig(key); AssertUtils.notNull(config); InternalConfigHolder.set(config, env.getProperty(key, config.classType())); });