Flink + Paimon End-to-End Exactly-Once: Production Implementation Guide

Background and Scope

The article originates from a real-world scenario where a new developer assumed that a Flink job configured with a 10-second or 10,000-record flush interval to Paimon would yield sub-second latency. The author explains that the job's checkpoint interval was set to 3 minutes, and that data visibility in Flink + Paimon is bound to the checkpoint cycle, not the flush trigger. The guide applies to Apache Paimon 1.x and Apache Flink 1.17–1.20, covering OSS/HDFS storage in lakehouse streaming write scenarios. It serves as a production operations manual defining configuration standards, acceptance criteria, troubleshooting SOPs, and operational requirements for core data pipeline consistency.

1. Semantic Definitions and Technical Boundaries

1.1 Consistency Semantics Tiers and Applicability

Distributed data pipelines classify consistency into three tiers by reliability (low to high): At-Most-Once (possible loss), At-Least-Once (possible duplicates), and Exactly-Once (no loss, no duplicates). Selection must match business fault-tolerance cost, not blindly pursue the highest tier. In Flink + Paimon, Exactly-Once delivers three core values: trustworthy data metrics (no extra reconciliation), fault traceability (state rolls back to latest consistent snapshot), and downstream semantic compatibility (StarRocks, real-time compute consume consistent changelogs).

1.2 Hard Constraints

Latency floor: Data visibility is tightly coupled to checkpoint period; cannot achieve sub-second visibility beyond checkpoint interval.

Overhead cost: Barrier alignment, state persistence, and two-phase commit introduce 10–30% write performance penalty.

Scope boundary: Guarantees only Flink → Paimon single-link write consistency; cross-table transactions and end-to-end upstream/downstream consistency require additional solutions.

1.3 Common Misconceptions Corrected

Misconception: Paimon data visibility requires compaction completion. Correction: Visibility is decided solely by snapshot commit; checkpoint completion atomically generates a new snapshot, making data immediately readable. Compaction is an async background optimization affecting only query performance.

Misconception: Primary key table idempotence equals Exactly-Once. Correction: Upsert idempotence only solves duplicate-write eventual consistency, cannot prevent data loss. Exactly-Once includes "no loss, no duplicate" and relies on checkpoint + 2PC full-chain coordination.

Misconception: 30s checkpoint inevitably destabilizes jobs. Correction: With incremental checkpoint + RocksDB state backend, 30s is a standard production setting; timeout risk appears only when single TaskManager state exceeds GB scale or small files proliferate.

2. Core Implementation Mechanism: Three-Layer Coordination

Exactly-Once in Flink + Paimon results from deep collaboration of three layers: Flink state consistency, Paimon snapshot atomicity, and two-phase commit protocol.

2.1 Layer 1: Flink Checkpoint Distributed State Consistency

Based on an improved Chandy-Lamport distributed snapshot algorithm, using checkpoint barriers for cross-operator state alignment:

JobManager injects barriers into all sources at configured intervals.

Barriers flow downstream; multi-input operators perform barrier alignment, blocking early streams until late barriers arrive.

All operators complete state snapshots and persist to state backend.

JobManager collects acknowledgments and marks checkpoint globally successful.

Core guarantee: All operator states within the same checkpoint period are strictly consistent, providing a globally consistent time boundary for downstream sink commits.

2.2 Layer 2: Paimon Snapshot Atomic Commit Mechanism

Paimon uses LSM-tree architecture + snapshot versioning for write atomicity, centered on atomic snapshot pointer switching :

Each write generates new data files and manifest metadata; no committed files are modified.

Commit updates only the snapshot root pointer to a new manifest list — a single atomic operation.

On commit failure, written temp files auto-expire, leaving original snapshot consistency intact.

Core guarantee: Write operations either fully take effect or fully roll back; no intermediate visible state exists, forming the storage-layer foundation for 2PC.

2.3 Layer 3: Two-Phase Commit (2PC) End-to-End Coordination

