Verify no data loss when catalog commit conflicts occur. When a CatalogCommitConflicts error occurs during an Iceberg commit, the sink must retry the commit so that no data is lost. Strategy: Run a background thread that modifies the Polaris table metadata (adding dummy snapshots)
(c: Composition)
| 128 | |
| 129 | |
| 130 | def workflow_commit_conflict(c: Composition) -> None: |
| 131 | """Verify no data loss when catalog commit conflicts occur. |
| 132 | |
| 133 | When a CatalogCommitConflicts error occurs during an Iceberg commit, |
| 134 | the sink must retry the commit so that no data is lost. |
| 135 | |
| 136 | Strategy: Run a background thread that modifies the Polaris table |
| 137 | metadata (adding dummy snapshots) to race with the sink's commits. |
| 138 | When a modification lands between the sink's table refresh and its |
| 139 | commit POST, the sink gets a CatalogCommitConflicts error. |
| 140 | |
| 141 | Verification uses DuckDB's iceberg_scan to count the rows in the |
| 142 | Iceberg table. All inserted rows must be present. |
| 143 | """ |
| 144 | key = _setup(c) |
| 145 | |
| 146 | # Phase 1: Create sink with short commit interval (2s) and initial data |
| 147 | c.run_testdrive_files( |
| 148 | f"--var=s3-access-key={key}", |
| 149 | "--var=aws-endpoint=minio:9000", |
| 150 | "commit-conflict-setup.td", |
| 151 | ) |
| 152 | |
| 153 | # Phase 2: Wait for initial snapshot batch to commit |
| 154 | print("Waiting 10s for initial snapshot batch to commit...") |
| 155 | time.sleep(10) |
| 156 | |
| 157 | # Phase 3: Set up direct HTTP access to Polaris from host |
| 158 | polaris_port = c.port("polaris", 8181) |
| 159 | base_url = f"http://localhost:{polaris_port}" |
| 160 | table_url = ( |
| 161 | f"{base_url}/api/catalog/v1/default_catalog" |
| 162 | f"/namespaces/default_namespace/tables/conflict_table" |
| 163 | ) |
| 164 | |
| 165 | # Get access token via direct HTTP |
| 166 | token_req = urllib.request.Request( |
| 167 | f"{base_url}/api/catalog/v1/oauth/tokens", |
| 168 | data=b"grant_type=client_credentials&client_id=root&client_secret=root&scope=PRINCIPAL_ROLE:ALL", |
| 169 | headers={"Content-Type": "application/x-www-form-urlencoded"}, |
| 170 | ) |
| 171 | token_resp = urllib.request.urlopen(token_req) |
| 172 | access_token = json.loads(token_resp.read())["access_token"] |
| 173 | print(f"Got Polaris access token (len={len(access_token)})") |
| 174 | |
| 175 | # Phase 4: Start background modification loop (~100/sec). |
| 176 | # This rate triggers CatalogCommitConflicts on ~50% of sink commits. |
| 177 | stop_event = threading.Event() |
| 178 | stats = {"modifications": 0, "self_conflicts": 0, "errors": 0} |
| 179 | |
| 180 | def modify_table_loop() -> None: |
| 181 | """Add dummy snapshots to race with the sink's commits.""" |
| 182 | while not stop_event.is_set(): |
| 183 | try: |
| 184 | data = _polaris_get(table_url, access_token) |
| 185 | metadata = data["metadata"] |
| 186 | snap_id = metadata["current-snapshot-id"] |
| 187 |