FEAT: support for flink 2.1.3

.github/workflows/ci.yml

@@ -101,16 +101,14 @@ jobs:
         run: |
           bin/build-release.sh --host-only
           version="$(mvn -q -DforceStdout help:evaluate -Dexpression=project.version)"
-          jar tf "streamfusion-kafka/target/streamfusion-kafka-${version}.jar" | grep -q 'native/kafka/linux/x86_64/libstreamfusion_kafka.so'
-          jar tf "streamfusion-json/target/streamfusion-json-${version}.jar" | grep -q 'native/json/linux/x86_64/libstreamfusion_json.so'
-          jar tf "streamfusion-csv/target/streamfusion-csv-${version}.jar" | grep -q 'native/csv/linux/x86_64/libstreamfusion_csv.so'
-          jar tf "streamfusion-raw/target/streamfusion-raw-${version}.jar" | grep -q 'native/raw/linux/x86_64/libstreamfusion_raw.so'
-          jar tf "streamfusion-avro/target/streamfusion-avro-${version}.jar" | grep -q 'native/avro/linux/x86_64/libstreamfusion_avro.so'
-          if jar tf "streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-${version}.jar" | grep -q 'libstreamfusion_avro'; then
+          line="$(mvn -q -DforceStdout help:evaluate -Dexpression=flink.line)"
+          for module in kafka json csv raw avro protobuf parquet; do
+            jar tf "streamfusion-${module}/target/streamfusion-${module}-flink${line}-${version}.jar" \
+              | grep -q "native/${module}/linux/x86_64/libstreamfusion_${module}.so"
+          done
+          if jar tf "streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-flink${line}-${version}.jar" | grep -q 'libstreamfusion_avro'; then
             exit 1
           fi
-          jar tf "streamfusion-protobuf/target/streamfusion-protobuf-${version}.jar" | grep -q 'native/protobuf/linux/x86_64/libstreamfusion_protobuf.so'
-          jar tf "streamfusion-parquet/target/streamfusion-parquet-${version}.jar" | grep -q 'native/parquet/linux/x86_64/libstreamfusion_parquet.so'
           bin/check-artifacts.sh --host-only
           bin/build-flink-image.sh --tag streamfusion-flink:image-it --load --skip-release-build
 

bin/build-flink-image.sh

@@ -13,6 +13,7 @@ Builds a job-neutral StreamFusion Flink base image.
   --load                  Build one platform and load it into the local Docker daemon.
   --platform <platform>   Platform for --load (default: Docker server platform).
   --flink-image <image>   Flink base image (default: flink:2.2.1-scala_2.12-java17).
+  --flink-line <line>     StreamFusion Flink line to build (default: 2.2). Must match --flink-image.
   --skip-release-build    Reuse the already-built StreamFusion JARs.
 EOF
   exit 64
@@ -21,6 +22,7 @@ EOF
 script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
 repo_root=$(cd "$script_dir/.." && pwd)
 flink_image=flink:2.2.1-scala_2.12-java17
+flink_line=2.2
 image_tag=
 mode=
 platform=
@@ -48,6 +50,11 @@ while [ "$#" -gt 0 ]; do
       flink_image=$2
       shift 2
       ;;
+    --flink-line)
+      [ "$#" -ge 2 ] || usage
+      flink_line=$2
+      shift 2
+      ;;
     --skip-release-build)
       skip_release_build=true
       shift
@@ -67,12 +74,12 @@ command -v docker >/dev/null 2>&1 || {
 docker buildx version >/dev/null
 
 if [ "$skip_release_build" = false ]; then
-  "$repo_root/bin/build-release.sh" --linux-only
+  "$repo_root/bin/build-release.sh" --linux-only --flink-line "$flink_line"
 fi
 
 artifact_version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version)
-loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-$artifact_version.jar
-core_jar=$repo_root/streamfusion-core/target/streamfusion-core-$artifact_version-runtime.jar
+loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$artifact_version.jar
+core_jar=$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$artifact_version-runtime.jar
 [ -f "$loader_jar" ] && [ -f "$core_jar" ] || {
   echo "StreamFusion release JARs are missing; run bin/build-release.sh first." >&2
   exit 66
@@ -104,6 +111,7 @@ docker buildx build \
   --platform "$platforms" \
   --build-arg "FLINK_IMAGE=$flink_image" \
   --build-arg "STREAMFUSION_VERSION=$artifact_version" \
+  --build-arg "STREAMFUSION_FLINK_LINE=$flink_line" \
   --tag "$image_tag" \
   --file "$repo_root/docker/flink-base.Dockerfile" \
   "$output" \

bin/build-release.sh

@@ -2,24 +2,44 @@
 
 set -eu
 
-if [ "$#" -gt 1 ] || { [ "$#" -eq 1 ] && [ "$1" != "--host-only" ] && [ "$1" != "--linux-only" ]; }; then
-  echo "usage: $0 [--host-only | --linux-only]" >&2
+usage() {
+  echo "usage: $0 [--host-only | --linux-only] [--flink-line <line>]" >&2
   exit 64
-fi
+}
 
 script_dir=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd)
 repo_root=$(cd "$script_dir/.." && pwd)
 native_dir=$repo_root/native
 stage_dir=$native_dir/target/universal
 host_only=false
 linux_only=false
