Reject mixed Flink payloads before image daemons start

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"]

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 "$@"

docs/deployment.md

@@ -10,14 +10,30 @@ 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
 by the outstanding validation in [#182](https://github.com/datafusion-contrib/StreamFusion/issues/182).
 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:

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;
+    }
+  }
+}

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<String, String> 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<String> 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<Path> installed) throw
       return;
     }
     try (Stream<Path> 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);

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));