This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c3cbf4ea6ce2 feat(agent): Add Flink multi writer contract to Hudi
Architect (#20472)
c3cbf4ea6ce2 is described below
commit c3cbf4ea6ce2a99d52a8c55ea76b13b29e807dea
Author: fhan <[email protected]>
AuthorDate: Sat Oct 31 12:37:24 2026 +0800
feat(agent): Add Flink multi writer contract to Hudi Architect (#20472)
* feat(agent): Add Flink multi writer contract to Hudi Architect
* feat(agent): Harden Flink multi-writer contract evidence reconciliation
* feat(agent): Fix Mypy type conflict in Flink evidence validation
---------
Co-authored-by: fhan <[email protected]>
---
.github/workflows/agent_gateway_ci.yml | 2 +
.../TestFlinkArchitectWriterContractFixtures.java | 114 +++
.../tests/test_hudi_architect_skill.py | 791 ++++++++++++++++++++-
hudi-ai-operator/skills/hudi-architect/README.md | 16 +-
hudi-ai-operator/skills/hudi-architect/RUNBOOK.md | 14 +-
hudi-ai-operator/skills/hudi-architect/SKILL.md | 21 +-
.../skills/hudi-architect/flink_writer_contract.py | 767 ++++++++++++++++++++
.../flink-1.20-hudi-1.2.0-capabilities.md | 29 +-
.../flink-1.20-hudi-1.2.0-capabilities.toml | 110 ++-
.../references/flink-config-templates.md | 13 +-
.../references/flink-decision-overrides.md | 29 +-
.../flink-existing-table-reconciliation.md | 27 +-
.../references/flink-question-flow.md | 23 +-
.../hudi-architect/references/flink-warnings.md | 50 +-
.../references/flink-writer-contract.md | 105 +++
.../hudi-architect/validate_flink_capabilities.py | 94 ++-
.../skills/hudi-architect/validate_flink_design.py | 27 +-
.../validate_flink_existing_table.py | 245 ++++++-
18 files changed, 2419 insertions(+), 58 deletions(-)
diff --git a/.github/workflows/agent_gateway_ci.yml
b/.github/workflows/agent_gateway_ci.yml
index 1cae2b79ea2b..768c485b4922 100644
--- a/.github/workflows/agent_gateway_ci.yml
+++ b/.github/workflows/agent_gateway_ci.yml
@@ -87,12 +87,14 @@ jobs:
run: >-
ruff check src tests
../hudi-ai-operator/skills/hudi-architect/redact_sensitive_values.py
+ ../hudi-ai-operator/skills/hudi-architect/flink_writer_contract.py
../hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
../hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
../hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
- name: Type-check
run: >-
mypy src
+ ../hudi-ai-operator/skills/hudi-architect/flink_writer_contract.py
../hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
../hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
../hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
diff --git
a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectWriterContractFixtures.java
b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectWriterContractFixtures.java
new file mode 100644
index 000000000000..63d0b2817a08
--- /dev/null
+++
b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectWriterContractFixtures.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.agent.architect;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Source-backed facts used by the assessment-only PR7 Writer Contract. */
+class TestFlinkArchitectWriterContractFixtures {
+
+ @Test
+ void testPinnedMultiWriterModesRequireLazyFailedWritesPolicy() {
+
assertTrue(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.supportsMultiWriter());
+
assertTrue(WriteConcurrencyMode.NON_BLOCKING_CONCURRENCY_CONTROL.supportsMultiWriter());
+
+ HoodieCleanConfig eagerCleanConfig = HoodieCleanConfig.newBuilder()
+ .withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.EAGER)
+ .build();
+ for (WriteConcurrencyMode mode : new WriteConcurrencyMode[] {
+ WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL,
+ WriteConcurrencyMode.NON_BLOCKING_CONCURRENCY_CONTROL}) {
+ HoodieWriteConfig config = mode ==
WriteConcurrencyMode.NON_BLOCKING_CONCURRENCY_CONTROL
+ ? validNbccBuilder().withCleanConfig(eagerCleanConfig).build()
+ : HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/hudi-architect-occ")
+ .withCleanConfig(eagerCleanConfig)
+ .withWriteConcurrencyMode(mode)
+ .build();
+ assertEquals(HoodieFailedWritesCleaningPolicy.LAZY,
config.getFailedWritesCleanPolicy());
+ }
+ }
+
+ @Test
+ void testPinnedNbccRejectsCowAndNonSimpleBucketIndex() {
+ IllegalArgumentException cowFailure = assertThrows(
+ IllegalArgumentException.class,
+ () -> nbccBuilder(HoodieTableType.COPY_ON_WRITE,
HoodieIndex.IndexType.SIMPLE)
+ .build());
+ assertTrue(cowFailure.getMessage().contains(
+ "Non-blocking concurrency control requires the MOR table with simple
bucket index"));
+
+ IllegalArgumentException indexFailure = assertThrows(
+ IllegalArgumentException.class,
+ () -> nbccBuilder(HoodieTableType.MERGE_ON_READ,
HoodieIndex.IndexType.FLINK_STATE)
+ .build());
+ assertTrue(indexFailure.getMessage().contains(
+ "Non-blocking concurrency control requires the MOR table with simple
bucket index"));
+ }
+
+ @Test
+ void testPinnedWriteOperationNames() {
+ Set<String> operations = Arrays.stream(WriteOperationType.values())
+ .map(Enum::name)
+ .collect(Collectors.toSet());
+ assertTrue(operations.containsAll(Set.of(
+ "INSERT", "UPSERT", "BULK_INSERT", "DELETE", "COMPACT", "CLUSTER",
"UNKNOWN")));
+ }
+
+ private static HoodieWriteConfig.Builder validNbccBuilder() {
+ return nbccBuilder(HoodieTableType.MERGE_ON_READ,
HoodieIndex.IndexType.BUCKET);
+ }
+
+ private static HoodieWriteConfig.Builder nbccBuilder(
+ HoodieTableType tableType, HoodieIndex.IndexType indexType) {
+ TypedProperties properties = new TypedProperties();
+ properties.put(HoodieTableConfig.TYPE.key(), tableType.name());
+ properties.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "id");
+ HoodieIndexConfig.Builder indexBuilder = HoodieIndexConfig.newBuilder()
+ .fromProperties(properties)
+ .withIndexType(indexType);
+ if (indexType == HoodieIndex.IndexType.BUCKET) {
+
indexBuilder.withBucketIndexEngineType(HoodieIndex.BucketIndexEngineType.SIMPLE);
+ }
+ return HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/hudi-architect-nbcc")
+ .withIndexConfig(indexBuilder.build())
+
.withWriteConcurrencyMode(WriteConcurrencyMode.NON_BLOCKING_CONCURRENCY_CONTROL);
+ }
+}
diff --git a/hudi-agent-gateway/tests/test_hudi_architect_skill.py
b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
index 90dd4ff53b0e..3677e7aed649 100644
--- a/hudi-agent-gateway/tests/test_hudi_architect_skill.py
+++ b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
@@ -62,6 +62,9 @@ PR2_JAVA_FIXTURE = (
/ "TestFlinkArchitectSqlFixtures.java"
)
PR6_JAVA_FIXTURE =
PR2_JAVA_FIXTURE.with_name("TestFlinkArchitectCatalogFixtures.java")
+PR7_JAVA_FIXTURE = PR2_JAVA_FIXTURE.with_name(
+ "TestFlinkArchitectWriterContractFixtures.java"
+)
FLINK_REFERENCES = (
"flink-1.20-hudi-1.2.0-capabilities.md",
@@ -72,6 +75,7 @@ FLINK_REFERENCES = (
"flink-config-templates.md",
"flink-existing-table-reconciliation.md",
"flink-catalog-composition.md",
+ "flink-writer-contract.md",
)
@@ -340,6 +344,45 @@ def _compatible_existing_envelope(design: dict) -> dict:
}
+def _pr7_writer_contract(
+ concurrency_mode: str = "OPTIMISTIC_CONCURRENCY_CONTROL",
+) -> dict:
+ return {
+ "contract_schema": 1,
+ "primary_writer_id": "flink-orders-stream",
+ "additional_writers": [
+ {
+ "writer_id": "spark-orders-backfill",
+ "role": "BACKFILL",
+ "relationship_to_primary": "INDEPENDENT_CONCURRENT",
+ "engine": "SPARK",
+ "engine_version": "3.5",
+ "hudi_version": "1.2.0",
+ "operation": "BULK_INSERT",
+ }
+ ],
+ "ownership": {
+ "schema_evolution_owner_id": "flink-orders-stream",
+ "table_configuration_owner_id": "flink-orders-stream",
+ },
+ "observed_concurrency": {
+ "mode": concurrency_mode,
+ "lock_provider_class": (
+ "org.apache.hudi.client.transaction.lock."
+ "FileSystemBasedLockProvider"
+ ),
+ "lock_configuration_consistency": "CONSISTENT",
+ },
+ }
+
+
+def _pr7_design(fixture: Path, concurrency_mode: str) -> dict:
+ design = json.loads(_read(fixture))
+ design["safety"]["writer_model"] = "multi_writer"
+ design["writer_contract"] = _pr7_writer_contract(concurrency_mode)
+ return design
+
+
def test_flink_references_exist_and_are_linked_from_skill() -> None:
skill = _read(SKILL_DIR / "SKILL.md")
for filename in FLINK_REFERENCES:
@@ -380,7 +423,7 @@ def
test_flink_baseline_has_bounded_append_and_mutable_contracts() -> None:
"fixture_version": "1.20.1",
}
assert PINNED_HUDI_REVISION in capability
- assert manifest["schema_version"] == 6
+ assert manifest["schema_version"] == 7
assert manifest["fixture_artifacts"] == {
"hudi_bundle": "org.apache.hudi:hudi-flink1.20-bundle:1.2.0",
"flink_version": "1.20.1",
@@ -444,6 +487,16 @@ def
test_flink_baseline_has_bounded_append_and_mutable_contracts() -> None:
assert catalog_contract["contract_schema"] == 1
assert catalog_contract["supported_design_contract_schemas"] == [1, 2, 3]
assert catalog_contract["catalog_freshness"] == "post_commit_eventual"
+ writer_contract = manifest["multi_writer_contract"]
+ assert writer_contract["contract_schema"] == 1
+ assert writer_contract["supported_design_contract_schemas"] == [1, 2, 3]
+ assert writer_contract["assessment_only"] is True
+ assert writer_contract["executable_sql"] is False
+ assert writer_contract["capability_results"] == [
+ "NOT_APPLICABLE",
+ "UNVERIFIED",
+ "UNSUPPORTED",
+ ]
assert [profile["id"] for profile in manifest["catalog_profiles"]] == [
"direct_sql_none",
"direct_sql_hive_metastore",
@@ -471,7 +524,7 @@ def
test_flink_capability_validation_emits_pinned_evidence() -> None:
assert result.returncode == 0, result.stderr
assert json.loads(result.stdout) == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 6,
+ "capability_manifest_schema": 7,
"catalog_contract_schema": 1,
"catalog_profile": "direct_sql_none",
"design_contract_schema": 1,
@@ -485,6 +538,7 @@ def
test_flink_capability_validation_emits_pinned_evidence() -> None:
"hudi_hms_catalog_hive_metastore",
],
"supported_design_contract_schemas": [1, 2, 3],
+ "writer_contract_schema": 1,
}
@@ -575,7 +629,11 @@ def
test_flink_capability_manifest_records_executable_acceptance_contracts() ->
for node in ast.walk(ast.parse(_read(Path(__file__))))
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
- java_fixture = _read(PR2_JAVA_FIXTURE) + _read(PR6_JAVA_FIXTURE)
+ java_fixture = (
+ _read(PR2_JAVA_FIXTURE)
+ + _read(PR6_JAVA_FIXTURE)
+ + _read(PR7_JAVA_FIXTURE)
+ )
effective_settings = {
setting["applies_to"]: {
candidate["key"]: candidate["value"]
@@ -720,6 +778,14 @@ def
test_flink_capability_manifest_records_executable_acceptance_contracts() ->
"test_pr4_valid_mutable_mor_design_renders_exact_executable_sql",
"testMutableMorSinkAndCompactionPlan",
),
+ "FLINK_MULTI_WRITER_REVIEW": (
+ "test_pr7_complete_occ_contract_is_assessment_only",
+ "testPinnedMultiWriterModesRequireLazyFailedWritesPolicy",
+ ),
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED": (
+ "test_pr7_nbcc_is_unsupported_for_existing_bounded_indexes",
+ "testPinnedNbccRejectsCowAndNonSimpleBucketIndex",
+ ),
}
for check in manifest["implemented_acceptance_checks"]:
assert check["python_test"] in python_tests
@@ -831,7 +897,7 @@ def
test_pr2_valid_designs_render_exact_executable_sql(fixture_name: str) -> Non
).rstrip("\n")
assert assessment["validation_evidence"] == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 6,
+ "capability_manifest_schema": 7,
"catalog_contract_schema": 1,
"catalog_profile": "direct_sql_none",
"design_contract_schema": 1,
@@ -845,6 +911,7 @@ def
test_pr2_valid_designs_render_exact_executable_sql(fixture_name: str) -> Non
"hudi_hms_catalog_hive_metastore",
],
"supported_design_contract_schemas": [1, 2, 3],
+ "writer_contract_schema": 1,
}
@@ -861,7 +928,7 @@ def
test_pr3_valid_mutable_design_renders_exact_executable_sql() -> None:
PR3_FIXTURE_DIR / "mutable_cow.sql"
).rstrip("\n")
assert assessment["validation_evidence"]["design_contract_schema"] == 2
- assert assessment["validation_evidence"]["capability_manifest_schema"] == 6
+ assert assessment["validation_evidence"]["capability_manifest_schema"] == 7
def test_pr4_valid_mutable_mor_design_renders_exact_executable_sql() -> None:
@@ -878,7 +945,7 @@ def
test_pr4_valid_mutable_mor_design_renders_exact_executable_sql() -> None:
).rstrip("\n")
assert assessment["validation_evidence"] == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 6,
+ "capability_manifest_schema": 7,
"catalog_contract_schema": 1,
"catalog_profile": "direct_sql_none",
"design_contract_schema": 3,
@@ -892,6 +959,7 @@ def
test_pr4_valid_mutable_mor_design_renders_exact_executable_sql() -> None:
"hudi_hms_catalog_hive_metastore",
],
"supported_design_contract_schemas": [1, 2, 3],
+ "writer_contract_schema": 1,
}
@@ -1496,6 +1564,715 @@ def
test_pr6_existing_catalog_mutation_remains_unverified(tmp_path: Path) -> Non
assert "verify_hms_connectivity" in assessment["catalog_deployment_checks"]
+def test_pr7_keeps_single_writer_contract_not_applicable_and_sql_stable() ->
None:
+ result = _run_flink_design_validator(PR2_FIXTURE_DIR / "stable_key.json")
+
+ assert result.returncode == 0, result.stderr
+ assessment = json.loads(result.stdout)
+ assert assessment["writer_contract_result"] == "NOT_APPLICABLE"
+ assert assessment["multi_writer_capability"] == "NOT_APPLICABLE"
+ assert assessment["normalized_writer_contract"] == {}
+ assert assessment["writer_capability_notes"] == []
+ assert assessment["writer_deployment_checks"] == []
+ assert assessment["artifacts"]["combined_sql"] == _read_sql_golden(
+ PR2_FIXTURE_DIR / "stable_key.sql"
+ ).rstrip("\n")
+
+
+def test_pr7_requires_contract_for_confirmed_multi_writer(tmp_path: Path) ->
None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["safety"]["writer_model"] = "multi_writer"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["writer_contract_result"] == "INCOMPLETE"
+ assert assessment["multi_writer_capability"] == "UNVERIFIED"
+ assert "FLINK_MULTI_WRITER_REVIEW" in assessment["finding_codes"]
+ assert "FLINK_WRITER_CONTRACT_REQUIRED" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+def test_pr7_complete_occ_contract_is_assessment_only(tmp_path: Path) -> None:
+ design = _pr7_design(
+ PR2_FIXTURE_DIR / "stable_key.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["writer_contract_result"] == "COMPLETE"
+ assert assessment["multi_writer_capability"] == "UNVERIFIED"
+ assert assessment["normalized_writer_contract"]["primary_writer_facts"] ==
{
+ "engine": "FLINK",
+ "engine_version": "1.20.1",
+ "hudi_version": "1.2.0",
+ "operation": "INSERT",
+ }
+ assert [note["code"] for note in assessment["writer_capability_notes"]] ==
[
+ "FLINK_OCC_DEPLOYMENT_UNVERIFIED",
+ "FLINK_CROSS_ENGINE_WRITER_UNVERIFIED",
+ ]
+ assert len(assessment["writer_deployment_checks"]) == 6
+ assert assessment["finding_codes"] == ["FLINK_MULTI_WRITER_REVIEW"]
+ assert assessment["executable_eligible"] is False
+ assert "artifacts" not in assessment
+
+
+def test_pr7_additional_flink_writer_version_drift_is_unverified(
+ tmp_path: Path,
+) -> None:
+ design = _pr7_design(
+ PR2_FIXTURE_DIR / "stable_key.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ additional_writer = design["writer_contract"]["additional_writers"][0]
+ additional_writer["engine"] = "FLINK"
+ additional_writer["engine_version"] = "1.19.1"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["writer_contract_result"] == "COMPLETE"
+ assert "FLINK_VERSION_UNVERIFIED" in assessment["finding_codes"]
+ assert "FLINK_CROSS_ENGINE_WRITER_UNVERIFIED" not in {
+ note["code"] for note in assessment["writer_capability_notes"]
+ }
+ assert assessment["executable_eligible"] is False
+
+
+def test_pr7_invalid_writer_fact_messages_are_deterministic(tmp_path: Path) ->
None:
+ design = _pr7_design(
+ PR2_FIXTURE_DIR / "stable_key.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ design["writer_contract"]["additional_writers"] = [{}]
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ invalid = next(
+ finding
+ for finding in assessment["findings"]
+ if finding["code"] == "FLINK_WRITER_CONTRACT_INVALID"
+ )
+ assert invalid["message"] == (
+ "writer_contract.additional_writers[0].engine must be a non-empty
string; "
+ "writer_contract.additional_writers[0].engine_version must be a
non-empty string; "
+ "writer_contract.additional_writers[0].hudi_version must be a
non-empty string; "
+ "writer_contract.additional_writers[0].operation must be a non-empty
string; "
+ "writer_contract.additional_writers[0].relationship_to_primary must be
a non-empty "
+ "string; writer_contract.additional_writers[0].role must be a
non-empty string; "
+ "writer_contract.additional_writers[0].writer_id must be a non-empty
string"
+ )
+
+
[email protected](
+ "fixture",
+ [
+ PR2_FIXTURE_DIR / "stable_key.json",
+ PR4_FIXTURE_DIR / "mutable_mor.json",
+ ],
+)
+def test_pr7_nbcc_is_unsupported_for_existing_bounded_indexes(
+ tmp_path: Path, fixture: Path
+) -> None:
+ design = _pr7_design(fixture, "NON_BLOCKING_CONCURRENCY_CONTROL")
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "BLOCKED"
+ assert assessment["writer_contract_result"] == "COMPLETE"
+ assert assessment["multi_writer_capability"] == "UNSUPPORTED"
+ assert "FLINK_WRITER_COMBINATION_UNSUPPORTED" in
assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+ assert "artifacts" not in assessment
+
+
[email protected](
+ "contract_patch",
+ [
+ {"contract_schema": True},
+ {"additional_writers": []},
+ {"unsupported_field": "value"},
+ ],
+)
+def test_pr7_malformed_writer_contract_fails_closed(
+ tmp_path: Path, contract_patch: dict
+) -> None:
+ design = _pr7_design(
+ PR2_FIXTURE_DIR / "stable_key.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ design["writer_contract"].update(contract_patch)
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["writer_contract_result"] == "INCOMPLETE"
+ assert assessment["multi_writer_capability"] == "UNVERIFIED"
+ assert assessment["normalized_writer_contract"] == {}
+ assert "FLINK_WRITER_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+ assert "artifacts" not in assessment
+
+
+def test_pr7_unknown_writer_model_does_not_fabricate_contract(
+ tmp_path: Path,
+) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["safety"]["writer_model"] = "unknown"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["writer_contract_result"] == "NOT_APPLICABLE"
+ assert assessment["multi_writer_capability"] == "NOT_APPLICABLE"
+ assert assessment["normalized_writer_contract"] == {}
+ assert assessment["writer_deployment_checks"] == []
+ assert assessment["finding_codes"] == ["FLINK_WRITER_MODEL_UNRESOLVED"]
+ assert "artifacts" not in assessment
+
+
[email protected]("writer_model", [[], {}])
+def test_pr7_non_string_writer_model_fails_closed(
+ tmp_path: Path, writer_model: object
+) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["safety"]["writer_model"] = writer_model
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["writer_contract_result"] == "NOT_APPLICABLE"
+ assert assessment["multi_writer_capability"] == "NOT_APPLICABLE"
+ assert assessment["finding_codes"] == ["FLINK_WRITER_MODEL_UNRESOLVED"]
+ assert "artifacts" not in assessment
+
+
[email protected]("writer_model", [[], {}])
+def test_pr7_existing_table_rejects_non_string_writer_model_without_traceback(
+ tmp_path: Path, writer_model: object
+) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ envelope = _compatible_existing_envelope(design)
+ envelope["desired_contract"]["safety"]["writer_model"] = writer_model
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["writer_contract_result"] == "NOT_APPLICABLE"
+ assert "FLINK_WRITER_MODEL_UNRESOLVED" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+
+
+def test_pr7_single_writer_inventory_disagreement_stays_unverified(
+ tmp_path: Path,
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"]["active_writers"] = ["writer1", "writer2"]
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["writer_contract_result"] == "NOT_APPLICABLE"
+ assert assessment["multi_writer_capability"] == "NOT_APPLICABLE"
+ assert "FLINK_MULTI_WRITER_REVIEW" in assessment["finding_codes"]
+ assert "FLINK_EXISTING_EVIDENCE_CONFLICT" in assessment["finding_codes"]
+ comparison = next(
+ item
+ for item in assessment["comparisons"]
+ if item["field"] == "writer_inventory.active_writers"
+ )
+ assert comparison == {
+ "desired": {"maximum_count": 1},
+ "field": "writer_inventory.active_writers",
+ "observed": ["writer1", "writer2"],
+ "result": "UNVERIFIED",
+ "source_ids": ["writers-1"],
+ }
+ assert assessment["executable_eligible"] is False
+
+
+def test_pr7_separate_inventory_snapshots_do_not_fabricate_internal_conflict(
+ tmp_path: Path,
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ second_inventory = copy.deepcopy(inventory)
+ second_inventory["evidence_id"] = "writers-2"
+ second_inventory["sha256"] = "8" * 64
+ second_inventory["facts"] = {"active_writers": ["writer1", "writer2"]}
+ envelope["evidence"].append(second_inventory)
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert "FLINK_MULTI_WRITER_REVIEW" in assessment["finding_codes"]
+ assert "FLINK_EXISTING_EVIDENCE_CONFLICT" not in
assessment["finding_codes"]
+
+
[email protected](
+ ("concurrency_mode", "expected_status", "expected_capability"),
+ [
+ ("OPTIMISTIC_CONCURRENCY_CONTROL", "REVIEW_REQUIRED", "UNVERIFIED"),
+ ("NON_BLOCKING_CONCURRENCY_CONTROL", "BLOCKED", "UNSUPPORTED"),
+ ],
+)
+def test_pr7_single_writer_concurrency_evidence_prevents_compatibility(
+ tmp_path: Path,
+ concurrency_mode: str,
+ expected_status: str,
+ expected_capability: str,
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ envelope = _compatible_existing_envelope(design)
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"].update(
+ {
+ "write_concurrency_mode": concurrency_mode,
+ "lock_configuration_consistency": "CONFLICTING",
+ }
+ )
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == expected_status
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["writer_contract_result"] == "NOT_APPLICABLE"
+ assert assessment["multi_writer_capability"] == expected_capability
+ assert "FLINK_MULTI_WRITER_REVIEW" in assessment["finding_codes"]
+ assert "FLINK_EXISTING_EVIDENCE_CONFLICT" in assessment["finding_codes"]
+ if concurrency_mode == "NON_BLOCKING_CONCURRENCY_CONTROL":
+ assert "FLINK_WRITER_COMBINATION_UNSUPPORTED" in
assessment["finding_codes"]
+ writer_fields = {
+ item["field"]: item for item in assessment["comparisons"]
+ }
+ assert writer_fields["writer_config.write_concurrency_mode"]["observed"]
== (
+ concurrency_mode
+ )
+ assert writer_fields["writer_config.lock_configuration_consistency"][
+ "observed"
+ ] == "CONFLICTING"
+ assert assessment["executable_eligible"] is False
+
+
[email protected](
+ ("fact_name", "invalid_value"),
+ [
+ ("write_concurrency_mode", "MAGIC_CONCURRENCY"),
+ ("lock_configuration_consistency", "PARTIALLY_CONSISTENT"),
+ ],
+)
+def test_pr7_rejects_unbounded_concurrency_evidence_enums(
+ tmp_path: Path, fact_name: str, invalid_value: str
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ envelope = _compatible_existing_envelope(design)
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"][fact_name] = invalid_value
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "INCOMPLETE"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert "FLINK_EXISTING_EVIDENCE_INVALID" in assessment["finding_codes"]
+ assert not any(
+ item["field"] == f"writer_config.{fact_name}"
+ for item in assessment["comparisons"]
+ )
+
+
+def test_pr7_existing_table_matches_optional_writer_facts(tmp_path: Path) ->
None:
+ design = _pr7_design(
+ PR3_FIXTURE_DIR / "mutable_cow.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"].update(
+ {
+ "active_writers": ["spark-orders-backfill", "flink-orders-stream"],
+ "concurrent_writers": True,
+ "schema_evolution_owner_id": "flink-orders-stream",
+ }
+ )
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"].update(
+ {
+ "write_concurrency_mode": "OPTIMISTIC_CONCURRENCY_CONTROL",
+ "lock_provider_class": (
+ "org.apache.hudi.client.transaction.lock."
+ "FileSystemBasedLockProvider"
+ ),
+ "lock_configuration_consistency": "CONSISTENT",
+ }
+ )
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["writer_contract_result"] == "COMPLETE"
+ assert assessment["multi_writer_capability"] == "UNVERIFIED"
+ assert "FLINK_WRITER_CONTRACT_CONFLICT" not in assessment["finding_codes"]
+ writer_fields = {
+ item["field"]: item["result"]
+ for item in assessment["comparisons"]
+ if item["field"].startswith(("writer_inventory.", "writer_config."))
+ }
+ assert writer_fields["writer_inventory.active_writers"] == "MATCH"
+ assert writer_fields["writer_inventory.schema_evolution_owner_id"] ==
"MATCH"
+ assert writer_fields["writer_config.write_concurrency_mode"] == "MATCH"
+ assert writer_fields["writer_config.lock_provider_class"] == "MATCH"
+ assert writer_fields["writer_config.lock_configuration_consistency"] ==
"MATCH"
+ assert "confirm_single_writer" not in {
+ item["check"] for item in assessment["deployment_checks"]
+ }
+ assert "verify_concurrent_writer_inventory" in {
+ item["check"] for item in assessment["writer_deployment_checks"]
+ }
+
+
+def test_pr7_inactive_declared_writer_does_not_conflict_with_active_snapshot(
+ tmp_path: Path,
+) -> None:
+ design = _pr7_design(
+ PR3_FIXTURE_DIR / "mutable_cow.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"]["active_writers"] = ["flink-orders-stream"]
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["writer_contract_result"] == "COMPLETE"
+ assert "FLINK_WRITER_CONTRACT_CONFLICT" not in assessment["finding_codes"]
+ comparison = next(
+ item
+ for item in assessment["comparisons"]
+ if item["field"] == "writer_inventory.active_writers"
+ )
+ assert comparison == {
+ "desired": {
+ "allowed_writer_ids": [
+ "flink-orders-stream",
+ "spark-orders-backfill",
+ ]
+ },
+ "field": "writer_inventory.active_writers",
+ "observed": ["flink-orders-stream"],
+ "result": "MATCH",
+ "source_ids": ["writers-1"],
+ }
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+
+
+def test_pr7_undeclared_active_writer_conflicts_with_declared_topology(
+ tmp_path: Path,
+) -> None:
+ design = _pr7_design(
+ PR3_FIXTURE_DIR / "mutable_cow.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"].update(
+ {
+ "active_writers": ["flink-orders-stream", "undeclared-writer"],
+ "concurrent_writers": True,
+ }
+ )
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["writer_contract_result"] == "CONFLICT"
+ assert "FLINK_WRITER_CONTRACT_CONFLICT" in assessment["finding_codes"]
+ comparison = next(
+ item
+ for item in assessment["comparisons"]
+ if item["field"] == "writer_inventory.active_writers"
+ )
+ assert comparison["result"] == "UNVERIFIED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+
+
[email protected](
+ ("fact_name", "fact_value", "expected_contract_result",
"expected_capability"),
+ [
+ (
+ "write_concurrency_mode",
+ "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "COMPLETE",
+ "UNSUPPORTED",
+ ),
+ (
+ "lock_configuration_consistency",
+ "CONFLICTING",
+ "CONFLICT",
+ "UNVERIFIED",
+ ),
+ ],
+)
+def test_pr7_assesses_concurrency_evidence_without_optional_contract_block(
+ tmp_path: Path,
+ fact_name: str,
+ fact_value: str,
+ expected_contract_result: str,
+ expected_capability: str,
+) -> None:
+ design = _pr7_design(
+ PR3_FIXTURE_DIR / "mutable_cow.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ del design["writer_contract"]["observed_concurrency"]
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"]["concurrent_writers"] = True
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"][fact_name] = fact_value
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["writer_contract_result"] == expected_contract_result
+ assert assessment["multi_writer_capability"] == expected_capability
+ expected_finding = (
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED"
+ if fact_name == "write_concurrency_mode"
+ else "FLINK_WRITER_CONTRACT_CONFLICT"
+ )
+ assert expected_finding in assessment["finding_codes"]
+ assert "observed_concurrency" not in
assessment["normalized_writer_contract"]
+ comparison = next(
+ item
+ for item in assessment["comparisons"]
+ if item["field"] == f"writer_config.{fact_name}"
+ )
+ assert comparison == {
+ "desired": None,
+ "field": f"writer_config.{fact_name}",
+ "observed": fact_value,
+ "result": "UNVERIFIED",
+ "source_ids": ["writer-config-1"],
+ }
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["executable_eligible"] is False
+
+
[email protected]("contract_state", ["missing", "invalid"])
+def test_pr7_assesses_nbcc_evidence_even_when_contract_is_incomplete(
+ tmp_path: Path, contract_state: str
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["safety"]["writer_model"] = "multi_writer"
+ if contract_state == "invalid":
+ design["writer_contract"] = []
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"]["concurrent_writers"] = True
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"]["write_concurrency_mode"] = (
+ "NON_BLOCKING_CONCURRENCY_CONTROL"
+ )
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "BLOCKED"
+ assert assessment["writer_contract_result"] == "INCOMPLETE"
+ assert assessment["multi_writer_capability"] == "UNSUPPORTED"
+ expected_contract_finding = (
+ "FLINK_WRITER_CONTRACT_REQUIRED"
+ if contract_state == "missing"
+ else "FLINK_WRITER_CONTRACT_INVALID"
+ )
+ assert expected_contract_finding in assessment["finding_codes"]
+ assert "FLINK_WRITER_COMBINATION_UNSUPPORTED" in
assessment["finding_codes"]
+ assert assessment["normalized_writer_contract"] == {}
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["executable_eligible"] is False
+
+
+def test_pr7_conflicting_lock_evidence_does_not_reclassify_missing_contract(
+ tmp_path: Path,
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["safety"]["writer_model"] = "multi_writer"
+ envelope = _compatible_existing_envelope(design)
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"]["lock_configuration_consistency"] = "CONFLICTING"
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["writer_contract_result"] == "INCOMPLETE"
+ assert assessment["multi_writer_capability"] == "UNVERIFIED"
+ assert "FLINK_WRITER_CONTRACT_REQUIRED" in assessment["finding_codes"]
+ assert "FLINK_EXISTING_EVIDENCE_CONFLICT" in assessment["finding_codes"]
+ assert "FLINK_WRITER_CONTRACT_CONFLICT" not in assessment["finding_codes"]
+
+
+def test_pr7_evidence_nbcc_overrides_explicit_occ_capability_conservatively(
+ tmp_path: Path,
+) -> None:
+ design = _pr7_design(
+ PR3_FIXTURE_DIR / "mutable_cow.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"]["concurrent_writers"] = True
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"]["write_concurrency_mode"] = (
+ "NON_BLOCKING_CONCURRENCY_CONTROL"
+ )
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["writer_contract_result"] == "CONFLICT"
+ assert assessment["multi_writer_capability"] == "UNSUPPORTED"
+ assert "FLINK_WRITER_CONTRACT_CONFLICT" in assessment["finding_codes"]
+ assert "FLINK_WRITER_COMBINATION_UNSUPPORTED" in
assessment["finding_codes"]
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["executable_eligible"] is False
+
+
+def test_pr7_existing_writer_evidence_conflict_stays_unverified(
+ tmp_path: Path,
+) -> None:
+ design = _pr7_design(
+ PR3_FIXTURE_DIR / "mutable_cow.json",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ )
+ envelope = _compatible_existing_envelope(design)
+ inventory = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"writer_inventory"
+ )
+ inventory["facts"].update(
+ {
+ "active_writers": ["flink-orders-stream", "spark-orders-backfill"],
+ "concurrent_writers": True,
+ "schema_evolution_owner_id": "flink-orders-stream",
+ }
+ )
+ writer_config = next(
+ item for item in envelope["evidence"] if item["kind"] ==
"flink_writer_config"
+ )
+ writer_config["facts"].update(
+ {
+ "write_concurrency_mode": "OPTIMISTIC_CONCURRENCY_CONTROL",
+ "lock_provider_class": (
+ "org.apache.hudi.client.transaction.lock."
+ "FileSystemBasedLockProvider"
+ ),
+ "lock_configuration_consistency": "CONSISTENT",
+ }
+ )
+ conflicting_config = copy.deepcopy(writer_config)
+ conflicting_config["evidence_id"] = "writer-config-2"
+ conflicting_config["sha256"] = "7" * 64
+ conflicting_config["facts"]["lock_provider_class"] = (
+ "org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider"
+ )
+ envelope["evidence"].append(conflicting_config)
+
+ result = _run_flink_existing_table_validator(_write_design(tmp_path,
envelope))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert assessment["writer_contract_result"] == "CONFLICT"
+ assert "FLINK_EXISTING_EVIDENCE_CONFLICT" in assessment["finding_codes"]
+ assert "FLINK_WRITER_CONTRACT_CONFLICT" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+
+
@pytest.mark.parametrize(
("field", "value"),
[
@@ -3130,7 +3907,7 @@ def test_safety_gate_order_is_stable() -> None:
"## F9 — Streaming checkpoint contract",
"## F10 — Mutable table type, query freshness, and MOR ownership",
"## F11 — Existing-table evidence and writer transition",
- "## PR2/PR3/PR4/PR5/PR6 validation and completion",
+ "## PR2/PR3/PR4/PR5/PR6/PR7 validation and completion",
]
offsets = [flow.index(heading) for heading in headings]
assert offsets == sorted(offsets)
diff --git a/hudi-ai-operator/skills/hudi-architect/README.md
b/hudi-ai-operator/skills/hudi-architect/README.md
index fbd437b75272..50edb8333403 100644
--- a/hudi-ai-operator/skills/hudi-architect/README.md
+++ b/hudi-ai-operator/skills/hudi-architect/README.md
@@ -44,7 +44,9 @@ Target Hudi version: **1.2.0**.
an existing table backed by structured evidence. For new tables, those same
design schemas can
compose direct Hive sync, Hudi DFS Catalog, or external Hudi HMS Catalog
through five pinned
profiles. Glue, managed HMS, existing-table catalog mutation, and live
connectivity claims
- remain fail-closed.
+ remain fail-closed. A versioned PR7 Writer Contract can record independent
writers and ownership,
+ but remains assessment-only: OCC is unverified, current schemas are
incompatible with NBCC, and
+ no multi-writer SQL is emitted.
## How to invoke
@@ -93,6 +95,8 @@ Even without invoking as a Skill, the files in `references/`
are readable design
rules for the assessment-only existing-table path.
- `flink-catalog-composition.md` — the five PR6 profiles, lifecycle/path
rules, Hive sync ownership,
MOR naming, freshness boundary, and deployment checks.
+- `flink-writer-contract.md` — the PR7 writer inventory, ownership,
observed-concurrency, evidence,
+ and non-executable capability rules.
The Skill itself is defined in `SKILL.md`.
@@ -136,6 +140,8 @@ The mutable MOR schema-3 fixture is
`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.json`.
PR6 profile inputs are recorded in
`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr6/catalog_profiles.toml`.
+PR7 source-backed concurrency fixtures are implemented in
+`hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectWriterContractFixtures.java`.
Validate a structured existing-table reconciliation contract without rendering
SQL:
@@ -160,18 +166,20 @@ This is **Milestone 1 of a longer arc** — meant to be
shareable and playable,
4. **Read the ADR output.** Are the revisit conditions actually measurable? Do
the durability tables cover the one-way decisions relevant to your workload?
5. **Declare a second writer.** On the Spark route, check the selected lock
provider, emitted block,
and pre-launch checklist. On the Flink route, check that the result is
`REVIEW_REQUIRED` and
- that no concurrency-sensitive configuration is emitted.
+ that the Writer Contract is normalized without emitting
concurrency-sensitive configuration.
## What's out of scope in Milestone 1
- Flink append-only MOR, standalone compaction, detailed compaction tuning,
clustering,
existing-table SQL generation or catalog takeover, AWS Glue,
managed/embedded HMS, live catalog
- probing, multiple writers, alternative indexes, custom
+ probing, executable multi-writer operation, alternative indexes, custom
mergers, retract normalization, partition movement, key-only deletes, full
CDC history,
DataStream API, `HoodieFlinkStreamer`, source connector generation, and
submit/deploy commands.
Those paths do not reuse the bounded SQL output as if it were generally
applicable.
-- Multi-writer contention tuning — the mode, lock provider, and config bundle
are derived, but retry/timeout tuning and early conflict detection stay at
defaults (they depend on observed behavior, not design-time facts).
+- Multi-writer contention tuning — the Spark route may derive its established
bundle; the Flink
+ Writer Contract records only observed mode/provider facts. Retry/timeout
tuning and early
+ conflict detection depend on observed behavior, not design-time facts.
- Benchmarking / scale-characterization — different flow shape, future
revision.
- CONSISTENT_HASHING bucket recommendations at design time.
- Partial-update MERGE nudging (Spark SQL only).
diff --git a/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
b/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
index 9647fc762956..93a589c6f11d 100644
--- a/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
+++ b/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
@@ -29,7 +29,9 @@ bounded append-only COW, mutable COW, and mutable MOR
new-table, single-writer,
streaming paths. New-table designs can optionally compose direct Hive sync,
Hudi DFS Catalog, or
external Hudi HMS Catalog through five pinned profiles. For an existing table,
it can instead
produce a read-only exact-match reconciliation report from provenance-bearing
table, schema,
-writer, and timeline evidence; that report never contains SQL or mutation
commands.
+writer, and timeline evidence; that report never contains SQL or mutation
commands. For a
+confirmed independent writer, PR7 records a non-executable Writer Contract and
deployment checks
+without selecting a concurrency mode, lock provider, or index.
---
@@ -155,6 +157,8 @@ explicit-column `INSERT INTO`. The assessment retains
effective catalog configur
deployment-check codes separately; generated SQL is never executed and
deployment remains manual.
An existing-table session emits comparisons, evidence provenance, findings,
and deployment rechecks
with executable eligibility `false`, including when the static result is
`COMPATIBLE`.
+A multi-writer session additionally reports Writer Contract completeness and
an `UNVERIFIED` or
+`UNSUPPORTED` capability. It never reports multi-writer support or emits SQL.
Then: land a first commit in a staging path, run your real read patterns
against it, and check the ADR's operational playbook section for what to
monitor from day one (commit duration, pending compactions, active timeline
size, small-file ratio).
@@ -166,12 +170,14 @@ Be aware of what the agent will *decline* to decide — it
defers honestly rathe
new-table, single-writer streaming SQL sinks are implemented, with five
direct SQL / DFS Catalog
/ HMS Catalog compositions. Append-only MOR, standalone compaction, detailed
compaction tuning,
clustering, existing-table SQL or takeover, existing-table catalog mutation,
AWS Glue, managed
- or embedded HMS, live catalog probing, multiple writers, alternative
indexes, custom mergers,
+ or embedded HMS, live catalog probing, executable multi-writer operation,
alternative indexes,
+ custom mergers,
retract normalization, partition movement, key-only deletes, full CDC
history, DataStream API,
`HoodieFlinkStreamer`, source connector generation, and submit/deploy
commands remain out of
- scope.
+ scope. PR7 can inventory multiple writers, but OCC remains unverified and
NBCC is incompatible
+ with the current COW and FLINK_STATE-indexed MOR contracts.
-- **Multi-writer contention tuning** — on the Spark route, the agent asks
whether anything else writes the table (every tier but the lightest), derives
the concurrency mode, picks a lock provider, and emits a complete, runnable
block. What it does *not* do is tune for observed contention: lock
retry/timeout values, conflict-retry counts, and early conflict detection stay
at their defaults, because the right values depend on measured behavior rather
than design-time facts. That is Operati [...]
+- **Multi-writer contention tuning** — on the Spark route, the agent asks
whether anything else writes the table (every tier but the lightest), derives
the concurrency mode, picks a lock provider, and emits a complete, runnable
block. What it does *not* do is tune for observed contention: lock
retry/timeout values, conflict-retry counts, and early conflict detection stay
at their defaults, because the right values depend on measured behavior rather
than design-time facts. That is Operati [...]
- **Catalog / metastore sync beyond the common paths** — on the Spark route,
the agent asks which engines query the table and derives sync config for Hive
Metastore, AWS Glue, BigQuery, and DataHub, including the partition-extractor
and MOR `_ro`/`_rt` consequences. The Flink route emits only the five pinned
direct SQL / DFS Catalog / external HMS Catalog profiles and treats catalog
visibility as post-commit eventual. Not covered: Glue execution, live HMS
checks, existing-table registrat [...]
- **Non-Kafka source configs** — Kafka sources get full source-class +
schema-provider derivation; for DFS/JDBC/Pulsar/Kinesis sources you'll fill in
the `--source-class` and source properties from Hudi docs.
- **Cross-format interoperability (Apache XTable)** — explained and linked,
not configured. XTable translates Hudi metadata so the same files can be read
as Iceberg or Delta, but it is a separate incubating project: its config keys
live in its codebase rather than Hudi's, so this skill cannot machine-verify
them the way it verifies every `hoodie.*` key it emits. The agent names the one
durable coupling — **XTable's FAQ lists MOR as unsupported, so cross-format
interop argues for CoW**, w [...]
diff --git a/hudi-ai-operator/skills/hudi-architect/SKILL.md
b/hudi-ai-operator/skills/hudi-architect/SKILL.md
index 8d369d389a6d..bfcb78873dac 100644
--- a/hudi-ai-operator/skills/hudi-architect/SKILL.md
+++ b/hudi-ai-operator/skills/hudi-architect/SKILL.md
@@ -69,6 +69,9 @@ Within the Flink route, load the remaining references only
when needed:
table; use it to collect provenance-bearing evidence and produce the
assessment-only PR5 report.
- Load `references/flink-catalog-composition.md` only when the user needs
named catalog resolution
or external discovery; use its PR6 contract without changing design schema
1, 2, or 3.
+- Load `references/flink-writer-contract.md` only after F2 confirms an
independent writer or
+ standalone service; collect its PR7 facts without selecting concurrency,
locking, or index
+ settings.
The verified Flink baseline is Apache Hudi 1.2.0 at source revision
`f05c83f2b97732de7a558ff9b26959e1139c05f5` with Apache Flink 1.20; build and
test fixtures use
@@ -80,7 +83,8 @@ another Hudi or Flink version, do not silently reuse the
baseline: add a `REVIEW
unless a separate verified capability reference exists.
The Flink implementation includes the PR1 safety foundation, three bounded
executable paths, one
-existing-table assessment path, and PR6 catalog/metastore composition. PR2
append-only COW, PR3
+existing-table assessment path, PR6 catalog/metastore composition, and the PR7
assessment-only
+multi-writer Writer Contract. PR2 append-only COW, PR3
mutable COW, and PR4 mutable MOR all require a new table, confirmed single
writer, streaming
checkpointing, and `CONFIG_VALIDATED` from `validate_flink_design.py`. They
retain direct SQL by
default; PR6 may compose one of five pinned direct/DFS/HMS profiles without
changing the table
@@ -93,8 +97,10 @@ or 3 desired contract with structured evidence from an
existing table, but it ne
authorizes mutation. PR6 supports new-table direct Hive sync, Hudi DFS
Catalog, and external Hudi
HMS Catalog composition; AWS Glue, managed HMS, existing-table catalog
mutation, and live service
claims fail closed. Append-only MOR, standalone table services, detailed
compaction tuning,
-separate-writer or cross-engine takeover, multi-writer, alternative indexes,
custom mergers,
-retract normalization, and full CDC history still fail closed.
+separate-writer or cross-engine takeover, multi-writer executable output,
alternative indexes,
+custom mergers, retract normalization, and full CDC history still fail closed.
A confirmed
+multi-writer topology can be recorded by PR7, but OCC remains unverified and
the existing
+FLINK_STATE paths are incompatible with NBCC; the validator never chooses a
replacement.
## Flow structure
@@ -104,7 +110,8 @@ The conversation has three parts:
2. **Rounds 1-3** — workload questions, gated conditionally by tier.
3. **Output** — engine-specific output: the established Spark artifacts, a
non-executable Flink
assessment, a PR5 existing-table reconciliation report, or bounded
validator-rendered
- PR2/PR3/PR4 SQL artifacts, optionally composed with validated PR6 catalog
setup SQL.
+ PR2/PR3/PR4 SQL artifacts, optionally composed with validated PR6 catalog
setup SQL. PR7 adds
+ only a non-executable Writer Contract assessment.
Load `references/question-flow.md` for the full round-by-round question list
with conditional gating.
@@ -224,6 +231,10 @@ When a catalog block is present, retain the validator's
`catalog_setup_sql`, eff
configuration, and deployment checks as separate machine fields; `--emit-sql`
is only their
canonical ordered composition. Catalog deployment checks never imply that a
live service was
probed, and the generated SQL is never executed by this skill.
+When F2 confirms another writer, add the version-1 `writer_contract` described
by
+`flink-writer-contract.md`. Retain its normalized contract, capability notes,
and deployment checks
+as assessment fields. Never turn `COMPLETE` into a support claim: multi-writer
capability remains
+`UNVERIFIED` or `UNSUPPORTED`, and executable eligibility remains `false`.
For an existing table, create the same explicit desired contract, collect and
confirm evidence per
`flink-existing-table-reconciliation.md`, and run
`validate_flink_existing_table.py` instead. Emit
its report and deployment checks exactly as assessment data; never reinterpret
`COMPATIBLE` as
@@ -284,6 +295,8 @@ fact `INCOMPLETE`; do not ask the user to reveal the secret.
- Hand-write, repair, or extend SQL after `validate_flink_design.py` renders
it.
- Claim catalog/commit atomicity, execute generated catalog SQL, probe live
HMS/Glue, or emit
existing-table catalog mutations.
+- Recommend or emit an OCC/NBCC mode, lock provider, alternative index, or
multi-writer SQL from a
+ Writer Contract assessment.
- Treat a successful `HoodieTableFactory` construction as proof that the
source changelog is
append-only.
diff --git a/hudi-ai-operator/skills/hudi-architect/flink_writer_contract.py
b/hudi-ai-operator/skills/hudi-architect/flink_writer_contract.py
new file mode 100644
index 000000000000..cf8a4628fefb
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-architect/flink_writer_contract.py
@@ -0,0 +1,767 @@
+#!/usr/bin/env python3
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Assess the bounded Flink multi-writer Writer Contract.
+
+This module validates explicit writer facts. It does not select a concurrency
+mode, lock provider, index, failed-writes policy, or recovery procedure, and it
+never makes a multi-writer design executable.
+"""
+
+from __future__ import annotations
+
+from typing import Any, TypeGuard
+
+WRITER_CONTRACT_SCHEMA_VERSION = 1
+
+WRITER_ROLES = frozenset(
+ {
+ "INGESTION",
+ "BACKFILL",
+ "CLEANUP",
+ "STANDALONE_COMPACTION",
+ "STANDALONE_CLUSTERING",
+ "OTHER",
+ }
+)
+WRITER_RELATIONSHIPS = frozenset({"INDEPENDENT_CONCURRENT", "MAY_OVERLAP"})
+WRITER_ENGINES = frozenset({"FLINK", "SPARK", "OTHER", "UNKNOWN"})
+WRITE_OPERATIONS = frozenset(
+ {
+ "INSERT",
+ "INSERT_PREPPED",
+ "UPSERT",
+ "UPSERT_PREPPED",
+ "BULK_INSERT",
+ "BULK_INSERT_PREPPED",
+ "DELETE",
+ "DELETE_PREPPED",
+ "BOOTSTRAP",
+ "INSERT_OVERWRITE",
+ "BUCKET_RESCALE",
+ "CLUSTER",
+ "DELETE_PARTITION",
+ "INSERT_OVERWRITE_TABLE",
+ "COMPACT",
+ "INDEX",
+ "ALTER_SCHEMA",
+ "LOG_COMPACT",
+ "UNKNOWN",
+ }
+)
+CONCURRENCY_MODES = frozenset(
+ {
+ "SINGLE_WRITER",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "UNKNOWN",
+ }
+)
+LOCK_CONFIGURATION_CONSISTENCY = frozenset({"CONSISTENT", "CONFLICTING",
"UNKNOWN"})
+RESERVED_OWNER_IDS = frozenset({"EXTERNAL_CONTROL_PLANE", "UNKNOWN"})
+
+_CONTRACT_KEYS = {
+ "contract_schema",
+ "primary_writer_id",
+ "additional_writers",
+ "ownership",
+ "observed_concurrency",
+}
+_WRITER_KEYS = {
+ "writer_id",
+ "role",
+ "relationship_to_primary",
+ "engine",
+ "engine_version",
+ "hudi_version",
+ "operation",
+}
+_OWNERSHIP_KEYS = {
+ "schema_evolution_owner_id",
+ "table_configuration_owner_id",
+}
+_CONCURRENCY_KEYS = {
+ "mode",
+ "lock_provider_class",
+ "lock_configuration_consistency",
+}
+
+
+def _non_empty_string(value: Any) -> TypeGuard[str]:
+ return isinstance(value, str) and bool(value.strip())
+
+
+def _finding(code: str, message: str) -> dict[str, str]:
+ return {"code": code, "message": message}
+
+
+def _add_finding(findings: list[dict[str, str]], code: str, message: str) ->
None:
+ existing = next((item for item in findings if item["code"] == code), None)
+ if existing is None:
+ findings.append(_finding(code, message))
+ elif message not in existing["message"]:
+ existing["message"] += f"; {message}"
+
+
+def _invalid(findings: list[dict[str, str]], message: str) -> None:
+ _add_finding(findings, "FLINK_WRITER_CONTRACT_INVALID", message)
+
+
+def _check_keys(
+ value: dict[str, Any], allowed: set[str], field: str, findings:
list[dict[str, str]]
+) -> None:
+ unknown = sorted(set(value) - allowed)
+ if unknown:
+ _invalid(findings, f"{field} contains unsupported fields: {unknown}")
+
+
+def _deployment_checks() -> list[dict[str, Any]]:
+ return [
+ {"check": "verify_concurrent_writer_inventory", "required": True},
+ {"check": "verify_uniform_concurrency_mode", "required": True},
+ {"check": "verify_lock_provider_configuration", "required": True},
+ {"check": "verify_lazy_failed_writes_cleaner_policy", "required":
True},
+ {"check": "verify_writer_recovery_procedure", "required": True},
+ {"check": "verify_schema_evolution_ownership", "required": True},
+ ]
+
+
+def _not_applicable(
+ findings: list[dict[str, str]],
+ *,
+ capability: str = "NOT_APPLICABLE",
+ notes: list[dict[str, str]] | None = None,
+) -> dict[str, Any]:
+ return {
+ "writer_contract_result": "NOT_APPLICABLE",
+ "multi_writer_capability": capability,
+ "normalized_writer_contract": {},
+ "writer_capability_notes": notes or [],
+ "writer_deployment_checks": [],
+ "findings": findings,
+ }
+
+
+def _normalize_writer(
+ raw_writer: Any,
+ index: int,
+ findings: list[dict[str, str]],
+) -> dict[str, str] | None:
+ field = f"writer_contract.additional_writers[{index}]"
+ if not isinstance(raw_writer, dict):
+ _invalid(findings, f"{field} must be an object")
+ return None
+ _check_keys(raw_writer, _WRITER_KEYS, field, findings)
+
+ normalized: dict[str, str] = {}
+ for name in sorted(_WRITER_KEYS):
+ value = raw_writer.get(name)
+ if not _non_empty_string(value):
+ _invalid(findings, f"{field}.{name} must be a non-empty string")
+ continue
+ normalized[name] = value.strip()
+
+ enum_fields = {
+ "role": WRITER_ROLES,
+ "relationship_to_primary": WRITER_RELATIONSHIPS,
+ "engine": WRITER_ENGINES,
+ "operation": WRITE_OPERATIONS,
+ }
+ for name, allowed in enum_fields.items():
+ value = normalized.get(name)
+ if value is not None and value not in allowed:
+ _invalid(
+ findings,
+ f"{field}.{name} must be one of {sorted(allowed)}",
+ )
+ return normalized if set(normalized) == _WRITER_KEYS else None
+
+
+def _normalize_ownership(
+ raw_ownership: Any,
+ writer_ids: set[str],
+ findings: list[dict[str, str]],
+) -> dict[str, str] | None:
+ field = "writer_contract.ownership"
+ if not isinstance(raw_ownership, dict):
+ _invalid(findings, f"{field} must be an object")
+ return None
+ _check_keys(raw_ownership, _OWNERSHIP_KEYS, field, findings)
+
+ normalized: dict[str, str] = {}
+ schema_owner = raw_ownership.get("schema_evolution_owner_id")
+ if not _non_empty_string(schema_owner):
+ _invalid(findings, f"{field}.schema_evolution_owner_id is required")
+ else:
+ normalized["schema_evolution_owner_id"] = schema_owner.strip()
+
+ table_owner = raw_ownership.get("table_configuration_owner_id")
+ if table_owner is not None:
+ if not _non_empty_string(table_owner):
+ _invalid(
+ findings,
+ f"{field}.table_configuration_owner_id must be a non-empty
string",
+ )
+ else:
+ normalized["table_configuration_owner_id"] = table_owner.strip()
+
+ for name, owner_id in normalized.items():
+ if owner_id not in writer_ids | RESERVED_OWNER_IDS:
+ _invalid(
+ findings,
+ f"{field}.{name} must reference a declared writer or a
reserved owner",
+ )
+ elif owner_id == "UNKNOWN":
+ _add_finding(
+ findings,
+ "FLINK_WRITER_OWNERSHIP_UNRESOLVED",
+ f"{field}.{name} is unresolved",
+ )
+ return normalized if "schema_evolution_owner_id" in normalized else None
+
+
+def _normalize_concurrency(
+ raw_concurrency: Any,
+ findings: list[dict[str, str]],
+) -> dict[str, str] | None:
+ if raw_concurrency is None:
+ return None
+ field = "writer_contract.observed_concurrency"
+ if not isinstance(raw_concurrency, dict):
+ _invalid(findings, f"{field} must be an object")
+ return None
+ _check_keys(raw_concurrency, _CONCURRENCY_KEYS, field, findings)
+ normalized: dict[str, str] = {}
+ for name in sorted(_CONCURRENCY_KEYS):
+ value = raw_concurrency.get(name)
+ if not _non_empty_string(value):
+ _invalid(findings, f"{field}.{name} must be a non-empty string")
+ continue
+ normalized[name] = value.strip()
+ mode = normalized.get("mode")
+ if mode is not None and mode not in CONCURRENCY_MODES:
+ _invalid(findings, f"{field}.mode must be one of
{sorted(CONCURRENCY_MODES)}")
+ consistency = normalized.get("lock_configuration_consistency")
+ if consistency is not None and consistency not in
LOCK_CONFIGURATION_CONSISTENCY:
+ _invalid(
+ findings,
+ f"{field}.lock_configuration_consistency must be one of "
+ f"{sorted(LOCK_CONFIGURATION_CONSISTENCY)}",
+ )
+ return normalized if set(normalized) == _CONCURRENCY_KEYS else None
+
+
+def _add_note(notes: list[dict[str, str]], code: str, message: str) -> None:
+ if not any(note["code"] == code for note in notes):
+ notes.append({"code": code, "message": message})
+
+
+def _assess_contract_concurrency(
+ concurrency: dict[str, Any],
+ findings: list[dict[str, str]],
+ notes: list[dict[str, str]],
+) -> tuple[bool, bool]:
+ """Assess explicit multi-writer contract concurrency facts."""
+
+ conflict = False
+ unsupported = False
+ mode = concurrency.get("mode")
+ consistency = concurrency.get("lock_configuration_consistency")
+ if mode == "SINGLE_WRITER":
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "A multi-writer contract cannot observe SINGLE_WRITER concurrency
mode",
+ )
+ conflict = True
+ elif mode == "OPTIMISTIC_CONCURRENCY_CONTROL":
+ _add_note(
+ notes,
+ "FLINK_OCC_DEPLOYMENT_UNVERIFIED",
+ "OCC is recognized by the pinned source, but concurrent, lock, and
"
+ "recovery behavior is not exercised by this assessment.",
+ )
+ elif mode == "NON_BLOCKING_CONCURRENCY_CONTROL":
+ unsupported = True
+ _add_finding(
+ findings,
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED",
+ "The existing COW or FLINK_STATE-indexed MOR paths do not satisfy
the "
+ "pinned NBCC requirement for MOR with simple bucket index",
+ )
+ if consistency == "CONFLICTING":
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "Observed lock configuration is conflicting across writers",
+ )
+ conflict = True
+ return conflict, unsupported
+
+
+def _evidence_concurrency(
+ evidence: dict[str, Any],
+) -> dict[str, Any]:
+ return {
+ name: evidence[field]
+ for name, field in {
+ "mode": "flink_writer_config.write_concurrency_mode",
+ "lock_provider_class": "flink_writer_config.lock_provider_class",
+ "lock_configuration_consistency": (
+ "flink_writer_config.lock_configuration_consistency"
+ ),
+ }.items()
+ if field in evidence
+ }
+
+
+def _assess_evidence_concurrency(
+ concurrency: dict[str, Any],
+ writer_model: Any,
+ findings: list[dict[str, str]],
+ notes: list[dict[str, str]],
+) -> tuple[bool, bool, bool, bool]:
+ """Assess observed concurrency independently of Writer Contract
validity."""
+
+ if not concurrency:
+ return False, False, False, False
+
+ conflict = False
+ unsupported = False
+ relevant = False
+ conflicting_lock_evidence = False
+ mode = concurrency.get("mode")
+ consistency = concurrency.get("lock_configuration_consistency")
+ if mode == "SINGLE_WRITER":
+ relevant = writer_model == "multi_writer"
+ if writer_model == "multi_writer":
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "A multi-writer design cannot observe SINGLE_WRITER
concurrency mode",
+ )
+ conflict = True
+ elif mode == "OPTIMISTIC_CONCURRENCY_CONTROL":
+ relevant = True
+ _add_note(
+ notes,
+ "FLINK_OCC_DEPLOYMENT_UNVERIFIED",
+ "OCC is recognized by the pinned source, but concurrent, lock, and
"
+ "recovery behavior is not exercised by this assessment.",
+ )
+ if writer_model != "multi_writer":
+ _add_finding(
+ findings,
+ "FLINK_MULTI_WRITER_REVIEW",
+ "Observed concurrency configuration requires multi-writer
review",
+ )
+ elif mode == "NON_BLOCKING_CONCURRENCY_CONTROL":
+ relevant = True
+ unsupported = True
+ _add_finding(
+ findings,
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED",
+ "The existing COW or FLINK_STATE-indexed MOR paths do not satisfy
the "
+ "pinned NBCC requirement for MOR with simple bucket index",
+ )
+ if writer_model != "multi_writer":
+ _add_finding(
+ findings,
+ "FLINK_MULTI_WRITER_REVIEW",
+ "Observed concurrency configuration requires multi-writer
review",
+ )
+
+ if consistency == "CONFLICTING":
+ relevant = True
+ conflicting_lock_evidence = True
+ _add_finding(
+ findings,
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ "Observed lock configuration is conflicting across writers",
+ )
+ if writer_model != "multi_writer":
+ _add_finding(
+ findings,
+ "FLINK_MULTI_WRITER_REVIEW",
+ "Observed concurrency configuration requires multi-writer
review",
+ )
+ return conflict, unsupported, relevant, conflicting_lock_evidence
+
+
+def _writer_topology_evidence_findings(
+ writer_model: Any,
+ evidence: dict[str, Any],
+ conflicting_evidence_fields: set[str],
+ evidence_source_ids: dict[str, set[str]],
+ findings: list[dict[str, str]],
+) -> None:
+ inventory_fields = {
+ "writer_inventory.active_writers",
+ "writer_inventory.concurrent_writers",
+ }
+ conflicting = sorted(inventory_fields & conflicting_evidence_fields)
+ if conflicting:
+ _add_finding(
+ findings,
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ f"Existing-table evidence disagrees about writer inventory:
{conflicting}",
+ )
+
+ active_writers = evidence.get("writer_inventory.active_writers")
+ concurrent_writers = evidence.get("writer_inventory.concurrent_writers")
+ multiple_active = isinstance(active_writers, list) and len(active_writers)
> 1
+ if writer_model != "multi_writer" and (
+ multiple_active or concurrent_writers is True
+ ):
+ _add_finding(
+ findings,
+ "FLINK_MULTI_WRITER_REVIEW",
+ "Existing-table writer inventory requires multi-writer review",
+ )
+ active_sources =
evidence_source_ids.get("writer_inventory.active_writers", set())
+ concurrent_sources = evidence_source_ids.get(
+ "writer_inventory.concurrent_writers", set()
+ )
+ if (
+ multiple_active
+ and concurrent_writers is False
+ and bool(active_sources & concurrent_sources)
+ ):
+ _add_finding(
+ findings,
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ "Writer inventory reports multiple active writers while
concurrent_writers is false",
+ )
+
+
+def _evidence_conflicts(
+ normalized: dict[str, Any],
+ evidence: dict[str, Any],
+ conflicting_evidence_fields: set[str],
+ findings: list[dict[str, str]],
+) -> bool:
+ contract_fact_fields = {
+ "writer_inventory.active_writers",
+ "writer_inventory.schema_evolution_owner_id",
+ "flink_writer_config.write_concurrency_mode",
+ "flink_writer_config.lock_provider_class",
+ "flink_writer_config.lock_configuration_consistency",
+ }
+ conflicting = sorted(contract_fact_fields & conflicting_evidence_fields)
+ conflict = False
+ if conflicting:
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ f"Existing-table evidence disagrees about Writer Contract facts:
{conflicting}",
+ )
+ conflict = True
+
+ declared_writer_ids = {
+ normalized["primary_writer_id"],
+ *(writer["writer_id"] for writer in normalized["additional_writers"]),
+ }
+ expected: dict[str, Any] = {
+ "writer_inventory.schema_evolution_owner_id": normalized["ownership"][
+ "schema_evolution_owner_id"
+ ],
+ }
+ concurrency = normalized.get("observed_concurrency")
+ if isinstance(concurrency, dict):
+ expected.update(
+ {
+ "flink_writer_config.write_concurrency_mode":
concurrency["mode"],
+ "flink_writer_config.lock_provider_class": concurrency[
+ "lock_provider_class"
+ ],
+ "flink_writer_config.lock_configuration_consistency":
concurrency[
+ "lock_configuration_consistency"
+ ],
+ }
+ )
+
+ mismatches = [
+ field
+ for field, observed in evidence.items()
+ if field in expected and observed != expected[field]
+ ]
+ active_writers = evidence.get("writer_inventory.active_writers")
+ if isinstance(active_writers, list) and not set(active_writers) <=
declared_writer_ids:
+ mismatches.append("writer_inventory.active_writers")
+ mismatches = sorted(mismatches)
+ if mismatches:
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ f"Existing-table evidence conflicts with Writer Contract facts:
{mismatches}",
+ )
+ conflict = True
+ return conflict
+
+
+def assess_writer_contract(
+ design: dict[str, Any],
+ *,
+ evidence: dict[str, Any] | None = None,
+ conflicting_evidence_fields: set[str] | None = None,
+ evidence_source_ids: dict[str, set[str]] | None = None,
+) -> dict[str, Any]:
+ """Return a deterministic, assessment-only Writer Contract result."""
+
+ findings: list[dict[str, str]] = []
+ raw_safety = design.get("safety")
+ safety = raw_safety if isinstance(raw_safety, dict) else {}
+ writer_model = safety.get("writer_model")
+ raw_contract = design.get("writer_contract")
+ supplied_evidence = evidence or {}
+ supplied_conflicts = conflicting_evidence_fields or set()
+ supplied_sources = evidence_source_ids or {}
+ notes: list[dict[str, str]] = []
+
+ _writer_topology_evidence_findings(
+ writer_model,
+ supplied_evidence,
+ supplied_conflicts,
+ supplied_sources,
+ findings,
+ )
+ (
+ evidence_conflict,
+ evidence_unsupported,
+ evidence_relevant,
+ conflicting_lock_evidence,
+ ) = _assess_evidence_concurrency(
+ _evidence_concurrency(supplied_evidence),
+ writer_model,
+ findings,
+ notes,
+ )
+ capability = (
+ "UNSUPPORTED"
+ if evidence_unsupported
+ else "UNVERIFIED"
+ if writer_model == "multi_writer" or evidence_relevant
+ else "NOT_APPLICABLE"
+ )
+
+ if writer_model == "single_writer":
+ if raw_contract is not None:
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "writer_contract must be absent for a confirmed single-writer
design",
+ )
+ result = _not_applicable(findings, capability=capability,
notes=notes)
+ result["writer_contract_result"] = "CONFLICT"
+ return result
+ return _not_applicable(findings, capability=capability, notes=notes)
+
+ if writer_model != "multi_writer":
+ if raw_contract is not None:
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "Writer topology must be resolved before a Writer Contract is
assessed",
+ )
+ return _not_applicable(findings, capability=capability, notes=notes)
+
+ _add_finding(
+ findings,
+ "FLINK_MULTI_WRITER_REVIEW",
+ "Multi-writer execution remains assessment-only and requires
deployment review",
+ )
+ if raw_contract is None:
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_REQUIRED",
+ "A Writer Contract is required for a confirmed multi-writer
design",
+ )
+ return {
+ "writer_contract_result": "INCOMPLETE",
+ "multi_writer_capability": capability,
+ "normalized_writer_contract": {},
+ "writer_capability_notes": notes,
+ "writer_deployment_checks": _deployment_checks(),
+ "findings": findings,
+ }
+ if not isinstance(raw_contract, dict):
+ _invalid(findings, "writer_contract must be an object")
+ return {
+ "writer_contract_result": "INCOMPLETE",
+ "multi_writer_capability": capability,
+ "normalized_writer_contract": {},
+ "writer_capability_notes": notes,
+ "writer_deployment_checks": _deployment_checks(),
+ "findings": findings,
+ }
+
+ _check_keys(raw_contract, _CONTRACT_KEYS, "writer_contract", findings)
+ contract_schema = raw_contract.get("contract_schema")
+ if (
+ type(contract_schema) is not int
+ or contract_schema != WRITER_CONTRACT_SCHEMA_VERSION
+ ):
+ _invalid(
+ findings,
+ f"writer_contract.contract_schema must be
{WRITER_CONTRACT_SCHEMA_VERSION}",
+ )
+
+ primary_writer_id = raw_contract.get("primary_writer_id")
+ if not _non_empty_string(primary_writer_id):
+ _invalid(findings, "writer_contract.primary_writer_id is required")
+ normalized_primary_id = ""
+ else:
+ normalized_primary_id = primary_writer_id.strip()
+
+ raw_writers = raw_contract.get("additional_writers")
+ normalized_writers: list[dict[str, str]] = []
+ if not isinstance(raw_writers, list) or not raw_writers:
+ _invalid(
+ findings,
+ "writer_contract.additional_writers must be a non-empty array",
+ )
+ else:
+ for index, raw_writer in enumerate(raw_writers):
+ normalized_writer = _normalize_writer(raw_writer, index, findings)
+ if normalized_writer is not None:
+ normalized_writers.append(normalized_writer)
+
+ writer_ids = [writer["writer_id"] for writer in normalized_writers]
+ if len(writer_ids) != len(set(writer_ids)):
+ _invalid(
+ findings,
+ "writer_contract.additional_writers has duplicate writer_id
values",
+ )
+ if normalized_primary_id and normalized_primary_id in writer_ids:
+ _invalid(
+ findings,
+ "writer_contract.primary_writer_id must differ from every
additional writer_id",
+ )
+
+ declared_writer_ids = set(writer_ids)
+ if normalized_primary_id:
+ declared_writer_ids.add(normalized_primary_id)
+ ownership = _normalize_ownership(
+ raw_contract.get("ownership"), declared_writer_ids, findings
+ )
+ concurrency = _normalize_concurrency(
+ raw_contract.get("observed_concurrency"), findings
+ )
+
+ invalid = any(
+ finding["code"] == "FLINK_WRITER_CONTRACT_INVALID" for finding in
findings
+ )
+ if invalid:
+ return {
+ "writer_contract_result": "INCOMPLETE",
+ "multi_writer_capability": capability,
+ "normalized_writer_contract": {},
+ "writer_capability_notes": notes,
+ "writer_deployment_checks": _deployment_checks(),
+ "findings": findings,
+ }
+
+ raw_write = design.get("write")
+ write = raw_write if isinstance(raw_write, dict) else {}
+ primary_operation = str(write.get("operation", "UNKNOWN")).upper()
+ if primary_operation not in WRITE_OPERATIONS:
+ primary_operation = "UNKNOWN"
+ normalized: dict[str, Any] = {
+ "contract_schema": WRITER_CONTRACT_SCHEMA_VERSION,
+ "primary_writer_id": normalized_primary_id,
+ "primary_writer_facts": {
+ "engine": "FLINK",
+ "engine_version": design.get("flink_version", "UNKNOWN"),
+ "hudi_version": design.get("hudi_version", "UNKNOWN"),
+ "operation": primary_operation,
+ },
+ "additional_writers": sorted(
+ normalized_writers, key=lambda writer: writer["writer_id"]
+ ),
+ "ownership": ownership,
+ }
+ if concurrency is not None:
+ normalized["observed_concurrency"] = concurrency
+
+ conflict = evidence_conflict
+ if conflicting_lock_evidence:
+ _add_finding(
+ findings,
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "Observed lock configuration is conflicting across writers",
+ )
+ conflict = True
+ if concurrency is not None:
+ contract_conflict, contract_unsupported = _assess_contract_concurrency(
+ concurrency, findings, notes
+ )
+ conflict = conflict or contract_conflict
+ if contract_unsupported:
+ capability = "UNSUPPORTED"
+
+ if any(writer["engine"] != "FLINK" for writer in normalized_writers):
+ _add_note(
+ notes,
+ "FLINK_CROSS_ENGINE_WRITER_UNVERIFIED",
+ "Cross-engine writer compatibility is not certified by this
assessment.",
+ )
+ if any(
+ writer["role"] in {"STANDALONE_COMPACTION", "STANDALONE_CLUSTERING"}
+ for writer in normalized_writers
+ ):
+ _add_note(
+ notes,
+ "FLINK_STANDALONE_TABLE_SERVICE_UNVERIFIED",
+ "Standalone table-service writers require deployment review.",
+ )
+ if any(
+ writer["hudi_version"] != design.get("hudi_version")
+ for writer in normalized_writers
+ ):
+ _add_finding(
+ findings,
+ "FLINK_VERSION_UNVERIFIED",
+ "Writer Hudi versions are not uniformly pinned to the design
baseline",
+ )
+ if any(
+ writer["engine"] == "FLINK"
+ and writer["engine_version"] != design.get("flink_version")
+ for writer in normalized_writers
+ ):
+ _add_finding(
+ findings,
+ "FLINK_VERSION_UNVERIFIED",
+ "Flink writer versions are not uniformly pinned to the design
baseline",
+ )
+
+ conflict = (
+ _evidence_conflicts(
+ normalized,
+ supplied_evidence,
+ supplied_conflicts,
+ findings,
+ )
+ or conflict
+ )
+ return {
+ "writer_contract_result": "CONFLICT" if conflict else "COMPLETE",
+ "multi_writer_capability": capability,
+ "normalized_writer_contract": normalized,
+ "writer_capability_notes": notes,
+ "writer_deployment_checks": _deployment_checks(),
+ "findings": findings,
+ }
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
index 26ac22defe8a..afef45c9dc80 100644
---
a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
+++
b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
@@ -71,7 +71,10 @@ the current checkout contains it.
| Existing-table schema evolution or version upgrade | `REVIEW_REQUIRED`; no
commands emitted |
| Separate-writer or cross-engine takeover | `REVIEW_REQUIRED` |
| Confirmed single-writer detection | Supported |
-| Multi-writer or unknown writer model | `REVIEW_REQUIRED`; deferred to PR7 |
+| Unknown writer model | `REVIEW_REQUIRED`; no contract is fabricated |
+| Multi-writer Writer Contract | Assessment-only through PR7; never executable
|
+| OCC multi-writer capability | `UNVERIFIED`; deployment checks required |
+| NBCC on schemas 1–3 | `BLOCKED`; current COW/FLINK_STATE paths do not meet
the simple-bucket requirement |
| External-catalog requirement detection | Supported |
| Five direct SQL / DFS Catalog / HMS Catalog profiles | Supported for new
tables through PR6 |
| Direct SQL or DFS Catalog with sink-owned HMS sync | Supported |
@@ -150,6 +153,30 @@ embedded modes, checks DFS path derivation and external
HMS posture, exercises D
an insert plan, and verifies Hive sync/naming defaults. It does not start HMS
or Glue. Live
connectivity and visibility are deployment checks rather than static status
findings.
+## Implemented PR7 multi-writer Writer Contract
+
+PR7 adds Writer Contract schema 1 alongside design schemas 1, 2, and 3. It
records the primary
+Flink writer, additional ingestion/backfill/cleanup or standalone-service
writers, engine and Hudi
+versions, canonical write operations, and schema/table-configuration
ownership. Primary engine,
+version, and operation facts come from the existing design contract rather
than duplicate inputs.
+
+Observed concurrency mode, lock-provider class, and cross-writer lock
consistency are optional
+facts, not configuration recommendations. The pinned `WriteConcurrencyMode`
source recognizes OCC
+and NBCC as multi-writer modes, and `HoodieWriteConfig` makes LAZY
failed-write handling effective
+for them. These static facts do not exercise concurrent commits, a lock
service, or recovery, so
+OCC remains `UNVERIFIED`.
+
+Pinned Hudi 1.2.0 requires NBCC on MOR with simple bucket index (or the
metadata table). Schemas 1
+and 2 are COW and schema 3 uses global `FLINK_STATE`, so PR7 reports those
combinations
+`UNSUPPORTED` and blocks them without changing the index. Cross-engine and
standalone-service
+contracts remain review-required.
+
+The PR5 evidence envelope may additionally report active writer IDs,
schema-evolution owner,
+concurrency mode, lock provider, and lock consistency. Conflicting sources
remain scoped and
+provenance-bearing; they produce an unverified reconciliation and Writer
Contract `CONFLICT`, not
+an inferred migration. `COMPLETE` means only that the bounded inventory is
present. PR7 never
+returns `SUPPORTED`, never emits multi-writer SQL, and never makes executable
eligibility true.
+
## Implemented PR5 existing-table reconciliation contract
PR5 reuses the explicit schema-1, schema-2, or schema-3 desired design and
changes only the table
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
index 2bbfa5913a98..ac529449f421 100644
---
a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
+++
b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
@@ -13,9 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-schema_version = 6
+schema_version = 7
baseline_id = "hudi-1.2.0-flink-1.20"
-validation_surface =
"flink-sql-sink-existing-table-reconciliation-and-catalog-composition"
+validation_surface =
"flink-sql-sink-existing-table-reconciliation-catalog-composition-and-multi-writer-contract"
[hudi]
version = "1.2.0"
@@ -125,6 +125,68 @@ FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED =
"REVIEW_REQUIRED"
FLINK_GLUE_CATALOG_UNVERIFIED = "REVIEW_REQUIRED"
FLINK_MOR_CATALOG_NAMING_UNVERIFIED = "REVIEW_REQUIRED"
+[multi_writer_contract]
+contract_schema = 1
+supported_design_contract_schemas = [1, 2, 3]
+assessment_only = true
+executable_sql = false
+writer_contract_results = ["NOT_APPLICABLE", "COMPLETE", "INCOMPLETE",
"CONFLICT"]
+capability_results = ["NOT_APPLICABLE", "UNVERIFIED", "UNSUPPORTED"]
+roles = [
+ "INGESTION",
+ "BACKFILL",
+ "CLEANUP",
+ "STANDALONE_COMPACTION",
+ "STANDALONE_CLUSTERING",
+ "OTHER",
+]
+relationships = ["INDEPENDENT_CONCURRENT", "MAY_OVERLAP"]
+engines = ["FLINK", "SPARK", "OTHER", "UNKNOWN"]
+operations = [
+ "INSERT",
+ "INSERT_PREPPED",
+ "UPSERT",
+ "UPSERT_PREPPED",
+ "BULK_INSERT",
+ "BULK_INSERT_PREPPED",
+ "DELETE",
+ "DELETE_PREPPED",
+ "BOOTSTRAP",
+ "INSERT_OVERWRITE",
+ "BUCKET_RESCALE",
+ "CLUSTER",
+ "DELETE_PARTITION",
+ "INSERT_OVERWRITE_TABLE",
+ "COMPACT",
+ "INDEX",
+ "ALTER_SCHEMA",
+ "LOG_COMPACT",
+ "UNKNOWN",
+]
+concurrency_modes = [
+ "SINGLE_WRITER",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "UNKNOWN",
+]
+lock_configuration_consistency = ["CONSISTENT", "CONFLICTING", "UNKNOWN"]
+deployment_checks = [
+ "verify_concurrent_writer_inventory",
+ "verify_uniform_concurrency_mode",
+ "verify_lock_provider_configuration",
+ "verify_lazy_failed_writes_cleaner_policy",
+ "verify_writer_recovery_procedure",
+ "verify_schema_evolution_ownership",
+]
+
+[multi_writer_contract.finding_status]
+FLINK_MULTI_WRITER_REVIEW = "REVIEW_REQUIRED"
+FLINK_WRITER_COMBINATION_UNSUPPORTED = "BLOCKED"
+FLINK_WRITER_CONTRACT_CONFLICT = "REVIEW_REQUIRED"
+FLINK_WRITER_CONTRACT_INVALID = "INCOMPLETE"
+FLINK_WRITER_CONTRACT_REQUIRED = "INCOMPLETE"
+FLINK_WRITER_OWNERSHIP_UNRESOLVED = "REVIEW_REQUIRED"
+
[[catalog_profiles]]
id = "direct_sql_none"
resolution_mode = "direct_sql"
@@ -220,7 +282,12 @@ required_identity_facts = ["table_path"]
required_schema_facts = ["columns"]
required_timeline_facts = ["pending_instants"]
required_writer_inventory_facts = ["concurrent_writers"]
-conditional_writer_inventory_facts = ["previous_writer_engine",
"previous_writer_stopped"]
+conditional_writer_inventory_facts = [
+ "active_writers",
+ "previous_writer_engine",
+ "previous_writer_stopped",
+ "schema_evolution_owner_id",
+]
comparison_fields = [
"schema.columns",
"table_config.base_file_format",
@@ -246,11 +313,16 @@ comparison_fields = [
"writer_config.index_global",
"writer_config.index_state_ttl_days",
"writer_config.index_type",
+ "writer_config.lock_configuration_consistency",
+ "writer_config.lock_provider_class",
"writer_config.operation",
"writer_config.timestamp_key_generator",
+ "writer_config.write_concurrency_mode",
+ "writer_inventory.active_writers",
"writer_inventory.concurrent_writers",
"writer_inventory.previous_writer_engine",
"writer_inventory.previous_writer_stopped",
+ "writer_inventory.schema_evolution_owner_id",
]
comparison_results = ["CONFLICT", "MATCH", "NOT_APPLICABLE", "UNVERIFIED"]
reconciliation_results = ["COMPATIBLE", "CONFLICT", "UNVERIFIED"]
@@ -346,6 +418,24 @@ path =
"hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecord.java
sha256 = "11587947231d5a5727fea25169c986a20a3079d7f62787d3efe44c1f6e3f066e"
purpose = "Canonical Hudi metadata field names reserved by the write path"
+[[source_anchors]]
+id = "write_concurrency_mode"
+path =
"hudi-common/src/main/java/org/apache/hudi/common/model/WriteConcurrencyMode.java"
+sha256 = "13e8e5b91401c9a9e68d98cd19c2a44d450e691396c2ded4541b5479dd62b225"
+purpose = "Single-writer, optimistic, and non-blocking concurrency mode
semantics"
+
+[[source_anchors]]
+id = "write_operation_type"
+path =
"hudi-common/src/main/java/org/apache/hudi/common/model/WriteOperationType.java"
+sha256 = "0cced364652815143b7265c8ccd7ea9a6db87f9c4f9be28b05923adda9132c68"
+purpose = "Canonical Hudi 1.2.0 write operation names"
+
+[[source_anchors]]
+id = "hoodie_write_config"
+path =
"hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java"
+sha256 = "359d69bcf5c8c442923d81544c15bec4e5ade91afca3dc71043271d1d2ee4d8c"
+purpose = "Multi-writer cleaner policy and non-blocking concurrency
constraints"
+
[[source_anchors]]
id = "hoodie_table_config"
path =
"hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java"
@@ -500,7 +590,7 @@ purpose = "Global-index record routing and partition-change
behavior"
id = "streamer_util"
path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java"
sha256 = "c771e137ebcd45c94a941857a246dc741e1dde2c0e0158900fd48fe8884f83e2"
-purpose = "Write-side record merge mode inference and table-property
persistence"
+purpose = "Flink lock-provider resolution, record merge mode inference, and
table-property persistence"
# Catalog option types are tracked separately from table connector options.
[[catalog_option_types]]
@@ -970,3 +1060,15 @@ id = "FLINK_MOR_CATALOG_NAMING_UNVERIFIED"
requirement = "Pin MOR Hive registrations to RO, RT, or ALL names derived from
query modes."
python_test = "test_pr6_rejects_mor_catalog_naming_drift"
java_test = "testPinnedMorHiveSyncNamingDefaults"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MULTI_WRITER_REVIEW"
+requirement = "Keep recognized OCC and NBCC multi-writer modes assessment-only
and require the pinned LAZY failed-writes policy."
+python_test = "test_pr7_complete_occ_contract_is_assessment_only"
+java_test = "testPinnedMultiWriterModesRequireLazyFailedWritesPolicy"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_WRITER_COMBINATION_UNSUPPORTED"
+requirement = "Reject NBCC for the existing COW and FLINK_STATE-indexed MOR
paths because Hudi 1.2.0 requires MOR with simple bucket index."
+python_test = "test_pr7_nbcc_is_unsupported_for_existing_bounded_indexes"
+java_test = "testPinnedNbccRejectsCowAndNonSimpleBucketIndex"
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
index 4b36cec91bd0..d64273fb157e 100644
---
a/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
+++
b/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
@@ -30,8 +30,9 @@ Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
Baseline ID: hudi-1.2.0-flink-1.20
Hudi source revision: f05c83f2b97732de7a558ff9b26959e1139c05f5
Fixture version: Apache Flink 1.20.1
-Capability manifest schema: 6
+Capability manifest schema: 7
Catalog contract schema: 1
+Writer contract schema: 1
Catalog profile: <one pinned profile or unresolved>
Design contract schema: <1 append-only COW | 2 mutable COW | 3 mutable MOR>
@@ -65,8 +66,9 @@ Use only the machine result from
`validate_flink_existing_table.py`:
Flink design status: <CONFIG_VALIDATED | INCOMPLETE | BLOCKED |
REVIEW_REQUIRED>
Reconciliation result: <COMPATIBLE | CONFLICT | UNVERIFIED>
Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
-Capability manifest schema: 6
+Capability manifest schema: 7
Catalog contract schema: 1
+Writer contract schema: 1
Catalog profile: <direct_sql_none or requested non-legacy profile>
Catalog setup SQL: <always empty for an existing table>
Effective catalog configuration: <validator-produced mapping>
@@ -74,6 +76,8 @@ Reconciliation schema: 1
Desired design contract schema: <1 | 2 | 3>
Writer transition: <same_job_restore | cold_start_new_job |
flink_writer_takeover |
cross_engine_takeover | unknown>
+Writer contract result: <NOT_APPLICABLE | COMPLETE | INCOMPLETE | CONFLICT>
+Multi-writer capability: <NOT_APPLICABLE | UNVERIFIED | UNSUPPORTED>
Evidence:
- <evidence ID, kind, table path, in-scope flag, observation time, and digest;
no raw secrets>
@@ -92,6 +96,8 @@ Deployment checks:
- For a mutable cold start, verify index bootstrap before relying on the first
checkpoint.
- Retain the validator-produced catalog deployment checks for any non-legacy
profile; these are
observations to perform, not permission to mutate an existing registration.
+- Retain Writer Contract checks when applicable; they do not establish live
lock or recovery
+ behavior.
Executable output:
- Withheld because PR5 is assessment-only, including for COMPATIBLE results.
@@ -112,8 +118,9 @@ Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
Baseline ID: hudi-1.2.0-flink-1.20
Hudi source revision: f05c83f2b97732de7a558ff9b26959e1139c05f5
Fixture version: Apache Flink 1.20.1
-Capability manifest schema: 6
+Capability manifest schema: 7
Catalog contract schema: 1
+Writer contract schema: 1
Catalog profile: <direct_sql_none | direct_sql_hive_metastore |
hudi_dfs_catalog_none |
hudi_dfs_catalog_hive_metastore | hudi_hms_catalog_hive_metastore>
Design contract schema: <1 append-only COW | 2 mutable COW | 3 mutable MOR>
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
index e17bcacfc70f..bfe27efde3ac 100644
---
a/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
+++
b/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
@@ -74,6 +74,11 @@ do not replace the shared table-design rules and do not form
a standalone planne
| Table lifecycle unknown | `FLINK_TABLE_LIFECYCLE_REQUIRED` | `INCOMPLETE` |
No |
| Independent writer exists | `FLINK_MULTI_WRITER_REVIEW` | `REVIEW_REQUIRED`
| No |
| Writer model unknown | `FLINK_WRITER_MODEL_UNRESOLVED` | `REVIEW_REQUIRED` |
No |
+| Confirmed multi-writer topology has no Writer Contract |
`FLINK_WRITER_CONTRACT_REQUIRED` | `INCOMPLETE` | No |
+| Writer Contract is malformed or contains unsupported fields |
`FLINK_WRITER_CONTRACT_INVALID` | `INCOMPLETE` | No |
+| Schema or table-configuration ownership is unresolved |
`FLINK_WRITER_OWNERSHIP_UNRESOLVED` | `REVIEW_REQUIRED` | No |
+| Writer Contract facts conflict internally or with scoped evidence |
`FLINK_WRITER_CONTRACT_CONFLICT` | `REVIEW_REQUIRED` | No |
+| Observed concurrency combination is incompatible with the existing bounded
design | `FLINK_WRITER_COMBINATION_UNSUPPORTED` | `BLOCKED` | No |
| External catalog required | `FLINK_EXTERNAL_CATALOG_REVIEW` |
`REVIEW_REQUIRED` | No |
| Catalog requirement unknown | `FLINK_CATALOG_REQUIREMENT_UNRESOLVED` |
`REVIEW_REQUIRED` | No |
| Physical schema missing, insufficient, or available only through an
unreadable pointer | `FLINK_PHYSICAL_SCHEMA_REQUIRED` | `INCOMPLETE` | No |
@@ -261,7 +266,7 @@ to the executable contract just like `F10_SAFE_APPEND`.
compaction-operation execution are enabled with `num_commits` and five delta
commits. Five is a
fixed Hudi 1.2.0 effective default, not a tuned SLA.
- A standalone compactor is an independent writer and remains
`REVIEW_REQUIRED`; concurrency
- configuration is deferred to PR7. Unknown ownership also withholds SQL.
+ configuration is not emitted by the PR7 Writer Contract. Unknown ownership
also withholds SQL.
- Snapshot readers see base plus log state at a visible completed instant.
Read-optimized readers
see compacted base-file state and advance only after compaction. Reader
`query.type` is not a
sink option and is not rendered into the DDL.
@@ -333,6 +338,28 @@ to the executable contract just like `F10_SAFE_APPEND`.
disposition remains `UNVERIFIED`, executable eligibility remains false, and
no catalog SQL is
emitted.
+## PR7 multi-writer Writer Contract invariants
+
+- Writer Contract schema 1 is orthogonal to design schemas 1, 2, and 3. It
never changes the
+ established single-writer SQL or creates a fourth design schema.
+- A confirmed single writer omits the block and returns `NOT_APPLICABLE`. An
unknown writer model
+ does not fabricate a contract. A confirmed independent writer requires the
block and always
+ retains `FLINK_MULTI_WRITER_REVIEW`.
+- The primary writer ID is unique. Additional writers record a bounded role,
relationship,
+ engine/version, Hudi version, and canonical Hudi 1.2.0 operation. Primary
engine/version and
+ operation facts are derived from the existing design contract.
+- Schema-evolution ownership is required. Table-configuration ownership is
optional. Same-job
+ compaction remains in schema 3; separately deployed compaction or clustering
is an additional
+ writer role.
+- Observed concurrency is optional and never a recommendation. OCC is
recognized but remains
+ `UNVERIFIED`. NBCC requires MOR with simple bucket index, so the existing
COW and
+ FLINK_STATE-indexed MOR paths are `UNSUPPORTED`; never switch indexes
automatically.
+- Existing-table writer observations retain PR5 provenance and scope.
Contradictory observations
+ make the Writer Contract `CONFLICT` and reconciliation/change disposition
`UNVERIFIED`, not an
+ invented migration.
+- `COMPLETE` means only that required writer facts were captured. Multi-writer
executable
+ eligibility is always false, no SQL is emitted, and deployment checks are
not live evidence.
+
## Evidence and secret handling
User-provided DDL, logs, schemas, table properties, and catalog output are
evidence, not
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
index dca4bc3f526f..3f6b44282c9c 100644
---
a/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
+++
b/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
@@ -26,7 +26,9 @@ the architecture, execute evidence, mutate the table, plan a
migration, or rende
The first compatible reconciliation path supports Hudi 1.2.0 with Flink 1.20,
one writer, direct
SQL with no external discovery, and desired contract schemas 1, 2, or 3. It
can validate an exact
existing-table match. Schema evolution, table upgrades, catalog registration,
multi-writer
-operation, separate-writer takeover, and cross-engine takeover remain
unverified or blocked.
+operation, separate-writer takeover, and cross-engine takeover remain
unverified or blocked. PR7
+can attach a Writer Contract to the same evidence envelope, but that path
always remains
+non-executable and cannot become `COMPATIBLE` merely because the contract is
complete.
`CONFIG_VALIDATED` in this assessment means that the static comparison is
complete. It always has
`executable_eligible: false`; the PR5 path emits a report and deployment
checks, never SQL.
@@ -93,6 +95,29 @@ contract. If redaction removes a load-bearing fact, leave it
incomplete.
- `writer_inventory` and `timeline_summary` are time-sensitive operational
observations. Always
require a deployment-time recheck even when the comparison succeeds.
+For a PR7 Writer Contract, `writer_inventory` may also supply `active_writers`
and
+`schema_evolution_owner_id`; `flink_writer_config` may supply
`write_concurrency_mode`,
+`lock_provider_class`, and `lock_configuration_consistency`. These are
optional observations and
+retain the same evidence scope and provenance. If sources disagree, or an
observation contradicts
+the explicit Writer Contract, report `FLINK_EXISTING_EVIDENCE_CONFLICT` and/or
+`FLINK_WRITER_CONTRACT_CONFLICT`, keep reconciliation and change disposition
`UNVERIFIED`, and do
+not infer migration work.
+
+Evaluate supplied concurrency observations even when the optional Writer
Contract concurrency
+block is absent. Retain them as provenance-bearing `UNVERIFIED` comparisons
without promoting
+them into the normalized contract. Continue the capability assessment when the
contract itself is
+missing or invalid, so structural incompleteness cannot hide an unsupported
bounded combination.
+Before accepting the single-writer branch, also reconcile the active-writer
inventory: more than
+one active writer contradicts the desired topology. Treat that snapshot as
internally inconsistent
+with `concurrent_writers: false` only when both facts share an evidence source.
+
+For a declared multi-writer topology, an active-writer snapshot may contain
any subset of declared
+writer IDs; inactive declared writers are not a mismatch. An undeclared active
writer remains a
+Writer Contract conflict. `concurrent_writers` is an operational observation,
not proof that every
+declared writer is active, so do not require it to be `true` merely because
the desired topology is
+multi-writer. General deployment checks require `confirm_single_writer` only
for single-writer
+designs.
+
Do not silently merge contradictory sources. Preserve provenance in each
comparison item.
## Writer transition
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
index c923c7b6c78d..d22894bcc9ab 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
@@ -84,11 +84,15 @@ An asynchronous table service that runs inside and is
coordinated by the same Fl
independent writer. A separately deployed compactor or clustering job is.
- Confirmed single writer → record `SINGLE_WRITER` as a confirmed fact and
continue.
-- Another independent writer → `REVIEW_REQUIRED` with
`FLINK_MULTI_WRITER_REVIEW`.
+- Another independent writer → load `flink-writer-contract.md`, collect the
version-1 Writer
+ Contract, and retain `REVIEW_REQUIRED` with `FLINK_MULTI_WRITER_REVIEW` even
when the contract is
+ complete.
- Not sure → `REVIEW_REQUIRED` with `FLINK_WRITER_MODEL_UNRESOLVED`.
+For a confirmed independent writer, collect identity, role, relationship,
engine/version, Hudi
+version, operation, and ownership. Collect concurrency mode and lock class
only as observed facts.
Do not recommend OCC, NBCC, a lock provider, or an index/concurrency
combination on this bounded
-path.
+path, and do not create a contract while the writer model is unknown.
## F3 — External catalog visibility
@@ -402,7 +406,7 @@ the report and deployment checks with executable
eligibility `false`, and never
known table-property or schema conflict is `CONFLICT`; incomplete,
contradictory, evolutionary,
or unsupported evidence is `UNVERIFIED`.
-## PR2/PR3/PR4/PR5/PR6 validation and completion
+## PR2/PR3/PR4/PR5/PR6/PR7 validation and completion
After all applicable questions, build a version-1 JSON contract for
append-only COW, a version-2
contract for mutable COW, or a version-3 contract for mutable MOR. All contain
confirmed safety
@@ -420,6 +424,14 @@ existing table, a non-legacy block is accepted only to
report
`FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED`; do not collect a new evidence
kind or infer that
registration already exists.
+When F2 confirms an independent writer, add Writer Contract schema 1 without
changing design
+schema 1, 2, or 3. The contract records writer inventory,
schema/table-configuration ownership,
+and optional observed concurrency facts. Run the same validator, retain
+`writer_contract_result`, `multi_writer_capability`, normalized facts, notes,
and deployment
+checks, and withhold all SQL. A complete contract is assessment completeness,
not an executable
+support claim. OCC remains `UNVERIFIED`; NBCC is `UNSUPPORTED` for the current
COW and
+FLINK_STATE-indexed MOR paths.
+
For a new table, run:
```bash
@@ -437,5 +449,6 @@ Executable eligible: true
```
Any validation failure withholds all SQL and reports executable eligibility as
`false`. Consult
-`flink-decision-overrides.md` for precedence, `flink-catalog-composition.md`
for PR6 profiles, and
-`flink-config-templates.md` for both envelopes.
+`flink-decision-overrides.md` for precedence, `flink-catalog-composition.md`
for PR6 profiles,
+`flink-writer-contract.md` for PR7 assessment rules, and
`flink-config-templates.md` for both
+envelopes.
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
index 210408fe2448..db9018e7ea5a 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
@@ -194,10 +194,11 @@ The table lifecycle must be known before choosing a path.
**Trigger:** Another pipeline, backfill, cleanup writer, cross-engine writer,
standalone compactor,
or standalone clustering job can commit to the table.
-**Message:** The single-writer assumptions are invalid. This path does not
claim an OCC, NBCC,
-lock-provider, index, or table-service combination is safe.
+**Message:** The single-writer assumptions are invalid. PR7 can record a
Writer Contract, but it
+does not claim an OCC, NBCC, lock-provider, index, or table-service
combination is safe.
-**Action:** `REVIEW_REQUIRED`; withhold concurrency-sensitive configuration.
+**Action:** `REVIEW_REQUIRED`; collect the Writer Contract and withhold
concurrency-sensitive
+configuration and executable output.
## FLINK_WRITER_MODEL_UNRESOLVED
@@ -209,6 +210,49 @@ be treated as safe.
**Action:** `REVIEW_REQUIRED`; do not infer `SINGLE_WRITER` or recommend
concurrency settings.
+## FLINK_WRITER_CONTRACT_REQUIRED
+
+**Trigger:** The writer model is confirmed multi-writer but no version-1
Writer Contract exists.
+
+**Message:** The writer inventory and ownership facts are incomplete.
+
+**Action:** `INCOMPLETE`; collect the bounded PR7 contract without selecting
concurrency settings.
+
+## FLINK_WRITER_CONTRACT_INVALID
+
+**Trigger:** A Writer Contract has an invalid type, unknown field, duplicate
writer ID, invalid
+enum, undeclared owner, or another structural error.
+
+**Message:** The validator cannot normalize the writer topology without
guessing.
+
+**Action:** `INCOMPLETE`; correct the explicit facts and emit no executable
output.
+
+## FLINK_WRITER_OWNERSHIP_UNRESOLVED
+
+**Trigger:** Schema-evolution or supplied table-configuration ownership is
explicitly `UNKNOWN`.
+
+**Message:** Concurrent schema or configuration changes have no confirmed
owner.
+
+**Action:** `REVIEW_REQUIRED`; establish ownership before deployment.
+
+## FLINK_WRITER_CONTRACT_CONFLICT
+
+**Trigger:** The contract contradicts the writer model or scoped
existing-table evidence, or the
+observed lock configuration is conflicting.
+
+**Message:** PR7 preserves both facts and does not select one as authoritative.
+
+**Action:** `REVIEW_REQUIRED`; return `CONFLICT`, withhold SQL, and reconcile
the evidence.
+
+## FLINK_WRITER_COMBINATION_UNSUPPORTED
+
+**Trigger:** NBCC is observed for design schema 1 or 2 COW, or schema 3 MOR
with `FLINK_STATE`.
+
+**Message:** Pinned Hudi 1.2.0 requires NBCC on MOR with simple bucket index
(or the metadata
+table), which none of the existing executable schemas uses.
+
+**Action:** `BLOCKED`; do not switch the table type or index automatically.
+
## FLINK_EXTERNAL_CATALOG_REVIEW
**Trigger:** An external engine or BI tool must discover the table through a
catalog or metastore.
diff --git
a/hudi-ai-operator/skills/hudi-architect/references/flink-writer-contract.md
b/hudi-ai-operator/skills/hudi-architect/references/flink-writer-contract.md
new file mode 100644
index 000000000000..e03113a88546
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-writer-contract.md
@@ -0,0 +1,105 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+# Flink multi-writer Writer Contract
+
+Use this reference only after F2 confirms an independent writer or separately
deployed table
+service. The contract is orthogonal to design schemas 1, 2, and 3. It records
the writer topology
+and assesses pinned Hudi 1.2.0 constraints; it never selects OCC, NBCC, a lock
provider, an index,
+or a recovery strategy, and it never makes multi-writer SQL executable.
+
+## Contract shape
+
+Set `writer_contract.contract_schema` to `1`. Record one unique
`primary_writer_id` for the Flink
+design and at least one `additional_writers` entry. Each additional writer
requires:
+
+- a unique `writer_id`;
+- role: `INGESTION`, `BACKFILL`, `CLEANUP`, `STANDALONE_COMPACTION`,
+ `STANDALONE_CLUSTERING`, or `OTHER`;
+- relationship: `INDEPENDENT_CONCURRENT` or `MAY_OVERLAP`;
+- engine: `FLINK`, `SPARK`, `OTHER`, or `UNKNOWN`;
+- concrete engine and Hudi version strings, using the literal `UNKNOWN` only
when the fact has
+ explicitly been established as unknown; and
+- a canonical Hudi 1.2.0 `WriteOperationType` name, or `UNKNOWN`.
+
+Do not duplicate primary engine, engine version, Hudi version, or operation in
the input block.
+The validator derives those facts from the existing Flink design contract.
+
+`ownership.schema_evolution_owner_id` is required and must reference a
declared writer,
+`EXTERNAL_CONTROL_PLANE`, or `UNKNOWN`. `table_configuration_owner_id` is
optional with the same
+reference rule. `UNKNOWN` ownership produces
`FLINK_WRITER_OWNERSHIP_UNRESOLVED`.
+
+Express a standalone compactor or clustering service as an additional writer
with the matching
+role. Same-job asynchronous compaction remains part of design schema 3 and is
not an independent
+writer.
+
+## Observed concurrency
+
+`observed_concurrency` is optional and observation-only. When supplied, it
requires:
+
+- `mode`: `SINGLE_WRITER`, `OPTIMISTIC_CONCURRENCY_CONTROL`,
+ `NON_BLOCKING_CONCURRENCY_CONTROL`, or `UNKNOWN`;
+- the observed lock-provider class name; and
+- `lock_configuration_consistency`: `CONSISTENT`, `CONFLICTING`, or `UNKNOWN`.
+
+Never ask the user to choose a mode or provider to make validation pass. Do
not collect provider
+credentials, timeout/retry tuning, namespaces, conflict rates, keyspace
assumptions, or file-group
+overlap guesses.
+
+The pinned source recognizes OCC and NBCC as multi-writer modes and requires
an effective LAZY
+failed-writes policy. PR7 still reports OCC as `UNVERIFIED`: its fixtures do
not run concurrent
+writers, a lock service, or recovery. NBCC requires MOR with simple bucket
index (or the metadata
+table). The existing schema-1/2 COW paths and schema-3 MOR path with
`FLINK_STATE` therefore return
+`UNSUPPORTED` and `FLINK_WRITER_COMBINATION_UNSUPPORTED`. Never switch the
index automatically.
+
+## Existing-table evidence
+
+PR5 evidence may optionally add `active_writers` and
`schema_evolution_owner_id` to
+`writer_inventory`, and `write_concurrency_mode`, `lock_provider_class`, and
+`lock_configuration_consistency` to `flink_writer_config`. Every item retains
the existing
+evidence ID, timestamp, digest, and exact table-path scope.
+
+Only in-scope, non-conflicting observations contribute. Evidence disagreement
or a mismatch with
+the explicit Writer Contract produces `FLINK_EXISTING_EVIDENCE_CONFLICT` and/or
+`FLINK_WRITER_CONTRACT_CONFLICT`. Keep the Writer Contract `CONFLICT`,
reconciliation
+`UNVERIFIED`, change disposition `UNVERIFIED`, and executable eligibility
`false`; do not infer a
+migration or silently replace either fact.
+
+Assess supplied concurrency evidence even when `observed_concurrency` is
absent from the explicit
+contract. Preserve that evidence and its provenance as an unverified
comparison; do not copy it
+into `normalized_writer_contract` or treat missing optional contract facts as
inferred values.
+Apply the same independent capability assessment when the Writer Contract is
missing or invalid:
+the contract remains `INCOMPLETE`, while bounded NBCC evidence still returns
`UNSUPPORTED`.
+Reject concurrency modes and lock-consistency values outside the enumerations
above.
+
+Treat `active_writers` as a time-sensitive snapshot, not as the complete
declared topology. Every
+observed active writer must be declared, but a declared writer need not be
active in that snapshot.
+Multiple active writers cannot validate a single-writer design. Report the
combination of multiple
+active writers and `concurrent_writers: false` as internally inconsistent only
when both facts come
+from the same evidence source; do not combine separate snapshots into a
fabricated contradiction.
+
+## Result semantics
+
+Return `writer_contract_result` as `NOT_APPLICABLE`, `COMPLETE`, `INCOMPLETE`,
or `CONFLICT` and
+`multi_writer_capability` as `NOT_APPLICABLE`, `UNVERIFIED`, or `UNSUPPORTED`.
Include the
+normalized contract, capability notes, and deployment checks. Never return
`SUPPORTED`.
+
+Deployment checks remind the operator to verify writer inventory, uniform
concurrency mode, lock
+configuration, LAZY failed-write handling, recovery procedure, and schema
ownership. They are not
+evidence that a deployment or service was inspected. The general
existing-table checks include
+`confirm_single_writer` only for a single-writer design; a multi-writer design
uses the bounded
+Writer Contract deployment checks instead.
diff --git
a/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
b/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
index cc28beee9221..46aefbe3d798 100644
--- a/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
+++ b/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
@@ -40,7 +40,7 @@ DEFAULT_MANIFEST = (
SKILL_DIR / "references" / "flink-1.20-hudi-1.2.0-capabilities.toml"
)
-EXPECTED_SCHEMA_VERSION = 6
+EXPECTED_SCHEMA_VERSION = 7
EXPECTED_BASELINE_ID = "hudi-1.2.0-flink-1.20"
EXPECTED_HUDI_VERSION = "1.2.0"
EXPECTED_HUDI_SOURCE_REVISION = "f05c83f2b97732de7a558ff9b26959e1139c05f5"
@@ -333,6 +333,73 @@ EXPECTED_CATALOG_OPTION_TYPES = [
"source_symbol": "TABLE_EXTERNAL",
},
]
+EXPECTED_MULTI_WRITER_CONTRACT = {
+ "contract_schema": 1,
+ "supported_design_contract_schemas": [1, 2, 3],
+ "assessment_only": True,
+ "executable_sql": False,
+ "writer_contract_results": [
+ "NOT_APPLICABLE",
+ "COMPLETE",
+ "INCOMPLETE",
+ "CONFLICT",
+ ],
+ "capability_results": ["NOT_APPLICABLE", "UNVERIFIED", "UNSUPPORTED"],
+ "roles": [
+ "INGESTION",
+ "BACKFILL",
+ "CLEANUP",
+ "STANDALONE_COMPACTION",
+ "STANDALONE_CLUSTERING",
+ "OTHER",
+ ],
+ "relationships": ["INDEPENDENT_CONCURRENT", "MAY_OVERLAP"],
+ "engines": ["FLINK", "SPARK", "OTHER", "UNKNOWN"],
+ "operations": [
+ "INSERT",
+ "INSERT_PREPPED",
+ "UPSERT",
+ "UPSERT_PREPPED",
+ "BULK_INSERT",
+ "BULK_INSERT_PREPPED",
+ "DELETE",
+ "DELETE_PREPPED",
+ "BOOTSTRAP",
+ "INSERT_OVERWRITE",
+ "BUCKET_RESCALE",
+ "CLUSTER",
+ "DELETE_PARTITION",
+ "INSERT_OVERWRITE_TABLE",
+ "COMPACT",
+ "INDEX",
+ "ALTER_SCHEMA",
+ "LOG_COMPACT",
+ "UNKNOWN",
+ ],
+ "concurrency_modes": [
+ "SINGLE_WRITER",
+ "OPTIMISTIC_CONCURRENCY_CONTROL",
+ "NON_BLOCKING_CONCURRENCY_CONTROL",
+ "UNKNOWN",
+ ],
+ "lock_configuration_consistency": ["CONSISTENT", "CONFLICTING", "UNKNOWN"],
+ "deployment_checks": [
+ "verify_concurrent_writer_inventory",
+ "verify_uniform_concurrency_mode",
+ "verify_lock_provider_configuration",
+ "verify_lazy_failed_writes_cleaner_policy",
+ "verify_writer_recovery_procedure",
+ "verify_schema_evolution_ownership",
+ ],
+ "finding_status": {
+ "FLINK_MULTI_WRITER_REVIEW": "REVIEW_REQUIRED",
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED": "BLOCKED",
+ "FLINK_WRITER_CONTRACT_CONFLICT": "REVIEW_REQUIRED",
+ "FLINK_WRITER_CONTRACT_INVALID": "INCOMPLETE",
+ "FLINK_WRITER_CONTRACT_REQUIRED": "INCOMPLETE",
+ "FLINK_WRITER_OWNERSHIP_UNRESOLVED": "REVIEW_REQUIRED",
+ },
+}
EXPECTED_EXISTING_TABLE_RECONCILIATION_CONTRACT = {
"reconciliation_schema": 1,
"supported_design_contract_schemas": [1, 2, 3],
@@ -387,8 +454,10 @@ EXPECTED_EXISTING_TABLE_RECONCILIATION_CONTRACT = {
"required_timeline_facts": ["pending_instants"],
"required_writer_inventory_facts": ["concurrent_writers"],
"conditional_writer_inventory_facts": [
+ "active_writers",
"previous_writer_engine",
"previous_writer_stopped",
+ "schema_evolution_owner_id",
],
"comparison_fields": [
"schema.columns",
@@ -415,11 +484,16 @@ EXPECTED_EXISTING_TABLE_RECONCILIATION_CONTRACT = {
"writer_config.index_global",
"writer_config.index_state_ttl_days",
"writer_config.index_type",
+ "writer_config.lock_configuration_consistency",
+ "writer_config.lock_provider_class",
"writer_config.operation",
"writer_config.timestamp_key_generator",
+ "writer_config.write_concurrency_mode",
+ "writer_inventory.active_writers",
"writer_inventory.concurrent_writers",
"writer_inventory.previous_writer_engine",
"writer_inventory.previous_writer_stopped",
+ "writer_inventory.schema_evolution_owner_id",
],
"comparison_results": ["CONFLICT", "MATCH", "NOT_APPLICABLE",
"UNVERIFIED"],
"reconciliation_results": ["COMPATIBLE", "CONFLICT", "UNVERIFIED"],
@@ -534,6 +608,14 @@ EXPECTED_ACCEPTANCE_EVIDENCE = {
"test_pr4_valid_mutable_mor_design_renders_exact_executable_sql",
"testMutableMorSinkAndCompactionPlan",
),
+ "FLINK_MULTI_WRITER_REVIEW": (
+ "test_pr7_complete_occ_contract_is_assessment_only",
+ "testPinnedMultiWriterModesRequireLazyFailedWritesPolicy",
+ ),
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED": (
+ "test_pr7_nbcc_is_unsupported_for_existing_bounded_indexes",
+ "testPinnedNbccRejectsCowAndNonSimpleBucketIndex",
+ ),
}
SHA1_PATTERN = re.compile(r"[0-9a-f]{40}")
SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
@@ -555,7 +637,8 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
"schema_version": EXPECTED_SCHEMA_VERSION,
"baseline_id": EXPECTED_BASELINE_ID,
"validation_surface": (
-
"flink-sql-sink-existing-table-reconciliation-and-catalog-composition"
+ "flink-sql-sink-existing-table-reconciliation-catalog-composition-"
+ "and-multi-writer-contract"
),
}
for key, expected in expected_scalars.items():
@@ -791,6 +874,12 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
f"catalog_option_types[{index}] references unknown source
anchors: {unknown}"
)
+ multi_writer_contract = manifest.get("multi_writer_contract")
+ if multi_writer_contract != EXPECTED_MULTI_WRITER_CONTRACT:
+ errors.append(
+ "multi_writer_contract does not match the bounded PR7 assessment
surface"
+ )
+
physical_types = manifest.get("physical_types")
if not isinstance(physical_types, dict):
errors.append("physical_types must be a table")
@@ -877,6 +966,7 @@ def validation_evidence(
"catalog_contract_schema": manifest["catalog_composition_contract"][
"contract_schema"
],
+ "writer_contract_schema":
manifest["multi_writer_contract"]["contract_schema"],
"catalog_profile": catalog_profile,
"design_contract_schema": (
manifest["executable_contract"]["contract_schema"]
diff --git a/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
b/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
index 14310e851594..88f43649bc9a 100644
--- a/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
+++ b/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
@@ -33,6 +33,7 @@ from pathlib import Path
from typing import Any, TypeGuard
from urllib.parse import SplitResult, parse_qsl, unquote, urlsplit
+from flink_writer_contract import assess_writer_contract
from validate_flink_capabilities import (
DEFAULT_MANIFEST,
load_manifest,
@@ -153,6 +154,11 @@ FINDING_STATUS = {
"FLINK_VERSION_REQUIRED": "INCOMPLETE",
"FLINK_VERSION_UNVERIFIED": "REVIEW_REQUIRED",
"FLINK_WRITER_MODEL_UNRESOLVED": "REVIEW_REQUIRED",
+ "FLINK_WRITER_CONTRACT_REQUIRED": "INCOMPLETE",
+ "FLINK_WRITER_CONTRACT_INVALID": "INCOMPLETE",
+ "FLINK_WRITER_OWNERSHIP_UNRESOLVED": "REVIEW_REQUIRED",
+ "FLINK_WRITER_CONTRACT_CONFLICT": "REVIEW_REQUIRED",
+ "FLINK_WRITER_COMBINATION_UNSUPPORTED": "BLOCKED",
"FLINK_PR2_WRITE_PATH_UNSUPPORTED": "BLOCKED",
"FLINK_PR3_WRITE_PATH_UNSUPPORTED": "BLOCKED",
"FLINK_PR4_WRITE_PATH_UNSUPPORTED": "BLOCKED",
@@ -1220,6 +1226,7 @@ def assess_design(
"table",
"table_services",
"write",
+ "writer_contract",
},
"contract",
assessment,
@@ -1303,13 +1310,14 @@ def assess_design(
writer_model = safety.get("writer_model")
if writer_model is None:
assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is
required")
- elif writer_model == "multi_writer":
- assessment.add(
- "FLINK_MULTI_WRITER_REVIEW",
- "The executable contracts support only a confirmed single writer",
- )
- elif writer_model != "single_writer":
+ elif not isinstance(writer_model, str) or writer_model not in {
+ "single_writer",
+ "multi_writer",
+ }:
assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is
unresolved")
+ writer_contract_state = assess_writer_contract(contract)
+ for finding in writer_contract_state["findings"]:
+ assessment.add(finding["code"], finding["message"])
external_catalog = safety.get("external_catalog")
if type(external_catalog) is not bool:
assessment.add("FLINK_CATALOG_REQUIREMENT_UNRESOLVED", "Catalog
requirement is unresolved")
@@ -2055,6 +2063,13 @@ def assess_design(
"effective_configuration"
],
"catalog_deployment_checks": catalog_state["deployment_checks"],
+ "writer_contract_result":
writer_contract_state["writer_contract_result"],
+ "multi_writer_capability":
writer_contract_state["multi_writer_capability"],
+ "normalized_writer_contract": writer_contract_state[
+ "normalized_writer_contract"
+ ],
+ "writer_capability_notes":
writer_contract_state["writer_capability_notes"],
+ "writer_deployment_checks":
writer_contract_state["writer_deployment_checks"],
"finding_codes": [finding["code"] for finding in assessment.findings],
"findings": assessment.findings,
"advisory_codes": assessment.advisories,
diff --git
a/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
b/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
index 5be76ab0eb67..216bc66fd3f9 100644
--- a/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
+++ b/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
@@ -34,6 +34,11 @@ from datetime import datetime
from pathlib import Path
from typing import Any, TypeGuard
+from flink_writer_contract import (
+ CONCURRENCY_MODES,
+ LOCK_CONFIGURATION_CONSISTENCY,
+ assess_writer_contract,
+)
from validate_flink_capabilities import (
DEFAULT_MANIFEST,
load_manifest,
@@ -90,14 +95,19 @@ ALLOWED_EVIDENCE_FACTS = {
"index_global",
"index_state_ttl_days",
"index_type",
+ "lock_configuration_consistency",
+ "lock_provider_class",
"operation",
"timestamp_key_generator",
+ "write_concurrency_mode",
},
"timeline_summary": {"pending_instants"},
"writer_inventory": {
+ "active_writers",
"concurrent_writers",
"previous_writer_engine",
"previous_writer_stopped",
+ "schema_evolution_owner_id",
},
}
REQUIRED_EVIDENCE_KINDS = {
@@ -120,6 +130,7 @@ REQUIRED_TABLE_FACTS = {
"timeline_layout_version",
}
LIST_FACTS = {
+ "active_writers",
"ordering_fields",
"partition_fields",
"pending_instants",
@@ -134,6 +145,8 @@ UPPERCASE_FACTS = {
"previous_writer_engine",
"record_merge_mode",
"table_type",
+ "write_concurrency_mode",
+ "lock_configuration_consistency",
}
KNOWN_KEY_GENERATORS = {
"AUTO_RECORD",
@@ -268,6 +281,8 @@ def _normalize_fact(
if name == "partition_fields":
# HoodieTableConfig stores custom partition-key types after a
colon.
return [item.split(":", 1)[0] for item in values]
+ if name == "active_writers":
+ return sorted(values)
return values
if name == "columns":
return _normalize_columns(value, field, assessment)
@@ -325,7 +340,19 @@ def _normalize_fact(
f"{field} must be a credential-free path",
)
return None
- return normalized.upper() if name in UPPERCASE_FACTS else normalized
+ normalized = normalized.upper() if name in UPPERCASE_FACTS else normalized
+ enum_values = {
+ "write_concurrency_mode": CONCURRENCY_MODES,
+ "lock_configuration_consistency": LOCK_CONFIGURATION_CONSISTENCY,
+ }
+ allowed_enum_values = enum_values.get(name)
+ if allowed_enum_values is not None and normalized not in
allowed_enum_values:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_INVALID",
+ f"{field} must be one of {sorted(allowed_enum_values)}",
+ )
+ return None
+ return normalized
def _parse_evidence(
@@ -342,6 +369,7 @@ def _parse_evidence(
evidence_ids: set[str] = set()
collected: dict[tuple[str, str], list[tuple[Any, str]]] = {}
in_scope_ids: list[str] = []
+ conflicting_facts: set[str] = set()
for index, raw_item in enumerate(raw_evidence):
field = f"evidence[{index}]"
if not isinstance(raw_item, dict):
@@ -449,12 +477,16 @@ def _parse_evidence(
"FLINK_EXISTING_EVIDENCE_CONFLICT",
f"Evidence sources {', '.join(source_ids)} disagree about
{kind}.{name}",
)
+ conflicting_facts.add(f"{kind}.{name}")
continue
observations.setdefault(kind, {})[name] = Observation(first_value,
source_ids)
if target_table_path is not None and in_scope_ids:
observations.setdefault("evidence_scope", {})["table_path"] =
Observation(
target_table_path, tuple(in_scope_ids)
)
+ observations["evidence_conflicts"] = {
+ name: Observation(True, ()) for name in sorted(conflicting_facts)
+ }
return observations, summaries
@@ -763,6 +795,21 @@ def assess_reconciliation(
observations, evidence_summary = _parse_evidence(
envelope.get("evidence"), target_table_path, assessment
)
+ writer_evidence: dict[str, Any] = {}
+ writer_evidence_source_ids: dict[str, set[str]] = {}
+ for kind in ("writer_inventory", "flink_writer_config"):
+ for name, observation in observations.get(kind, {}).items():
+ field = f"{kind}.{name}"
+ writer_evidence[field] = observation.value
+ writer_evidence_source_ids[field] = set(observation.source_ids)
+ writer_contract_state = assess_writer_contract(
+ desired,
+ evidence=writer_evidence,
+ conflicting_evidence_fields=set(observations.get("evidence_conflicts",
{})),
+ evidence_source_ids=writer_evidence_source_ids,
+ )
+ for finding in writer_contract_state["findings"]:
+ assessment.add(finding["code"], finding["message"])
for kind in sorted(REQUIRED_EVIDENCE_KINDS - set(observations)):
assessment.add(
"FLINK_EXISTING_EVIDENCE_REQUIRED", f"{kind} evidence is required"
@@ -951,15 +998,176 @@ def assess_reconciliation(
concurrent = inventory.get("concurrent_writers")
stopped = inventory.get("previous_writer_stopped")
previous_engine = inventory.get("previous_writer_engine")
- _add_comparison(
- comparisons,
- "writer_inventory.concurrent_writers",
- False,
- concurrent,
- mismatch_code="FLINK_MULTI_WRITER_REVIEW",
- mismatch_result="UNVERIFIED",
- assessment=assessment,
+ raw_safety = desired.get("safety")
+ safety = raw_safety if isinstance(raw_safety, dict) else {}
+ writer_model = safety.get("writer_model")
+ if writer_model == "single_writer":
+ _add_comparison(
+ comparisons,
+ "writer_inventory.concurrent_writers",
+ False,
+ concurrent,
+ mismatch_code="FLINK_MULTI_WRITER_REVIEW",
+ mismatch_result="UNVERIFIED",
+ assessment=assessment,
+ )
+ elif concurrent is None:
+ comparisons.append(
+ {
+ "desired": None,
+ "field": "writer_inventory.concurrent_writers",
+ "observed": None,
+ "result": "UNVERIFIED",
+ "source_ids": [],
+ }
+ )
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED",
+ "Evidence is required for writer_inventory.concurrent_writers",
+ )
+ else:
+ comparisons.append(
+ {
+ "desired": None,
+ "field": "writer_inventory.concurrent_writers",
+ "observed": concurrent.value,
+ "result": "UNVERIFIED",
+ "source_ids": list(concurrent.source_ids),
+ }
+ )
+ active_writers = inventory.get("active_writers")
+ if writer_model == "single_writer" and active_writers is not None:
+ comparisons.append(
+ {
+ "desired": {"maximum_count": 1},
+ "field": "writer_inventory.active_writers",
+ "observed": active_writers.value,
+ "result": (
+ "MATCH" if len(active_writers.value) <= 1 else "UNVERIFIED"
+ ),
+ "source_ids": list(active_writers.source_ids),
+ }
+ )
+ normalized_writer_contract =
writer_contract_state["normalized_writer_contract"]
+ if normalized_writer_contract:
+ declared_writer_ids = sorted(
+ [normalized_writer_contract["primary_writer_id"]]
+ + [
+ writer["writer_id"]
+ for writer in normalized_writer_contract["additional_writers"]
+ ]
+ )
+ if active_writers is not None:
+ active_writer_result = (
+ "MATCH"
+ if set(active_writers.value) <= set(declared_writer_ids)
+ else "UNVERIFIED"
+ )
+ comparisons.append(
+ {
+ "desired": {"allowed_writer_ids": declared_writer_ids},
+ "field": "writer_inventory.active_writers",
+ "observed": active_writers.value,
+ "result": active_writer_result,
+ "source_ids": list(active_writers.source_ids),
+ }
+ )
+ if active_writer_result != "MATCH":
+ assessment.add(
+ "FLINK_WRITER_CONTRACT_CONFLICT",
+ "Existing-table value differs at
writer_inventory.active_writers",
+ )
+ schema_owner = inventory.get("schema_evolution_owner_id")
+ if schema_owner is not None:
+ _add_comparison(
+ comparisons,
+ "writer_inventory.schema_evolution_owner_id",
+
normalized_writer_contract["ownership"]["schema_evolution_owner_id"],
+ schema_owner,
+ mismatch_code="FLINK_WRITER_CONTRACT_CONFLICT",
+ mismatch_result="UNVERIFIED",
+ assessment=assessment,
+ )
+ elif writer_model != "single_writer" and active_writers is not None:
+ comparisons.append(
+ {
+ "desired": None,
+ "field": "writer_inventory.active_writers",
+ "observed": active_writers.value,
+ "result": "UNVERIFIED",
+ "source_ids": list(active_writers.source_ids),
+ }
+ )
+
+ observed_concurrency = (
+ normalized_writer_contract.get("observed_concurrency")
+ if normalized_writer_contract
+ else None
)
+ config_observations = observations.get("flink_writer_config", {})
+ concurrency_expectations = {
+ "writer_config.write_concurrency_mode": (
+ observed_concurrency.get("mode")
+ if isinstance(observed_concurrency, dict)
+ else None,
+ config_observations.get("write_concurrency_mode"),
+ ),
+ "writer_config.lock_provider_class": (
+ observed_concurrency.get("lock_provider_class")
+ if isinstance(observed_concurrency, dict)
+ else None,
+ config_observations.get("lock_provider_class"),
+ ),
+ "writer_config.lock_configuration_consistency": (
+ observed_concurrency.get("lock_configuration_consistency")
+ if isinstance(observed_concurrency, dict)
+ else None,
+ config_observations.get("lock_configuration_consistency"),
+ ),
+ }
+ for field, (expected, observed) in concurrency_expectations.items():
+ if observed is None:
+ continue
+ if writer_model == "multi_writer" and expected is not None:
+ _add_comparison(
+ comparisons,
+ field,
+ expected,
+ observed,
+ mismatch_code="FLINK_WRITER_CONTRACT_CONFLICT",
+ mismatch_result="UNVERIFIED",
+ assessment=assessment,
+ )
+ elif writer_model == "single_writer" and field.endswith(
+ "write_concurrency_mode"
+ ):
+ _add_comparison(
+ comparisons,
+ field,
+ "SINGLE_WRITER",
+ observed,
+ mismatch_code="FLINK_MULTI_WRITER_REVIEW",
+ mismatch_result="UNVERIFIED",
+ assessment=assessment,
+ )
+ else:
+ result = "UNVERIFIED"
+ if writer_model == "single_writer":
+ result = (
+ "UNVERIFIED"
+ if field.endswith("lock_configuration_consistency")
+ and observed.value == "CONFLICTING"
+ else "NOT_APPLICABLE"
+ )
+ comparisons.append(
+ {
+ "desired": None,
+ "field": field,
+ "observed": observed.value,
+ "result": result,
+ "source_ids": list(observed.source_ids),
+ }
+ )
if transition in {"same_job_restore", "cold_start_new_job",
"flink_writer_takeover"}:
_add_comparison(
comparisons,
@@ -1073,10 +1281,14 @@ def assess_reconciliation(
change_disposition = "DEPLOYMENT_CONFIGURATION_CHANGE"
else:
change_disposition = "UNVERIFIED"
- deployment_checks = [
- {"check": "confirm_single_writer", "required": True},
- {"check": "recheck_pending_instants", "required": True},
- ]
+ deployment_checks = []
+ if writer_model == "single_writer":
+ deployment_checks.append(
+ {"check": "confirm_single_writer", "required": True}
+ )
+ deployment_checks.append(
+ {"check": "recheck_pending_instants", "required": True}
+ )
if transition == "cold_start_new_job" and desired.get("contract_schema")
in (2, 3):
deployment_checks.append(
{"check": "verify_index_bootstrap_before_first_checkpoint",
"required": True}
@@ -1102,6 +1314,13 @@ def assess_reconciliation(
"catalog_deployment_checks": desired_result.get(
"catalog_deployment_checks", []
),
+ "writer_contract_result":
writer_contract_state["writer_contract_result"],
+ "multi_writer_capability":
writer_contract_state["multi_writer_capability"],
+ "normalized_writer_contract": writer_contract_state[
+ "normalized_writer_contract"
+ ],
+ "writer_capability_notes":
writer_contract_state["writer_capability_notes"],
+ "writer_deployment_checks":
writer_contract_state["writer_deployment_checks"],
"finding_codes": finding_codes,
"findings": assessment.findings,
"advisory_codes": assessment.advisories,
