Oracle-to-Hive Loading Without Sqoop (3) — Guaranteeing "Exactly Once" in a Distributed Flow with the Load Control API

By the end of Part 2, each partition reads data at the same SCN and writes it to HDFS as Parquet. What remains are the classic questions of distributed systems.

• Who decides that all partitions running in parallel on several nodes have finished?
• If the last two partitions finish at the same time, does the next step (validation) start twice?
• If we send the publish (INSERT OVERWRITE) and the response is lost, did it succeed or fail?

This post covers the design of the Load Control API (Python FastAPI + PostgreSQL), built to answer these questions.

1. The API does not move data

The Load Control API does not move a single row itself. It only records and decides state in a ledger in the PostgreSQL nifi_ops schema.

The ledger has six tables.

Even with several servers and workers running, contention is controlled by DB locks, leases and advisory locks. Most mechanisms in this post are variations of a conditional UPDATE (CAS): "change to Y only when the state is X."

2. Run state model

A run goes through the following states.

The first mechanism sits at run creation. Only one active run can exist for the same job_key + business_key. This is enforced by a PostgreSQL partial unique index, so even if the trigger fires twice for the same business date, the second request gets 409 DUPLICATE_ACTIVE_RUN and the existing run is unaffected. PUBLISH_UNKNOWN is also treated as an active run until an operator resolves it, so it blocks new runs.

3. Partition ownership: claim token

Partitions registered by PG-10 are spread across NiFi nodes by Round Robin. Because of resends or reissues, the same partition FlowFile can arrive at two nodes at the same time. So PG-20 acquires ownership first, before querying Oracle.

{ "claimToken": "a-worker-generated-uuid", "workerNode": "nifi-01" }

{ "claimed": true, "runStatus": "EXTRACTING", "partitionStatus": "RUNNING", "attempt": 1 }

• A claim is a PENDING/RETRY → RUNNING CAS. Only one request succeeds.
• A repeat request with the same token is treated as a retry after a lost response, and it succeeds again (idempotent).
• If another worker is already processing the partition or the run has ended, the response is claimed=false. This is not an error but normal contention, so NiFi does not query Oracle and quietly ends the FlowFile.

All later chunk reports and failure reports carry this token. Reports from anyone who is not the owner are rejected.

4. Chunk reports and completion decision

PG-20 reports to the API each time it writes one Parquet chunk to HDFS.

{
  "claimToken": "a-worker-generated-uuid",
  "chunkIndex": 2,
  "chunkCount": 3,
  "hdfsPath": "/data/nifi/stage/JOB/run_id=.../part-0000-000002.parquet",
  "recordCount": 5000,
  "byteCount": 183552
}

{
  "recorded": true,
  "partitionStatus": "SUCCESS",
  "runStatus": "EXTRACTED_VALIDATED",
  "receivedChunks": 3,
  "chunkCount": 3,
  "validationScheduled": true
}

On every report, the API checks the following.

• Does the claim token match the current owner?
• Is chunkIndex < chunkCount?
• Is the HDFS path directly under the run path, with no ..?
• Has different content arrived for a partition that has already succeeded?
• Are all chunk indexes 0..n-1 present with none missing, and did they all report the same chunkCount?
• Does the sum of counts equal the partition's expected count?
• Does the total across all partitions equal the source count?

The one last report that satisfies every condition moves the run to EXTRACTED_VALIDATED. To handle the case where the last two partitions finish at the same time, the API locks the run row with FOR UPDATE and runs the completion CAS, so only one of the two reports transitions the run.

This final decision and the reservation of the NiFi call "start validation" (dispatch INSERT) happen in the same transaction. The decision cannot be made while the call reservation is lost, and a call cannot be reserved while the decision is rolled back. This is the transactional outbox in the next section.

5. Outbox and ACK: calls at least once, execution exactly once

The API server does not call NiFi directly. It writes the call to the load_dispatch table, and the worker's dispatcher picks it up and sends it to PG-05 Control Receiver at the NiFi root.

There is an important distinction here.