+# Deployable coordinates carry the Flink line, so a release targets one line at a time.
+flink_line=2.2
+
+while [ "$#" -gt 0 ]; do
+  case $1 in
+    --host-only) host_only=true ;;
+    --linux-only) linux_only=true ;;
+    --flink-line)
+      [ "$#" -ge 2 ] || usage
+      flink_line=$2
+      shift
+      ;;
+    *) usage ;;
+  esac
+  shift
+done
 
-if [ "$#" -eq 1 ] && [ "$1" = "--host-only" ]; then
-  host_only=true
-elif [ "$#" -eq 1 ]; then
-  linux_only=true
+if [ "$host_only" = true ] && [ "$linux_only" = true ]; then
+  usage
 fi
 
+case $flink_line in
+  2.2) flink_profile= ;;
+  2.1) flink_profile=,flink-2.1 ;;
+  *) echo "unsupported Flink line: $flink_line" >&2; exit 64 ;;
+esac
+
 host_platform() {
   case "$(uname -s)" in
     Linux) printf '%s\n' linux ;;
@@ -176,4 +196,4 @@ fi
 # platform build. A release always starts from empty Java output directories. The release profile
 # builds the same source and javadoc attachments as the publish workflow, unsigned, so attachment
 # failures surface here instead of on the release runner.
-(cd "$repo_root" && mvn clean package -Pdist,universal,release -Dgpg.skip=true -DskipTests)
+(cd "$repo_root" && mvn clean package -Pdist,universal,release${flink_profile} -Dgpg.skip=true -DskipTests)

bin/check-artifacts.sh

@@ -25,6 +25,8 @@ fi
 script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
 repo_root=$(cd "$script_dir/.." && pwd)
 version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version)
+# Module directories are line-neutral; the artifacts they build are not.
+flink_line=${FLINK_LINE:-2.2}
 modules="core kafka json csv raw avro avro-confluent-registry protobuf parquet"
 entries=$(mktemp)
 native_entries=$(mktemp)
