From 2c0c084cb34383d10268af92c29e9e7d3459e16a Mon Sep 17 00:00:00 2001 From: Jordan Epstein Date: Sat, 19 Sep 2026 04:31:24 -0400 Subject: [PATCH] Reject mixed Flink payloads before image daemons start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Validate the host ABI, embedded core and every installed StreamFusion payload before delegating daemon startup to Flink’s original entrypoint. Reuse the loader’s identity checks without creating the planner or loading native libraries, and name the target line explicitly in host-version errors. Real container tests modify only identity metadata in the packaged loader, core and a renamed extension, and require nonzero exit before the JobManager launches. Matching payloads must reach the JobManager REST endpoint. Both 1.18 and 2.2 pass all four startup cases; all eleven packaged-loader tests pass on each line. Strict documentation builds successfully. Tests used Docker; Podman and cross-line savepoint upgrades remain unverified. Refs #188. --- docker/flink-base.Dockerfile | 4 + docker/streamfusion-entrypoint.sh | 14 ++ docs/deployment.md | 18 ++- .../imageit/MixedPayloadImageIT.java | 149 ++++++++++++++++++ .../table/planner/loader/PlannerModule.java | 24 ++- .../loader/StreamFusionPlannerLoaderTest.java | 15 +- 6 files changed, 216 insertions(+), 8 deletions(-) create mode 100644 docker/streamfusion-entrypoint.sh create mode 100644 streamfusion-image-it/src/test/java/tech/streamfusion/imageit/MixedPayloadImageIT.java diff --git a/docker/flink-base.Dockerfile b/docker/flink-base.Dockerfile index 8a67dc59..8823f7f0 100644 --- a/docker/flink-base.Dockerfile +++ b/docker/flink-base.Dockerfile @@ -27,3 +27,7 @@ COPY streamfusion-loader/target/streamfusion-loader${STREAMFUSION_ARTIFACT_SUFFI /opt/flink/lib/00-streamfusion-loader.jar COPY streamfusion-core/target/streamfusion-core${STREAMFUSION_ARTIFACT_SUFFIX}-${STREAMFUSION_VERSION}-runtime.jar \ /opt/flink/lib/streamfusion-core.jar + +COPY --chmod=755 docker/streamfusion-entrypoint.sh /streamfusion-entrypoint.sh +ENTRYPOINT ["/streamfusion-entrypoint.sh"] +CMD ["help"] diff --git a/docker/streamfusion-entrypoint.sh b/docker/streamfusion-entrypoint.sh new file mode 100644 index 00000000..db827b5f --- /dev/null +++ b/docker/streamfusion-entrypoint.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env sh + +set -eu + +case "${1:-}" in + jobmanager|taskmanager|standalone-job|history-server) + flink_lib="${FLINK_HOME:-/opt/flink}/lib" + "${JAVA_HOME}/bin/java" \ + -cp "$flink_lib/00-streamfusion-loader.jar:$flink_lib/*" \ + org.apache.flink.table.planner.loader.PlannerModule + ;; +esac + +exec /docker-entrypoint.sh "$@" diff --git a/docs/deployment.md b/docs/deployment.md index 35fb17c0..07da3e45 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -10,7 +10,7 @@ which installing them into `lib` on one shared image already ensures. Builds record the target Flink line and module identity in every payload's manifest. The loader checks its embedded core and installed StreamFusion JARs before creating the planner classloader, -including renamed extensions. A different line or a missing marker fails startup with an explicit +including renamed extensions. A different line or a missing marker fails loader initialization with an explicit compatibility error; rebuild or upgrade the loader, core and extensions together. The artifact coordinates for 2.2 remain unchanged. The `flink-1.18` development profile produces separate `streamfusion-*-flink1.18` coordinates and admits only Flink 1.18.1; release support remains gated @@ -18,6 +18,22 @@ by the outstanding validation in [#182](https://github.com/datafusion-contrib/St See [Flink line compatibility](flink-compatibility.md) for build commands, dependency selections and known host differences. Builds and deployments require Java 17. +Images built by `bin/build-flink-image.sh` perform the same checks before starting a JobManager, +TaskManager, standalone application or history server. The entrypoint checks the host ABI, the +loader's embedded core, and installed payload identities before handing control to Flink's +original entrypoint. An incompatible image exits with a message naming the conflicting lines; +it does not wait for a SQL query. The normal Flink configuration and command handling remain +owned by the original entrypoint. Bare-metal installations and custom images retain the checks +at loader initialization. + +The image suite injects conflicting identities into packaged loader, core and renamed extension +JARs on both supported build lines, and verifies nonzero exit before the JobManager starts. +These are startup-admission checks, not cross-version state recovery tests. Cross-line savepoint +upgrade and downgrade validation remain pending in +[#188](https://github.com/datafusion-contrib/StreamFusion/issues/188); no upgrade direction is +announced as supported yet. Testcontainers selects the container runtime from its normal +configuration, including a configured Podman endpoint, without a hard-coded socket path. + Release artifacts are available from Maven Central and already contain the optimized native libraries. Fetch the loader and the separate runtime-visible core payload directly into a Flink distribution; installing StreamFusion does not require a source checkout, Rust, or a local build: diff --git a/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/MixedPayloadImageIT.java b/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/MixedPayloadImageIT.java new file mode 100644 index 00000000..2c310508 --- /dev/null +++ b/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/MixedPayloadImageIT.java @@ -0,0 +1,149 @@ +package tech.streamfusion.imageit; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import com.github.dockerjava.api.DockerClient; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.Properties; +import java.util.jar.JarEntry; +import java.util.jar.JarFile; +import java.util.jar.JarOutputStream; +import java.util.jar.Manifest; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.startupcheck.StartupCheckStrategy; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.MountableFile; + +/** Exercises startup rejection using real packaged JARs with a mismatched line identity. */ +class MixedPayloadImageIT { + @TempDir Path directory; + + @Test + void matchingPayloadsStartTheJobManagerAfterValidation() { + assumeTrue( + DockerClientFactory.instance().isDockerAvailable(), + "No Docker-compatible container runtime is available (Docker or configured Podman)"); + DockerImageName image = DockerImageName.parse(System.getProperty("streamfusion.image.name")); + try (GenericContainer container = + new GenericContainer<>(image) + .withEnv("FLINK_PROPERTIES", "jobmanager.memory.process.size: 1024m") + .withCommand("jobmanager") + .withExposedPorts(8081) + .waitingFor( + Wait.forHttp("/overview") + .forPort(8081) + .withStartupTimeout(Duration.ofMinutes(1)))) { + container.start(); + + assertTrue( + container + .getLogs() + .contains( + "StreamFusion payloads verified for Flink " + + System.getProperty("streamfusion.flink.line")), + container.getLogs()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"loader", "core", "json"}) + void refusesMixedPayloadsBeforeStartingTheJobManager(String module) throws Exception { + assumeTrue( + DockerClientFactory.instance().isDockerAvailable(), + "No Docker-compatible container runtime is available (Docker or configured Podman)"); + String line = System.getProperty("streamfusion.flink.line"); + String otherLine = line.equals("1.18") ? "2.2" : "1.18"; + Path mismatched = mismatchedPayload(module, otherLine); + String destination = + switch (module) { + case "loader" -> "00-streamfusion-loader.jar"; + case "core" -> "streamfusion-core.jar"; + default -> "renamed-extension.jar"; + }; + DockerImageName image = DockerImageName.parse(System.getProperty("streamfusion.image.name")); + try (GenericContainer container = + new GenericContainer<>(image) + .withCopyFileToContainer( + MountableFile.forHostPath(mismatched), "/opt/flink/lib/" + destination) + .withCommand("jobmanager") + .withStartupCheckStrategy( + new ExitedContainerCheck().withTimeout(Duration.ofSeconds(30)))) { + container.start(); + + String logs = container.getLogs(); + var state = + container + .getDockerClient() + .inspectContainerCmd(container.getContainerId()) + .exec() + .getState(); + assertNotEquals(0L, state.getExitCodeLong(), logs); + assertTrue(logs.contains("StreamFusion"), logs); + assertTrue(logs.contains("Flink " + line), logs); + assertTrue(logs.contains("Flink " + otherLine), logs); + assertFalse(logs.contains("Starting Job Manager"), logs); + } + } + + private Path mismatchedPayload(String module, String otherLine) throws Exception { + Path root = Path.of(System.getProperty("streamfusion.project.dir")); + String suffix = System.getProperty("streamfusion.artifact.suffix", ""); + String version = System.getProperty("streamfusion.version"); + String classifier = module.equals("core") ? "-runtime" : ""; + Path original = + root.resolve("streamfusion-" + module) + .resolve("target") + .resolve("streamfusion-" + module + suffix + "-" + version + classifier + ".jar"); + assertTrue(Files.isRegularFile(original), "Missing packaged artifact: " + original); + Path modified = directory.resolve(module + ".jar"); + try (JarFile source = new JarFile(original.toFile())) { + Manifest manifest = new Manifest(source.getManifest()); + manifest.getMainAttributes().putValue("StreamFusion-Flink-Line", otherLine); + manifest + .getMainAttributes() + .putValue( + "StreamFusion-Module", + "streamfusion-" + module + (otherLine.equals("1.18") ? "-flink1.18" : "")); + try (var output = new JarOutputStream(Files.newOutputStream(modified), manifest)) { + var entries = source.entries(); + while (entries.hasMoreElements()) { + JarEntry entry = entries.nextElement(); + if (entry.getName().equalsIgnoreCase(JarFile.MANIFEST_NAME)) continue; + output.putNextEntry(new JarEntry(entry.getName())); + try (var input = source.getInputStream(entry)) { + if (entry.getName().endsWith("/streamfusion-loader.properties")) { + Properties properties = new Properties(); + properties.load(input); + properties.setProperty("flink.line", otherLine); + properties.store(output, null); + } else { + input.transferTo(output); + } + } + output.closeEntry(); + } + } + } + return modified; + } + + private static final class ExitedContainerCheck extends StartupCheckStrategy { + @Override + public StartupStatus checkStartupState(DockerClient client, String containerId) { + return "exited".equals(getCurrentState(client, containerId).getStatus()) + ? StartupStatus.SUCCESSFUL + : StartupStatus.NOT_YET_KNOWN; + } + } +} diff --git a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java index 973621f4..b5079913 100644 --- a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java +++ b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java @@ -84,9 +84,7 @@ public class PlannerModule { "org.apache.hadoop")) .toArray(String[]::new); - private static final String[] COMPONENT_CLASSPATH = { - "org.apache.flink", "tech.streamfusion" - }; + private static final String[] COMPONENT_CLASSPATH = {"org.apache.flink", "tech.streamfusion"}; private static final Map KNOWN_MODULE_ASSOCIATIONS = new HashMap<>(); @@ -165,6 +163,20 @@ public static PlannerModule getInstance() { return PlannerComponentsHolder.INSTANCE; } + /** Validates an installed image before its entrypoint starts any Flink daemon. */ + public static void main(String[] args) throws IOException { + verifyFlinkVersion(); + String line = FlinkPayloadIdentity.loaderLine(); + URL core = PlannerModule.class.getClassLoader().getResource(STREAMFUSION_PLANNER_JAR); + if (core == null) { + throw new TableException( + "Could not find planner resource '" + STREAMFUSION_PLANNER_JAR + "'."); + } + FlinkPayloadIdentity.verify(core, FlinkPayloadIdentity.attributes(core), line); + extensionJars(line); + System.out.println("StreamFusion payloads verified for Flink " + line); + } + private static void verifyFlinkVersion() throws IOException { String line = FlinkPayloadIdentity.loaderLine(); Set supportedVersions = @@ -178,8 +190,9 @@ private static void verifyFlinkVersion() throws IOException { if (version == null || !supportedVersions.contains(version)) { throw new TableException( String.format( - "StreamFusion's planner loader supports exactly Flink %s, but found %s." + "StreamFusion loader targets Flink %s (supported versions %s), but found %s." + " Refusing to cross an unverified planner ABI boundary.", + line, supportedVersions, version == null ? "an unversioned Flink API" : "Flink " + version)); } @@ -246,8 +259,7 @@ private static void collectExtensions(Path directory, Set installed) throw return; } try (Stream jars = Files.list(directory)) { - jars - .filter(Files::isRegularFile) + jars.filter(Files::isRegularFile) .filter(path -> path.getFileName().toString().endsWith(".jar")) .map(path -> path.toAbsolutePath().normalize()) .forEach(installed::add); diff --git a/streamfusion-loader/src/test/java/tech/streamfusion/loader/StreamFusionPlannerLoaderTest.java b/streamfusion-loader/src/test/java/tech/streamfusion/loader/StreamFusionPlannerLoaderTest.java index 1e114af8..0ea8dee6 100644 --- a/streamfusion-loader/src/test/java/tech/streamfusion/loader/StreamFusionPlannerLoaderTest.java +++ b/streamfusion-loader/src/test/java/tech/streamfusion/loader/StreamFusionPlannerLoaderTest.java @@ -50,12 +50,18 @@ void acceptsARenamedMatchingExtensionAndRejectsAnUnmarkedLegacyExtension() throw try { System.setProperty("java.class.path", original + java.io.File.pathSeparator + renamed); + PlannerModule.main(new String[0]); var module = constructor.newInstance(); try (var classLoader = module.getSubmoduleClassLoader()) { assertTrue(List.of(classLoader.getURLs()).contains(renamed.toUri().toURL())); } System.setProperty("java.class.path", original + java.io.File.pathSeparator + legacy); + var startupFailure = + assertThrows( + org.apache.flink.table.api.TableException.class, + () -> PlannerModule.main(new String[0])); + assertTrue(startupFailure.getMessage().contains("missing marker")); var failure = assertThrows(InvocationTargetException.class, constructor::newInstance); assertTrue(failure.getCause().getMessage().contains("missing marker")); } finally { @@ -81,6 +87,12 @@ void rejectsMixedInstalledPayloadsBeforeCreatingThePlanner(String module) throws try { System.setProperty("java.class.path", original + java.io.File.pathSeparator + jar); + var startupFailure = + assertThrows( + org.apache.flink.table.api.TableException.class, + () -> PlannerModule.main(new String[0])); + assertTrue(startupFailure.getMessage().contains("loader targets Flink " + line)); + assertTrue(startupFailure.getMessage().contains("targets Flink " + otherLine)); var failure = assertThrows(InvocationTargetException.class, constructor::newInstance); assertTrue(failure.getCause().getMessage().contains("loader targets Flink " + line)); @@ -106,7 +118,8 @@ void installsTheNativePlannerStageWithoutApplicationCode() throws Exception { PlannerModule.class.getResource("/streamfusion-planner.jar"), "the loader artifact must embed the StreamFusion runtime payload"); - TableEnvironment tableEnvironment = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); + TableEnvironment tableEnvironment = + TableEnvironment.create(EnvironmentSettings.inStreamingMode()); String sql = "SELECT c0 * 2 AS doubled FROM (VALUES (3), (4), (5)) AS t(c0)"; String explain = tableEnvironment.explainSql(sql);