• 202 is only a receipt, not a processing ACK. PG-05 checks only the request format and returns 202 immediately. The real ACK is when PG-40 calls /validation/start. If the NiFi node dies right after the 202, the sweeper resends the dispatch after ack_timeout (default 10 minutes).
• The dispatcher takes a short lease with FOR UPDATE SKIP LOCKED and makes the HTTP call after closing the transaction. If the worker dies while sending, another worker picks it up after the lease expires.
• So delivery is guaranteed at least once. The same validation call can arrive twice, and duplicates are removed by CAS on the receiving side.

6. Validation start is also a CAS

PG-40 notifies the API before starting validation. Only the first call gets started=true. If the same dispatch arrives again, the second and later ones end quietly inside PG-40.

{
  "started": true,
  "runStatus": "STAGE_VALIDATING",
  "snapshotScn": "2388556",
  "hdfsRunPath": "...",
  "stageTable": "...",
  "sourceCount": 105000,
  "extractedCount": 105000,
  "sourceMetrics": { "AMOUNT_SUM": "71853075", "MIN_TS": "...", "MAX_TS": "..." }
}

There is a reason the response includes the HDFS path, the staging table and the source expected values. The validation FlowFile starts from a new request created by the API worker, so it carries none of PG-10's attributes. Every value it needs is fetched again from the ledger.

PG-40 creates a Hive external table with the run path as its LOCATION, then computes and reports the following metrics.

NiFi computes PASS/FAIL, but the API does not trust that verdict as is. When it receives a stage-validated request, it reads the stored metrics again and transitions to STAGING_VALIDATED only if there is no FAIL and the metrics are not empty. PG-60's target validation and the final SUCCESS work the same way.

7. Publishing: publish token and PUBLISH_UNKNOWN

Publishing (PG-50) is a step that is hard to undo, because it replaces the target partition with INSERT OVERWRITE.

1. PG-50 creates a UUID publish token and calls /publish/claim. Only the one token that wins the STAGING_VALIDATED → PUBLISHING CAS runs the publish SQL.
2. INSERT OVERWRITE is not retried automatically.
3. The result is reported to /publish/result. It is one of PUBLISHED, FAILED_PUBLISH or PUBLISH_UNKNOWN.

Why failures are reported as "unknown"

With PutClouderaHiveQL, which we use for Hive publishing, the failure relationship alone cannot tell whether the SQL was never submitted or was cut off during execution. Our tests also confirmed that it leaves no error attribute on failure. If execution was cut off, the target may already have changed. Republishing automatically in that state makes it even harder to know what the correct result is.

So this project reports everything except a confirmed success as PUBLISH_UNKNOWN. A run that stays in PUBLISHING too long is also moved to PUBLISH_UNKNOWN by the sweeper after publish_stale (default 2 hours). PUBLISH_UNKNOWN is not a terminal state, and it blocks new runs for the same business date until an operator resolves it.

The operator's resolution procedure is as follows.

1. Stop the job's trigger and any manual reruns.
2. Get the run ID, the publish start time and the Hive query ID.
3. Check in the Hive query history whether it reached compile/submit/execute/commit.
4. Compare the target partition's count and all quality metrics against source/staging.
5. Resolve as PUBLISHED if the publish definitely completed, or as FAILED_PUBLISH if nothing changed or it definitely failed.

curl -X POST \
  -H 'Authorization: Bearer <operator-token>' \
  -H 'Content-Type: application/json' \
  -d '{"resolution":"FAILED_PUBLISH","reason":"Failed at compile stage; confirmed target unchanged"}' \
  <API>/v1/runs/<run_id>/publish-unknown/resolve

Record the basis for the decision in the resolution reason. If it is ambiguous, do not mark it as a success on a guess; hand it to target recovery or a business approval process. If you resolve it as PUBLISHED, PG-60 does not resume automatically, so carry out target validation and follow-up steps according to the operating procedure.

Note that the case where HiveServer2 cannot be reached at all is different. In that case the processor returns the FlowFile to its input queue and keeps retrying. Since the SQL has not been submitted yet, processing continues once Hive recovers. In our tests, while HiveServer2 was stopped the flow waited at the DDL step, and after the restart it went on to SUCCESS.

8. Failure reports and HTTP response rules