@@ -69,10 +71,11 @@ assert_no_native_payload() {
 
 for suffix in $modules; do
   module="streamfusion-$suffix"
+  artifact="$module-flink$flink_line"
   if [ "$suffix" = core ]; then
-    jar_file="$repo_root/$module/target/$module-$version-runtime.jar"
+    jar_file="$repo_root/$module/target/$artifact-$version-runtime.jar"
   else
-    jar_file="$repo_root/$module/target/$module-$version.jar"
+    jar_file="$repo_root/$module/target/$artifact-$version.jar"
   fi
   if [ ! -f "$jar_file" ]; then
     echo "missing artifact: $jar_file" >&2
@@ -87,8 +90,8 @@ for suffix in $modules; do
   fi
 done
 
-core_jar="$repo_root/streamfusion-core/target/streamfusion-core-$version-runtime.jar"
-core_main_jar="$repo_root/streamfusion-core/target/streamfusion-core-$version.jar"
+core_jar="$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$version-runtime.jar"
+core_main_jar="$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$version.jar"
 assert_native_payload "$core_main_jar" streamfusion-core libstreamfusion ""
 assert_native_payload "$core_jar" streamfusion-core libstreamfusion ""
 if jar tf "$core_jar" | grep -Eq '^tech/streamfusion/(kafka|parquet|format/(json|csv|raw|avro|avroconfluent|protobuf))/'; then
@@ -98,18 +101,18 @@ fi
 
 for suffix in kafka json csv raw avro protobuf parquet; do
   assert_native_payload \
-    "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-$version.jar" \
+    "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-flink$flink_line-$version.jar" \
     "streamfusion-$suffix" "libstreamfusion_$suffix" "$suffix"
 done
 
 assert_no_native_payload \
   "$repo_root/streamfusion-runtime/target/streamfusion-runtime-$version.jar" \
   streamfusion-runtime
 assert_no_native_payload \
-  "$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-$version.jar" \
+  "$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-flink$flink_line-$version.jar" \
   streamfusion-avro-confluent-registry
 
-loader_jar="$repo_root/streamfusion-loader/target/streamfusion-loader-$version.jar"
+loader_jar="$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$version.jar"
 if [ ! -f "$loader_jar" ]; then
   echo "missing artifact: $loader_jar" >&2
   exit 1
@@ -127,7 +130,7 @@ if [ -n "$duplicates" ]; then
   exit 1
 fi
 
-confluent_jar="$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-$version.jar"
+confluent_jar="$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-flink$flink_line-$version.jar"
 if jar tf "$confluent_jar" | grep -q 'libstreamfusion_avro'; then
   echo "the Confluent integration duplicates streamfusion-avro's native library" >&2
   exit 1

bin/flink-suite.sh

@@ -5,6 +5,14 @@ set -uo pipefail
 readonly REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 readonly FLINK_VERSION="${FLINK_VERSION:-2.2.1}"
 readonly FLINK_TAG="release-${FLINK_VERSION}"
+# The suite must build StreamFusion for the same Flink line it is about to run against.
+readonly FLINK_LINE="${FLINK_VERSION%.*}"
+readonly FLINK_KAFKA_CONNECTOR_VERSION="${KAFKA_CONNECTOR_VERSION:-5.0.0}-${FLINK_LINE}"
+if [[ "${FLINK_LINE}" == "2.2" ]]; then
+  readonly SF_FLINK_PROFILE_ARG=""
+else
+  readonly SF_FLINK_PROFILE_ARG="-Pflink-${FLINK_LINE}"
+fi
 readonly KAFKA_CONNECTOR_VERSION="${KAFKA_CONNECTOR_VERSION:-5.0.0}"
 readonly KAFKA_CONNECTOR_TAG="v${KAFKA_CONNECTOR_VERSION}"
 readonly SUITE_ROOT="${FLINK_SUITE_ROOT:-${REPO_ROOT}/.flink-suite}"
@@ -13,7 +21,9 @@ readonly KAFKA_CONNECTOR_ROOT="${SUITE_ROOT}/flink-connector-kafka-${KAFKA_CONNE
 readonly STREAMFUSION_BUILD_ROOT="${SUITE_ROOT}/streamfusion-source"
 readonly AGENT_ROOT="${REPO_ROOT}/dev/flink-suite/agent"
 readonly AGENT_JAR="${AGENT_ROOT}/target/streamfusion-flink-suite-agent-1.0-SNAPSHOT.jar"
-readonly CLASSPATH_FILE="${SUITE_ROOT}/streamfusion-classpath.txt"
+# Per line: the two lines resolve different Flink, Calcite and connector jars, so a shared file lets
+# a reused build run one line's tests against the other line's classpath.
+readonly CLASSPATH_FILE="${SUITE_ROOT}/streamfusion-classpath-${FLINK_VERSION}.txt"
 readonly MAVEN_SETTINGS="${REPO_ROOT}/dev/flink-suite/settings.xml"
 readonly SUITE_MAVEN_REPO="${SUITE_ROOT}/m2"
 readonly UNSHADED_PLANNER_JAR="${SUITE_ROOT}/flink-table-planner-${FLINK_VERSION}-unshaded.jar"
@@ -196,13 +206,32 @@ else
   ) || exit $?
 
   echo "Building and installing StreamFusion and its supported connector/format modules against the source-suite planner..."
+  # Flink pins Calcite per line, and the source-suite classpath must agree with it: a planner
+  # compiled against one Calcite cannot initialise its convertlet table against another.
+  readonly FLINK_TABLE_POM="${FLINK_ROOT}/flink-table/pom.xml"
+  CALCITE_VERSION="$(sed -n 's:.*<calcite\.version>\(.*\)</calcite\.version>.*:\1:p' "${FLINK_TABLE_POM}" | head -1)"
+  if [[ -z "${CALCITE_VERSION}" ]]; then
+    echo "Could not read calcite.version from ${FLINK_TABLE_POM}" >&2
+    exit 1
+  fi
+  echo "Flink ${FLINK_VERSION} pins Calcite ${CALCITE_VERSION}."
+  # Deployable coordinates carry the Flink line, so the module list is per-line too.
+  SF_MODULES=""
+  for module in core kafka json csv raw avro avro-confluent-registry protobuf parquet; do
+    SF_MODULES="${SF_MODULES}${SF_MODULES:+,}:streamfusion-${module}-flink${FLINK_LINE}"
+  done
   mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \
     -Dstreamfusion.flink-source-suite \
