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 5b41a55b..e7d2e876 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 65425d31..55888aac 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));