Failures from every PG are collected in PG-90 and normalized into the same format. How they are reported depends on the stage (load.stage).

A run failure report sends expectedStatus and failStatus together, and only allowed combinations are accepted. For example, trying to set a run that is not in STAGE_VALIDATING to FAILED_STAGE_VALIDATION returns 422 FAIL_TRANSITION_NOT_ALLOWED.

NiFi and the worker behave differently for each HTTP status.

Not every 409 is harmless. DUPLICATE_ACTIVE_RUN and CLAIM_MISMATCH right after a reissue are normal contention, but CHUNK_CONFLICT means the report for a file that already succeeded has changed, so it must be investigated.

9. Sweeper: the part that works when nobody asks

The API server only acts when a request arrives. But if a NiFi node dies or a FlowFile disappears, the request itself never comes. Finding these stalls is the job of the worker's sweeper.

stale must always be larger than PG-20's query timeout. Otherwise a query that is legitimately slow is mistaken for a stalled one.

REISSUE: resending a partition with the same SCN

The default recovery mode is FAIL (if a partition stalls, the run fails and a new run is started). If you switch to recovery.mode=REISSUE, only the stalled partitions are resent in the same run, with the same SCN.

A reissued partition regains ownership with a new claim token. Even if the original attempt comes back to life late and sends reports, the token differs, so they are rejected with 409 CLAIM_MISMATCH. The maximum number of claims per partition is max_attempts (default 3). Part 2 said that "re-reading only the failed partitions with a new SCN is forbidden." REISSUE uses the original SCN as is, so it does not conflict with that rule.

10. Failure scenarios confirmed by testing

These are the results from the Part 1 test environment (CFM 4.12 / NiFi 2.6.0 with 2 nodes, Oracle 23ai, Hive 4.0.1, PostgreSQL 16) that relate to the mechanisms in this post.

The chunk report failure scenario is a bug that testing found and we fixed. PG-90 reports a partition failure only for "a failure from a Worker whose claim succeeded" (the table in section 8). But when the chunk report (38_Report_Chunk) failed, the error response body overwrote the attribute that held the claim response, so PG-90 could not tell whether this Worker was the owner. As a result the failure was recorded only as an event, and the partition stayed in EXTRACTING until the sweeper's stale threshold (90 minutes). After we changed the flow to keep the claim result in a separate attribute (partition.claimed), report failures are also reported as partition failures, and the run ends as failed within 2 seconds.

The Load Control API's own test suite passed 139 of 140 tests. It covers concurrent completion, duplicate reports, concurrent claims, dispatcher contention, deadlock retries, TUI screens and operator tasks. The remaining test counts LISTEN connections across the whole server, so it was excluded in the test environment, where other workers were running on the same DB server.

11. Operations: what to watch, and what not to do

On the API host, bin/monitor.sh opens a terminal dashboard (TUI) that shows active runs, partition progress, validation metrics, the dispatch backlog and alerts. The recommended alerts are as follows.

The reprocessing rules can be summarized briefly.

• For manifest, extraction and staging validation failures, fix the cause and rerun everything as a new run.
• Do not rerun only some partitions with a new SCN. The only way to resend with the same SCN is REISSUE.
• For publish and target validation failures, do not republish automatically.
• PUBLISH_UNKNOWN must be resolved first before a new run can be created.
• Resend a DEAD dispatch only after fixing the cause (PG-05 port, route, job PG state). Do not keep resending without fixing the cause.
• HDFS files and staging tables of failed runs are kept as evidence for investigation during the retention period (default: 3 days for success, 14 days for failure).

Wrapping up

The conclusion of this series comes down to one point. In a distributed flow, "exactly once" is not achieved by sending a message only once. It is achieved by combining at-least-once delivery with conditional state transitions on the receiving side. Partition claims, the final completion decision, validation start and publish claims all follow the same pattern. And at the one point where the result cannot be known for certain (publishing), we chose to stop and ask a person rather than automate.

The NiFi flow only moves data and calls the API, and decisions are made by a small API on top of a PostgreSQL ledger. The core of this project is taking the completion decision that was hidden inside a single Sqoop run and making it explicit and external. The full code and manual are in the nifi-sqoop-removal repository.