Paimon Flink Sink implements standard 2PC via TwoPhaseCommitSinkFunction, strictly binding Flink state with storage commit. The process (illustrated in the article's diagram) coordinates pre-commit and commit phases across the checkpoint barrier.

2.4 Consistency Logic During Failure Recovery

When a job recovers from checkpoint/savepoint, the sink executes:

Load pre-commit handles from Flink state.

Verify global commit status of the corresponding checkpoint.

If checkpoint globally succeeded: execute formal commit to ensure no data loss.

If checkpoint not globally confirmed: clean temp data, roll back to consistent state, avoiding duplicate writes.

3. Production-Grade Standardized Configuration

3.1 Flink Mandatory Compliance Config (Core Pipeline Required)

-- ===== Runtime Mode =====
SET 'execution.runtime-mode' = 'streaming';
-- ===== Checkpoint Core Semantics =====
SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
SET 'execution.checkpointing.interval' = '120s';          -- Core pipeline recommended 60~300s
SET 'execution.checkpointing.timeout' = '600s';           -- Must be 3~5x interval
SET 'execution.checkpointing.max-concurrent-checkpoints' = '1';
SET 'execution.checkpointing.min-pause' = '60s';          -- Avoid checkpoint stacking
SET 'execution.checkpointing.tolerable-failed-checkpoints' = '3';
-- ===== State Backend & Persistence =====
SET 'state.backend' = 'rocksdb';
SET 'state.backend.incremental' = 'true';                 -- Mandatory for large state
SET 'state.checkpoints.dir' = 'oss://your-bucket/flink/checkpoints/xxx_job';
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';
-- ===== Fault-Tolerant Restart Strategy =====
SET 'restart-strategy' = 'fixed-delay';
SET 'restart-strategy.fixed-delay.attempts' = '3';
SET 'restart-strategy.fixed-delay.delay' = '30s';

Mandatory rationale: max-concurrent-checkpoints must be 1; concurrent checkpoints cause 2PC state confusion, breaking consistency.

Externalized checkpoints must be retained for failure recovery after abnormal job termination.

Tolerable failed checkpoints cannot be 0; transient storage jitter should not trigger immediate job restart.

3.2 Flink Performance Optimization Config (Optional)

-- RocksDB tuning
SET 'state.backend.rocksdb.memory.write-buffer-ratio' = '0.5';
SET 'state.backend.rocksdb.checkpoint.transfer.thread.num' = '4';
-- Alignment timeout optimization (avoid checkpoint blocking under extreme skew)
SET 'execution.checkpointing.alignment-timeout' = '60s';
SET 'execution.checkpointing.unaligned' = 'true';         -- Enable under high backpressure

3.3 Paimon Table-Level Consistency Config Standards

Primary Key Table (CDC Sync / Upsert Standard Template)

For business tables with updates/deletes (orders, user profiles) — core pipeline first choice:

CREATE TABLE dwd_order_detail (
    order_id BIGINT NOT NULL,
    user_id BIGINT,
    order_amount DECIMAL(18, 2),
    order_status TINYINT,
    create_time TIMESTAMP(3),
    update_time TIMESTAMP(3),
    dt STRING,
    PRIMARY KEY (dt, order_id) NOT ENFORCED
) PARTITIONED BY (dt)
WITH (
    'bucket' = '16',
    'bucket-key' = 'order_id',
    'file.format' = 'orc',
    -- Consistency core config
    'merge-engine' = 'deduplicate',
    'changelog-producer' = 'input',                     -- CDC preferred, pass-through upstream changes
    'commit.force-create-snapshot' = 'true',
    -- Snapshot lifecycle
    'snapshot.num-retained.min' = '20',
    'snapshot.num-retained.max' = '200',
    'snapshot.time-retained' = '48h',
    -- Concurrency control
    'commit.max-retries' = '20',
    'commit.retry-interval' = '1s'
);

Append-Only Table (Log / Event Template)

For pure append, no-update log data — optimal write performance:

CREATE TABLE dwd_behavior_log (
    log_id STRING,
    user_id BIGINT,
    event_type STRING,
    event_time TIMESTAMP(3),
    dt STRING
) PARTITIONED BY (dt)
WITH (
    'bucket' = '-1',                                      -- Dynamic bucket mode
    'file.format' = 'parquet',
    'write-mode' = 'append-only',
    'changelog-producer' = 'none',
    'snapshot.num-retained.min' = '10',
    'snapshot.num-retained.max' = '100',
    'snapshot.time-retained' = '24h'
);

Changelog Mode Selection Decision Rules

The article includes a decision diagram (image) for choosing changelog modes based on business scenario.

3.4 Storage Layer Adaptation Config

For OSS object storage, optimize commit stability:

-- Table-level config, adapt to OSS high-latency characteristics
'commit.ignore-non-existing-files' = 'true',
'oss.connection.timeout' = '30s',
'oss.socket.timeout' = '60s'

4. Concurrent Write & Partition Consistency Architecture Standards

4.1 Single Table Single Writer Principle

Production core tables must follow single writer principle : one Paimon table allows only one Flink write job at a time.

Concurrent multi-job writes trigger optimistic lock commit conflicts; retry mechanism handles only low-frequency occasional conflicts, not high-frequency concurrent writes.

Violating single writer raises commit failure rate, checkpoint timeouts, and can break data consistency.

Exception: Multi-job writes isolated by partition (non-overlapping ranges) can be treated as logical single writer, provided partition ranges are strictly non-intersecting.

4.2 Multi-Business Write Isolation Solutions

When multiple business lines write to the same logical table, choose by priority:

Upstream merge (recommended): Unify multiple data streams at ingestion layer, single job writes to Paimon.

Partitioned view merge: Each business writes to independent physical tables; downstream unified view encapsulates, fully avoiding conflicts.

Partition division: Assign partition ranges per business line; each job writes only to its dedicated partitions, cross-partition writes prohibited.

4.3 Dynamic Partition Write Consistency Guarantee

Paimon ensures cross-partition atomicity via global snapshot mechanism:

All partition changes in a single commit are recorded in the same manifest list, taking effect atomically with the snapshot.

No intermediate state where some partitions are visible and others not.

New partition metadata creation and data commit complete in the same snapshot; queries never see empty partitions.

5. Consistency Acceptance & Fault Injection Testing

Core pipelines must pass the following acceptance tests before go-live to verify Exactly-Once effectiveness.

5.1 Functional Consistency Verification

Method 1: Primary Key Exact Reconciliation

Select fixed time window, compare source vs Paimon primary key details:

-- Source stats (MySQL CDC example)
SELECT COUNT(DISTINCT order_id), SUM(order_amount)
FROM source_orders
WHERE update_time BETWEEN '2026-07-06 10:00:00' AND '2026-07-06 12:00:00';
-- Paimon stats
SELECT COUNT(DISTINCT order_id), SUM(order_amount)
FROM dwd_order_detail
WHERE dt = '2026-07-06'
  AND update_time BETWEEN '2026-07-06 10:00:00' AND '2026-07-06 12:00:00';

Acceptance standard: Primary key row count and aggregated amount fully match.

Method 2: Snapshot Time Travel Verification

Use Paimon time travel to validate historical snapshot against source data at corresponding timestamp:

-- Query snapshot data at specific time
SELECT COUNT(*)
FROM dwd_order_detail
FOR SYSTEM_TIME AS OF TIMESTAMP '2026-07-06 11:00:00';

Acceptance standard: Snapshot data fully matches source data already committed at that timestamp.

5.2 Fault Injection Tests (Production Mandatory)

Test 1: TaskManager Process Abnormal Termination

Job stable; record current Paimon table baseline data volume. kill -9 randomly terminate one TaskManager process.

Wait for auto-restart recovery, stable run for 3 checkpoint cycles.

Reconcile to verify data consistency.

Acceptance: No data loss, no duplicates, job recovers and runs normally.

Test 2: Storage Network Interruption

Simulate Paimon storage (OSS/HDFS) network interruption via iptables.

Observe checkpoint failures, job fault tolerance retries.

Restore network, wait for job stable recovery.

Reconcile to verify data consistency.

Acceptance: Job auto-recovers after network restore; no data loss, no duplicates.

Test 3: Savepoint Stop-Start Recovery

Trigger savepoint, record data baseline at that point.

Stop job, wait interval, restart from savepoint.

Stable run then reconcile.

Acceptance: Data continues writing from savepoint offset; no gaps, no duplicates.

6. Performance vs Consistency Trade-off Decisions

6.1 Checkpoint Interval Selection Matrix

The article includes a matrix diagram (image) mapping business latency tolerance, state size, and throughput to recommended checkpoint intervals.

6.2 Consistency Downgrade Rules

When write performance bottlenecks persist after optimization, downgrade semantics per:

Primary key table: Downgrade to At-Least-Once, rely on primary key upsert idempotence for eventual no-duplicate; risk: transient inconsistency during failure recovery, eventually converges.

Append-only: Downgrade only after confirming downstream has deduplication capability or business tolerates minor duplicates.

6.3 High-Throughput Optimization Path (Without Downgrading Semantics)

Increase write-buffer-size to reduce file count per checkpoint.

Optimize bucketing strategy to improve write parallelism.

Enable incremental checkpoint + unaligned checkpoint to reduce barrier alignment blocking.

Moderately relax checkpoint interval to reduce commit frequency.

7. Typical Production Troubleshooting SOPs

7.1 Checkpoint Timeout Failure

Symptom: Flink UI shows persistent checkpoint timeouts, frequent job restarts.

Locate phase: Check checkpoint details, identify bottleneck operator.

If sink operator sync time longest → pre-commit slow; investigate storage I/O, small file count.

If barrier alignment time long → data skew, backpressure causing alignment block.

Investigation steps:

Check table small file ratio; if >30%, trigger compaction first.

Check storage monitoring for I/O throttling, bandwidth saturation.

Check job for data skew; per-parallelism data volume diff >5x?

Remediation:

Increase checkpoint timeout to 5x interval.

Trigger full compaction to merge small files.

Optimize bucket key to resolve data skew.

Enable unaligned checkpoint under high backpressure.

7.2 Data Duplicate Writes

Symptom: Metrics inflated, primary key table shows duplicate records.

Investigation:

Confirm checkpoint mode is EXACTLY_ONCE.

Check table primary key definition covers all unique key fields.

Confirm no concurrent multi-job writes to same table.

Use $snapshots system table to locate snapshot timestamps of duplicate data.

Remediation:

Fix checkpoint config, restore Exactly-Once mode.

Complete primary key definition, execute deduplication repair.

Split concurrent write jobs, enforce single writer principle.

7.3 Commit Conflict (CommitConflictException)

Symptom: Job logs frequent CommitConflictException, checkpoint failure rate rises.

Root cause: Multiple write jobs committing concurrently, optimistic lock conflict.

Fix priority:

Architectural split of write jobs by partition/table isolation (root cure).

Temporarily increase commit.max-retries and retry interval to mitigate low-frequency conflicts.

Stagger checkpoint trigger times across jobs to reduce commit collision probability.

7.4 Snapshot Metadata Bloat

Symptom: Queries slow, manifest file count surges, storage metadata proportion excessive.

Root cause: Overly frequent checkpoints + loose snapshot retention policy.

Remediation:

-- Tighten snapshot retention
ALTER TABLE dwd_order_detail SET (
    'snapshot.num-retained.min' = '10',
    'snapshot.num-retained.max' = '50',
    'snapshot.time-retained' = '12h'
);
-- Manual expire cleanup
CALL sys.expire_snapshots('database.dwd_order_detail');

8. Production Operations System Construction

8.1 Pre-Go-Live Checklist

Core pipelines must confirm each item before launch:

✅ Checkpoint mode EXACTLY_ONCE, interval and timeout per spec.

✅ RocksDB incremental checkpoint enabled, externalized checkpoint retention correct.

✅ Paimon table primary key complete, aligned with business unique key.

✅ Changelog mode matches business scenario, downstream consumption semantics aligned.

✅ Snapshot retention policy reasonable, no unbounded growth risk.

✅ Job parallelism is integer multiple of bucket count.

✅ Fault injection tests and consistency reconciliation acceptance completed.

✅ Monitoring alerts configured, core metrics visualized.

Disclaimer: This article is based on Paimon 1.x and Flink 1.17/1.18; different versions may have API differences — refer to official documentation.

References:

Apache Paimon Official Docs: https://paimon.apache.org/docs/

Apache Flink Official Docs: https://flink.apache.org/docs/

Paimon Flink Connector Source: TwoPhaseCommitSinkFunction implementation