feat(loader): migrate CDC to Flink 2.2 on Java 17

.github/scripts/ci-policy.py

@@ -279,14 +279,38 @@ def require_current_pr(plan, fetch):
             or live["head"]["ref"] != plan["branch"]
             or live["base"]["repo"]["full_name"] != plan["repository"]):
         raise StaleInputError("PR inputs changed; refresh the branch and start a new PR run")
-    if "stack" in plan:
-        # Native stacks can leave REST base.sha at an older snapshot. The actual
-        # branch ref, rather than that snapshot, identifies the tested base.
+    if "baseBranch" in plan:
+        # Event and REST base.sha can be older snapshots. The actual branch
+        # ref, rather than those snapshots, identifies the tested base.
         ref = fetch(f"repos/{plan['repository']}/git/ref/heads/{quote(plan['baseBranch'], safe='')}")
-        if (live.get("stack") != plan["stack"] or live["base"].get("ref") != plan["baseBranch"]
+        if (live.get("stack") != plan.get("stack") or live["base"].get("ref") != plan["baseBranch"]
                 or ref.get("object", {}).get("sha") != plan["base"]
                 or live.get("merge_commit_sha") != plan["testedMergeSHA"]):
-            raise StaleInputError("PR stack or canonical base changed; start a new PR run")
+            raise StaleInputError("PR merge or canonical base changed; start a new PR run")
+
+
+def record_current_merge(plan, event_pr, parents, fetch):
+    """Recover an older event base only from GitHub's current merge identity."""
+    try:
+        live = fetch(f"repos/{plan['repository']}/pulls/{plan['pr']}")
+        if live.get("stack") is not None:
+            record_stack_merge(plan, event_pr, parents, fetch)
+            return
+        branch = live["base"]["ref"]
+        if not branch or event_pr["base"]["ref"] != branch:
+            raise StaleInputError("PR target branch changed; start a new PR run")
+        ref = fetch(f"repos/{plan['repository']}/git/ref/heads/{quote(branch, safe='')}")
+        canonical = ref["object"]["sha"]
+        if (event_pr.get("stack") is not None
+                or ref["object"].get("type") != "commit" or not canonical
+                or live.get("merge_commit_sha") != plan["testedMergeSHA"]
+                or parents != [canonical, plan["head"]]
+                or git("merge-base", plan["base"], canonical) != plan["base"]):
+            raise StaleInputError("checkout lacks a valid current PR merge identity")
+        plan.update(base=canonical, baseBranch=branch)
+        require_current_pr(plan, fetch)
+    except (ValueError, KeyError, TypeError, AttributeError, subprocess.SubprocessError, OSError) as error:
+        raise StaleInputError("current PR merge identity could not be verified") from error
 
 
 def record_stack_merge(plan, event_pr, parents, fetch):
@@ -343,7 +367,7 @@ def create_plan(project, event, repository, fetch=api, external_inputs=None):
                 raise StaleInputError("PR event has an empty input identity")
             parents = git("show", "-s", "--format=%P", plan["testedMergeSHA"]).split()
             if parents != [plan["base"], plan["head"]]:
-                record_stack_merge(plan, pr, parents, fetch)
+                record_current_merge(plan, pr, parents, fetch)
             else:
                 require_current_pr(plan, fetch)
             # head/base objects must exist locally; workflow fetches both before planning.
@@ -424,8 +448,10 @@ def write_summary(plan, results=None):
 def result_report(plan, results):
     report = {key: plan[key] for key in ["schema", "repository", "project", "pr", "source", "branch", "base",
                                         "head", "testedMergeSHA", "externalInputs"]}
+    if "baseBranch" in plan:
+        report["baseBranch"] = plan["baseBranch"]
     if "stack" in plan:
-        report.update(baseBranch=plan["baseBranch"], stack=plan["stack"])
+        report["stack"] = plan["stack"]
     report.update(runID=int(os.environ.get("GITHUB_RUN_ID", "0")),
                   runAttempt=int(os.environ.get("GITHUB_RUN_ATTEMPT", "1")),
                   executed=plan["expected"], results=results)

.github/scripts/test_candidate_sdk.py

@@ -105,6 +105,53 @@ def run(*arguments):
                 for path, original in original_poms.items():
                     self.assertEqual(original, path.read_bytes())
 
+    @unittest.skipUnless(os.environ.get("CANDIDATE_SDK_MAVEN_TEST"),
+                         "Set CANDIDATE_SDK_MAVEN_TEST for the consumer-version regression")
+    def test_maven_consumers_resolve_the_selected_sdk_version(self):
+        # A floating SDK changes coordinates; the consumer's fixed default must
+        # not silently resolve a different release from another repository.
+        import zipfile
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            group = "org.example.candidatesdk.g" + uuid.uuid4().hex
+            repository = Path.home() / ".m2/repository" / group.replace(".", "/")
+            self.addCleanup(shutil.rmtree, repository, True)
+            artifact = repository / "sdk/1.8.0"
+            artifact.mkdir(parents=True)
+            header = '<project xmlns="http://maven.apache.org/POM/4.0.0"><modelVersion>4.0.0</modelVersion>'
+            (artifact / "sdk-1.8.0.pom").write_text(
+                header + '<groupId>' + group + '</groupId><artifactId>sdk</artifactId>'
+                '<version>1.8.0</version></project>')
+            with zipfile.ZipFile(artifact / "sdk-1.8.0.jar", "w") as jar:
+                jar.writestr("sdk-source.txt", "selected upstream revision")
+            (artifact / "_remote.repositories").write_text("sdk-1.8.0.pom>=\nsdk-1.8.0.jar>=\n")
+            pom = root / "pom.xml"
+            pom.write_text(header + '<groupId>org.example.consumer</groupId><artifactId>consumer</artifactId>'
+                           '<version>1</version><properties><hugegraph.version>1.7.0</hugegraph.version>'
+                           '</properties><dependencies><dependency><groupId>' + group + '</groupId>'
+                           '<artifactId>sdk</artifactId><version>${hugegraph.version}</version></dependency>'
+                           '</dependencies></project>')
+            settings = root / "settings.xml"
+            settings.write_text("<settings/>")
+            command = [os.environ["CANDIDATE_SDK_MAVEN_TEST"], "-o", "-ntp", "--settings", str(settings),
+                       "-f", str(pom), "org.apache.maven.plugins:maven-dependency-plugin:3.1.1:build-classpath",
+                       "-Dmdep.outputFile=" + str(root / "classpath")]
+            env = dict(os.environ, MAVEN_ARGS="-Dhugegraph.version=1.8.0")
+            # Warm only Maven's public plugin dependencies when this opt-in
+            # regression runs on a fresh machine; coordinate checks stay offline.
+            prepared = subprocess.run([arg for arg in command if arg != "-o"], env=env,
+                                      capture_output=True, text=True, timeout=90)
+            self.assertEqual(0, prepared.returncode, prepared.stdout + prepared.stderr)
+            (root / "classpath").unlink()
+            env["MAVEN_ARGS"] = ""
+            wrong = subprocess.run(command, env=env, capture_output=True, text=True, timeout=90)
+            self.assertNotEqual(0, wrong.returncode, wrong.stdout + wrong.stderr)
+            self.assertIn(group + ":sdk:jar:1.7.0", wrong.stdout + wrong.stderr)
+            env["MAVEN_ARGS"] = "-Dhugegraph.version=1.8.0"
+            selected = subprocess.run(command, env=env, capture_output=True, text=True, timeout=90)
+            self.assertEqual(0, selected.returncode, selected.stdout + selected.stderr)
+            self.assertEqual(str(artifact / "sdk-1.8.0.jar"), (root / "classpath").read_text().strip())
+
 
 class CandidateDistributionTest(unittest.TestCase):
     def setUp(self):
@@ -113,16 +160,16 @@ def setUp(self):
         context.start()
         self.addCleanup(context.stop)
 
-    def fixture(self, root):
+    def fixture(self, root, revision="1.7.0"):
         repository = root / "m2"
         artifacts, modules = [], []
         for source_pom in sorted(sdk.REQUIRED_MODULES):
             artifact = Path(source_pom).parent.name if source_pom != "pom.xml" else "hugegraph"
             packaging = "pom" if (source_pom.count("/") == 1 and artifact != "hugegraph-struct") or source_pom == "pom.xml" else "jar"
             files = []
             for extension in (["pom", "jar"] if packaging == "jar" else ["pom"]):
-                filename = artifact + "-1.7.0." + extension
-                path = repository / "org/apache/hugegraph" / artifact / "1.7.0" / filename
+                filename = artifact + "-" + revision + "." + extension
+                path = repository / "org/apache/hugegraph" / artifact / revision / filename
                 path.parent.mkdir(parents=True, exist_ok=True)
                 path.write_bytes(filename.encode())
                 with (path.parent / "_remote.repositories").open("a") as origins:
@@ -131,14 +178,56 @@ def fixture(self, root):
                 files.append(relative)
                 artifacts.append({"path": relative, "sha256": sdk.digest(path), "source_reactor_install": True})
             modules.append({"group_id": "org.apache.hugegraph", "artifact_id": artifact,
-                            "version": "1.7.0", "packaging": packaging,
+                            "version": revision, "packaging": packaging,
                             "source_pom": source_pom, "files": files})
         manifest = {"repository": sdk.REPOSITORY, "commit": self.source_commit,
-                    "source_revision": "1.7.0", "java_version": "17",
+                    "source_revision": revision, "java_version": "17",
                     "required_sdk_modules": modules, "artifacts": artifacts}
         self.write_manifest(repository, manifest)
         return repository, manifest
 
+    def test_floating_sdk_version_preserves_verified_coordinates_and_maven_arguments(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            repository, manifest = self.fixture(root, "1.8.0")
+            environment = root / "github-env"
+            # Fixed-baseline validation must still reject the newer revision.
+            with self.assertRaises(RuntimeError):
+                sdk.validate_sdk(repository)
+            original = "--settings settings.xml -Dmaven.repo.local=" + str(repository)
+            with patch.dict(os.environ, {"MAVEN_ARGS": original}):
+                sdk.configure_upstream(repository, environment)
+            lines = dict(line.split("=", 1) for line in environment.read_text().splitlines())
+            self.assertEqual("1.8.0", lines["CANDIDATE_SDK_VERSION"])
+            self.assertEqual(original + " -Dhugegraph.version=1.8.0", lines["MAVEN_ARGS"])
+            with patch.dict(os.environ, lines):
+                sdk.validate_sdk(repository)
+                distribution = self.distribution(root, repository, manifest)
+                sdk.validate_distribution(repository, distribution, "loader")
+                # Hash checks must remain enforced for the floating version.
+                jar = repository / next(a["path"] for a in manifest["artifacts"]
+                                        if a["path"].endswith(".jar"))
+                jar.write_bytes(b"changed SDK")
+                with self.assertRaises(RuntimeError):
+                    sdk.validate_sdk(repository)
+            with self.assertRaises(RuntimeError):
+                sdk.configure_upstream(repository, root / "invalid-env")
+            self.assertFalse((root / "invalid-env").exists())
+
+    def test_configuration_and_distribution_cli_options_are_separate_steps(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            repository, manifest = self.fixture(root, "1.8.0")
+            distribution = self.distribution(root, repository, manifest)
+            environment = root / "github-env"
+            result = subprocess.run(["python3", sdk.__file__, str(repository),
+                                     "--configure-upstream-env", str(environment),
+                                     "--distribution", str(distribution), "--module", "loader"],
+                                    capture_output=True, text=True, timeout=10)
+            self.assertEqual(2, result.returncode)
+            self.assertIn("cannot be combined", result.stderr)
+            self.assertFalse(environment.exists())
+
     def write_manifest(self, repository, manifest):
         (repository / "candidate-sdk-manifest.json").write_text(json.dumps(manifest))
 

.github/scripts/test_ci_policy.py

@@ -421,6 +421,89 @@ def git(*args):
             finally:
                 os.chdir(old)
 
+    def test_regular_pr_recovers_stale_event_base_with_real_git(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            def git(*args):
+                return subprocess.check_output(["git", *args], cwd=root, text=True).strip()
+            git("init", "-q")
+            git("config", "user.email", "ci@example.invalid")
+            git("config", "user.name", "CI")
+            (root / "README.md").write_text("base")
+            (root / "hugegraph-hubble").mkdir()
+            (root / "hugegraph-hubble/README.md").write_text("packaged docs")
+            git("add", ".")
+            git("commit", "-qm", "base")
+            anchor = git("rev-parse", "HEAD")
+            git("checkout", "-qb", "feature")
+            (root / "hugegraph-loader").mkdir()
+            (root / "hugegraph-loader/A.java").write_text("source")
+            git("add", ".")
+            git("commit", "-qm", "feature")
+            head = git("rev-parse", "HEAD")
+            old_merge = git("commit-tree", git("rev-parse", "HEAD^{tree}"),
+                            "-p", anchor, "-p", head, "-m", "old merge")
+            git("checkout", "-qb", "target", anchor)
+            (root / "README.md").write_text("advanced master")
+            git("commit", "-qam", "advance target")
+            canonical = git("rev-parse", "HEAD")
+            git("merge", "--no-ff", "-qm", "current merge", "feature")
+            merge = git("rev-parse", "HEAD")
+            live = self.live_pr()
+            live["head"]["sha"] = head
+            live["base"].update(sha=anchor, ref="target")
+            event = {"pull_request": json.loads(json.dumps(dict(live, number=7)))}
+            live.update(merge_commit_sha=merge)
+            ref = {"object": {"type": "commit", "sha": canonical}}
+            def fetch(path):
+                if path == "repos/apache/t/pulls/7":
+                    return live
+                self.assertEqual("repos/apache/t/git/ref/heads/target", path)
+                return ref
+            old = os.getcwd()
+            try:
+                os.chdir(root)
+                # Neither event nor REST snapshot has stack metadata. The real
+                # checkout merges the current canonical target with the same head.
+                plan = policy.create_plan("toolchain", event, "apache/t", fetch)
+                self.assertEqual(canonical, plan["base"])
+                self.assertEqual("target", plan["baseBranch"])
+                self.assertNotIn("stack", plan)
+                self.assertEqual(["hubble", "loader"], plan["selected"])
+                results = {suite: {"result": "success"} for suite in plan["expected"]}
+                results.update(plan={"result": "success"}, fixture={"result": "success"},
+                               **{"hubble-fixture": {"result": "success"}})
+                report = policy.gate(plan, results, fetch)
+                self.assertEqual(canonical, report["base"])
+                self.assertEqual("target", report["baseBranch"])
+                for key, value in [("merge_commit_sha", old_merge),
+                                   ("state", "closed"),
+                                   ("head", dict(live["head"], sha=anchor)),
+                                   ("base", dict(live["base"], ref="other"))]:
+                    original = live[key]
+                    live[key] = value
+                    with self.subTest(field=key), self.assertRaises(policy.StaleInputError):
+                        policy.create_plan("toolchain", event, "apache/t", fetch)
+                    with self.subTest(gate=key), self.assertRaises(policy.StaleInputError):
+                        policy.gate(plan, results, fetch)
+                    live[key] = original
+                ref["object"]["sha"] = anchor
+                with self.assertRaises(policy.StaleInputError):
+                    policy.create_plan("toolchain", event, "apache/t", fetch)
+                with self.assertRaises(policy.StaleInputError):
+                    policy.gate(plan, results, fetch)
+                ref["object"]["sha"] = canonical
+                # A rerun of an old checkout cannot claim the fresh merge.
+                git("checkout", "--detach", "-q", old_merge)
+                with self.assertRaises(policy.StaleInputError):
+                    policy.gate(plan, results, lambda _: dict(live, head=dict(live["head"], sha=anchor)))
+                stale_event = json.loads(json.dumps(event))
+                stale_event["pull_request"]["base"]["sha"] = canonical
+                with self.assertRaises(policy.StaleInputError):
+                    policy.create_plan("toolchain", stale_event, "apache/t", fetch)
+            finally:
+                os.chdir(old)
+
     def test_native_stack_canonical_base_and_gate_freshness(self):
         # Reproduce GitHub native stacks: REST base.sha can remain at the old
         # head, while the merge's first parent wraps the actual branch head.