+    ${SF_FLINK_PROFILE_ARG} \
+    -Dcalcite.version="${CALCITE_VERSION}" \
     -f "${STREAMFUSION_BUILD_ROOT}/pom.xml" \
-    -pl :streamfusion-core,:streamfusion-kafka,:streamfusion-json,:streamfusion-csv,:streamfusion-raw,:streamfusion-avro,:streamfusion-avro-confluent-registry,:streamfusion-protobuf,:streamfusion-parquet \
+    -pl "${SF_MODULES}" \
     -am -DskipTests clean install || exit $?
   mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \
     -f "${REPO_ROOT}/dev/flink-suite/classpath-pom.xml" \
+    -Dflink.version="${FLINK_VERSION}" \
+    -Dflink.line="${FLINK_LINE}" \
+    -Dflink.connector.kafka.version="${FLINK_KAFKA_CONNECTOR_VERSION}" \
     dependency:build-classpath -Dmdep.outputFile="${CLASSPATH_FILE}" || exit $?
 
   if [[ "${SUITE_MODE}" == "formats" || "${SUITE_MODE}" == "parquet" ]]; then

bin/install-flink.sh

@@ -2,17 +2,24 @@
 
 set -eu
 
+flink_line=2.2
+if [ "${1:-}" = "--flink-line" ]; then
+  [ "$#" -ge 2 ] || { echo "usage: $0 [--flink-line <line>] <FLINK_HOME>" >&2; exit 64; }
+  flink_line=$2
+  shift 2
+fi
+
 if [ "$#" -ne 1 ]; then
-  echo "usage: $0 <FLINK_HOME>" >&2
+  echo "usage: $0 [--flink-line <line>] <FLINK_HOME>" >&2
   exit 64
 fi
 
 flink_home=$1
 script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
 repo_root=$(cd "$script_dir/.." && pwd)
 artifact_version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version)
-loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-$artifact_version.jar
-core_jar=$repo_root/streamfusion-core/target/streamfusion-core-$artifact_version-runtime.jar
+loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$artifact_version.jar
+core_jar=$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$artifact_version-runtime.jar
 
 if [ ! -d "$flink_home/lib" ]; then
   echo "Flink lib directory does not exist: $flink_home/lib" >&2

bin/package-release.sh

@@ -4,10 +4,17 @@ set -eu
 
 script_dir=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd)
 repo_root=$(cd "$script_dir/.." && pwd)
+# Deployable coordinates carry the Flink line, so a bundle holds one line's jars.
+flink_line=2.2
+if [ "${1:-}" = "--flink-line" ]; then
+  [ "$#" -ge 2 ] || { echo "usage: $0 [--flink-line <line>] [output-dir]" >&2; exit 64; }
+  flink_line=$2
+  shift 2
+fi
 version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version)
 output_dir=${1:-"$repo_root/target/release"}
 stage_dir=$(mktemp -d)
-bundle_dir=$stage_dir/streamfusion-$version
+bundle_dir=$stage_dir/streamfusion-flink$flink_line-$version
 
 cleanup() {
   rm -rf "$stage_dir"
@@ -16,15 +23,15 @@ trap cleanup EXIT HUP INT TERM
 
 mkdir -p "$bundle_dir" "$output_dir"
 cp "$repo_root/LICENSE" "$repo_root/readme.md" "$bundle_dir/"
-cp "$repo_root/streamfusion-loader/target/streamfusion-loader-$version.jar" "$bundle_dir/"
-cp "$repo_root/streamfusion-core/target/streamfusion-core-$version-runtime.jar" "$bundle_dir/"
+cp "$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$version.jar" "$bundle_dir/"
+cp "$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$version-runtime.jar" "$bundle_dir/"
 
 for suffix in kafka json csv raw avro avro-confluent-registry protobuf parquet; do
-  cp "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-$version.jar" "$bundle_dir/"
+  cp "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-flink$flink_line-$version.jar" "$bundle_dir/"
 done
 
-archive=$output_dir/streamfusion-$version-bin.tar.gz
-(cd "$stage_dir" && tar -czf "$archive" "streamfusion-$version")
+archive=$output_dir/streamfusion-flink$flink_line-$version-bin.tar.gz
+(cd "$stage_dir" && tar -czf "$archive" "streamfusion-flink$flink_line-$version")
 (cd "$output_dir" && shasum -a 256 "$(basename "$archive")" > "$(basename "$archive").sha256")
 
 printf '%s\n' "$archive"