Apache NiFi Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Apache NiFi is a flow-based data integration platform that routes FlowFiles — packets of data with attributes and content — through chains of processors. Teams use it for billing pipelines when the data source is a feed of usage events (Kafka, S3, database CDC) that needs transformation before a Stripe charge: NiFi can pull records, enrich them, fan out to per-customer billing tasks, and write the results back to a data warehouse. But three NiFi behaviors — the failure-relationship retry loop that re-invokes InvokeHTTP after only the downstream database write failed, Primary Node failover that re-executes billing FlowFiles mid-flight on the new cluster leader, and a Funnel routing the same FlowFile to two parallel InvokeHTTP processors simultaneously — introduce Stripe billing failure modes that neither NiFi's built-in monitoring nor Stripe's Restricted Key feature surfaces on their own.

This post covers all three failure modes with NiFi flow configuration snippets and Python ExecuteScript billing code, and the two-layer governance pattern — content-hash idempotency keys plus per-FlowFile vault keys via a spend-cap proxy — that eliminates all three without restructuring your NiFi flow.

Failure mode 1: the failure-relationship retry loop re-invokes InvokeHTTP after the Stripe charge succeeded but the downstream database write failed

NiFi processors expose named relationships — typically success and failure — that downstream connections can route on. A billing flow that calls Stripe uses InvokeHTTP to POST to https://api.stripe.com/v1/charges. The InvokeHTTP processor routes the response FlowFile to its Response relationship on success (HTTP 2xx) or its Failure relationship on a network error or HTTP 5xx. The charge response FlowFile then passes to a PutDatabaseRecord processor that writes the billing record — customer ID, charge ID, amount, billing period — to a Postgres or MySQL table.

The failure mode appears when a NiFi developer adds a retry loop for resilience: they draw a connection from PutDatabaseRecord's Failure relationship back to InvokeHTTP's input queue. The intent is to retry the entire billing step if the database write fails transiently. This works for idempotent operations, but billing is not idempotent by default. When PutDatabaseRecord raises a connection timeout after the Stripe charge already succeeded, the FlowFile routes back to InvokeHTTP. InvokeHTTP sends a new POST to v1/charges — without an idempotency key, because the InvokeHTTP processor doesn't carry one automatically — and Stripe creates a second charge for the same customer. The PutDatabaseRecord failure was transient and resolves on the second attempt, so both charge records write to the database: ch_A (from the first successful call) and ch_B (from the retry). The customer is billed twice.

<!-- NiFi flow XML snippet — UNSAFE: retry loop from PutDatabaseRecord failure back to InvokeHTTP -->

<!-- InvokeHTTP processor: POST to Stripe charges -->
<processor>
  <id>invoke-stripe</id>
  <name>InvokeHTTP - POST to Stripe</name>
  <class>org.apache.nifi.processors.standard.InvokeHTTP</class>
  <property name="HTTP Method">POST</property>
  <property name="Remote URL">https://api.stripe.com/v1/charges</property>
  <property name="Basic Authentication Username">${stripe.secret.key}</property>
  <!-- UNSAFE: no Idempotency-Key header — every POST may create a new charge -->
</processor>

<!-- PutDatabaseRecord processor: write billing record to Postgres -->
<processor>
  <id>write-billing-db</id>
  <name>PutDatabaseRecord - Write Billing Record</name>
  <class>org.apache.nifi.processors.standard.PutDatabaseRecord</class>
</processor>

<!-- UNSAFE: failure relationship loops back to InvokeHTTP input -->
<connection>
  <sourceId>write-billing-db</sourceId>
  <selectedRelationship>failure</selectedRelationship>
  <destinationId>invoke-stripe</destinationId>  <!-- re-charges Stripe on DB failure -->
</connection>

There are two complementary fixes. First, compute a content-hash idempotency key from the billing inputs in a preceding ExecuteScript processor and set it as a FlowFile attribute — stripe.idempotency.key. Then configure InvokeHTTP to send that attribute as the Idempotency-Key request header using NiFi's expression language: ${stripe.idempotency.key}. The key must be derived from values stable across all retry iterations: sha256(customer_id:amount_cents:billing_period:nifi-billing)[:32]. Do not use the NiFi FlowFile UUID as the idempotency key — a new FlowFile is created on some retry paths, so the UUID changes between attempts. Second, break the retry loop: route PutDatabaseRecord's failure relationship to a separate retry processor (a PutDatabaseRecord instance or an UpdateAttribute that sets a retry counter attribute) rather than back to InvokeHTTP. The Stripe call should never be part of the retry path for a database write failure.

<!-- ExecuteScript: compute content-hash idempotency key before InvokeHTTP -->
<processor>
  <id>compute-idempotency-key</id>
  <name>ExecuteScript - Compute Billing Idempotency Key</name>
  <class>org.apache.nifi.processors.script.ExecuteScript</class>
  <property name="Script Engine">python</property>
  <property name="Script Body">
import java.io.InputStream as InputStream
import hashlib

flowFile = session.get()
if flowFile:
    customer_id = flowFile.getAttribute("customer.id")
    amount_cents = flowFile.getAttribute("amount.cents")
    billing_period = flowFile.getAttribute("billing.period")

    raw = f"{customer_id}:{amount_cents}:{billing_period}:nifi-billing"
    idempotency_key = hashlib.sha256(raw.encode()).hexdigest()[:32]

    flowFile = session.putAttribute(flowFile, "stripe.idempotency.key", idempotency_key)
    session.transfer(flowFile, REL_SUCCESS)
  </property>
</processor>

<!-- InvokeHTTP: SAFE — sends idempotency key header, uses vault key via proxy -->
<processor>
  <id>invoke-stripe-safe</id>
  <name>InvokeHTTP - POST to Stripe via Keybrake proxy</name>
  <class>org.apache.nifi.processors.standard.InvokeHTTP</class>
  <property name="HTTP Method">POST</property>
  <property name="Remote URL">https://proxy.keybrake.com/stripe/v1/charges</property>
  <property name="Basic Authentication Username">${keybrake.vault.key}</property>
  <property name="Attributes to Send as Headers">stripe.idempotency.key:Idempotency-Key</property>
</processor>

<!-- PutDatabaseRecord: SAFE — failure routes to dead-letter queue, NOT back to InvokeHTTP -->
<connection>
  <sourceId>write-billing-db-safe</sourceId>
  <selectedRelationship>failure</selectedRelationship>
  <destinationId>dead-letter-queue</destinationId>  <!-- alert + manual review -->
</connection>

When you use Keybrake's spend-cap proxy as the InvokeHTTP target URL, the vault key issued per billing run enforces a per-customer daily USD cap. Even if the idempotency key is missing from a retry — because a FlowFile attribute was lost on a routing path — the proxy's audit log surfaces the duplicate call immediately, and the daily cap stops the second charge if the customer already hit their expected billing amount for the period.

Failure mode 2: Primary Node failover re-executes billing FlowFiles that were in-flight during cluster node election

In a NiFi cluster, certain processors are restricted to run only on the Primary Node — the single cluster member elected to run exclusive workloads. GenerateFlowFile with a cron schedule, ListS3, ListKafkaRecords, and ExecuteStreamCommand processors that kick off billing runs are commonly run as Primary-Node-only processors to prevent all cluster nodes from triggering billing simultaneously. The NiFi cluster uses ZooKeeper for Primary Node election: when the current Primary Node loses its ZooKeeper session (network partition, GC pause longer than the session timeout, node crash), ZooKeeper elects a new Primary Node from the remaining members.

The failure mode is specific to the timing of a Primary Node failover during a billing run. Suppose the old Primary Node has executed InvokeHTTP and created Stripe charge ch_A for customer cus_abc. The response FlowFile is in the connection queue between InvokeHTTP and PutDatabaseRecord — it has not yet been claimed by PutDatabaseRecord for processing. At this moment, the old Primary Node loses its ZooKeeper session. NiFi elects a new Primary Node. The new Primary Node inherits the connection queues, which are stored in the NiFi content repository (not in memory). The new Primary Node starts PutDatabaseRecord, which dequeues the FlowFile and writes the billing record successfully. But GenerateFlowFile on the new Primary Node also fires — its cron schedule triggers again for the current billing window — producing a new billing FlowFile for cus_abc. This new FlowFile flows through InvokeHTTP, which creates charge ch_B.

This failure mode does not require a node crash. A long NiFi GC pause (common on nodes with large heap sizes — 32 GB to 128 GB — and G1GC tuning mismatches) causes a ZooKeeper session timeout just as reliably as a hardware failure. The charge duplication is invisible in NiFi's provenance view: both the first and second flows show clean success relationships throughout, with no errors recorded. The duplicate surfaces only in the Stripe dashboard or a billing reconciliation query.

<!-- The failure sequence in a NiFi cluster during Primary Node failover -->

<!--
  T=0:  Old Primary Node runs GenerateFlowFile → triggers billing for billing_period=2026-07
  T=5s: InvokeHTTP POSTs to Stripe → charge ch_A created for cus_abc (HTTP 200)
  T=6s: Response FlowFile in queue between InvokeHTTP and PutDatabaseRecord
  T=7s: Old Primary Node loses ZooKeeper session (GC pause / network blip)
  T=8s: New Primary Node elected — inherits all connection queues
  T=9s: New Primary Node's PutDatabaseRecord dequeues FlowFile → writes ch_A to DB (success)
  T=10s: New Primary Node's GenerateFlowFile cron fires → new billing FlowFile for 2026-07
  T=15s: InvokeHTTP POSTs to Stripe again → charge ch_B created for cus_abc (HTTP 200)
  T=16s: PutDatabaseRecord writes ch_B to DB → two charges in DB, customer billed twice
-->

<!-- GenerateFlowFile configured without a billing-period lock -->
<processor>
  <id>trigger-billing</id>
  <name>GenerateFlowFile - Monthly Billing Trigger</name>
  <class>org.apache.nifi.processors.standard.GenerateFlowFile</class>
  <schedulingStrategy>CRON_DRIVEN</schedulingStrategy>
  <schedulingPeriod>0 0 1 * * ?</schedulingPeriod>  <!-- first of each month -->
  <executionNode>PRIMARY</executionNode>
  <!-- UNSAFE: no dedup check — will re-trigger even if billing already ran this period -->
</processor>

The fix combines a distributed billing lock with a pre-flight deduplication check. Before InvokeHTTP fires for any customer, an ExecuteScript processor checks a shared dedup table in your database — a simple billing_executions table with (customer_id, billing_period, UNIQUE) — for a row matching the current customer and period. If the row exists, the FlowFile is routed to a skip relationship and auto-terminated; no Stripe call is made. If the row does not exist, the processor attempts an INSERT INTO billing_executions with an ON CONFLICT DO NOTHING clause; the insert is atomic at the database level, so concurrent NiFi nodes that race to process the same FlowFile after a failover will both attempt the insert, but only one will succeed and proceed to InvokeHTTP. The content-hash idempotency key is the final backstop for the narrow race where both nodes pass the insert check before either commits.

# ExecuteScript: distributed billing lock via Postgres upsert (Python / Jython)
import jaydebeapi
import hashlib

flowFile = session.get()
if not flowFile:
    return

customer_id = flowFile.getAttribute("customer.id")
billing_period = flowFile.getAttribute("billing.period")
amount_cents = int(flowFile.getAttribute("amount.cents"))

# Connect to Postgres billing lock table
conn = jaydebeapi.connect(
    "org.postgresql.Driver",
    "jdbc:postgresql://billing-db:5432/billing",
    ["billing_user", "${billing.db.password}"],
    "/opt/nifi/lib/postgresql.jar"
)
cursor = conn.cursor()

# Pre-flight: check if already billed (handles failover and post-24h-window retries)
cursor.execute(
    "SELECT charge_id FROM billing_executions WHERE customer_id=%s AND billing_period=%s",
    (customer_id, billing_period)
)
existing = cursor.fetchone()

if existing:
    log.warn(f"Skipping {customer_id} for {billing_period}: already billed ({existing[0]})")
    flowFile = session.putAttribute(flowFile, "billing.skip.reason", "already_billed")
    session.transfer(flowFile, REL_FAILURE)  # route to auto-terminate skip relationship
    conn.close()
    return

# Attempt distributed lock insert (atomic — only one NiFi node wins on concurrent failover)
try:
    cursor.execute(
        """INSERT INTO billing_executions (customer_id, billing_period, status)
           VALUES (%s, %s, 'in_progress')
           ON CONFLICT (customer_id, billing_period) DO NOTHING""",
        (customer_id, billing_period)
    )
    conn.commit()
    rows_inserted = cursor.rowcount
except Exception as e:
    log.error(f"Lock insert failed for {customer_id}: {e}")
    session.transfer(flowFile, REL_FAILURE)
    conn.close()
    return

if rows_inserted == 0:
    log.warn(f"Lock not acquired for {customer_id} — another node is billing this period")
    session.transfer(flowFile, REL_FAILURE)
    conn.close()
    return

# Compute content-hash idempotency key
raw = f"{customer_id}:{amount_cents}:{billing_period}:nifi-billing"
idempotency_key = hashlib.sha256(raw.encode()).hexdigest()[:32]
flowFile = session.putAttribute(flowFile, "stripe.idempotency.key", idempotency_key)

session.transfer(flowFile, REL_SUCCESS)
conn.close()

Failure mode 3: a Funnel routing the same customer FlowFile to two parallel InvokeHTTP processors dispatches both Stripe calls simultaneously

NiFi Funnels are zero-overhead merge points that combine FlowFiles from multiple upstream connections into a single output queue. They are also used in the reverse direction — as a visual organization device to split a flow into parallel processing branches — but this usage is a common source of confusion. A Funnel does not clone FlowFiles; a single FlowFile entering a Funnel exits once through its single output connection. The failure mode appears when a NiFi developer uses a different pattern to achieve fan-out: they draw two output connections from the same processor to two different downstream processors, intending to send the FlowFile to both.

NiFi's default behavior for multiple connections from one processor's relationship is to load-balance FlowFiles across the connections — not to clone and duplicate them. Each FlowFile is sent to exactly one of the two downstream connections, round-robin or by priority. This is correct behavior for load balancing, but when the developer expects both processors to receive a copy of every FlowFile, they add a second InvokeHTTP that is supposed to call a different API (Twilio, Resend) in parallel with the Stripe call. A misconfiguration in the routing — connecting the Stripe InvokeHTTP's Response relationship to a second InvokeHTTP processor instead of PutDatabaseRecord — or a copy-paste error that creates a second InvokeHTTP connected to the same output queue — results in the same billing FlowFile reaching both Stripe InvokeHTTP processors, which both POST to v1/charges for the same customer. Because NiFi's threading model runs processors concurrently across multiple threads, both calls can complete before either result is written to the database.

<!-- UNSAFE: two InvokeHTTP processors both receiving billing FlowFiles from the same queue -->

<!-- A common copy-paste error: developer meant to add a Twilio InvokeHTTP in parallel,
     but accidentally connected a second Stripe InvokeHTTP to the billing input queue -->

<connection>
  <sourceId>build-billing-payload</sourceId>
  <selectedRelationship>success</selectedRelationship>
  <destinationId>invoke-stripe-1</destinationId>
</connection>

<connection>
  <sourceId>build-billing-payload</sourceId>
  <selectedRelationship>success</selectedRelationship>
  <destinationId>invoke-stripe-2</destinationId>  <!-- UNSAFE: second connection to same queue -->
</connection>

<!-- NiFi load-balances FlowFiles: each FlowFile goes to exactly ONE downstream processor.
     But if both processors are Stripe InvokeHTTP instances (not one Stripe + one Twilio),
     half the billing FlowFiles are charged by invoke-stripe-1 and half by invoke-stripe-2.

     The worse scenario: if the developer uses ReplicateFlowFile or if the same FlowFile
     attribute routes to both via a RouteOnAttribute misconfiguration, BOTH processors
     receive the same FlowFile and both POST to Stripe independently. -->

The fix combines correct NiFi flow topology with a Keybrake audit log cross-check. For genuine parallel API fan-out (Stripe + Twilio + Resend all from the same billing event), use the proper NiFi cloning pattern: route the FlowFile through an ExecuteScript processor that explicitly clones the FlowFile into N copies with session.create(flowFile), each tagged with a target.api attribute, then route by attribute to the correct InvokeHTTP. Never connect two InvokeHTTP processors to the same output relationship of an upstream processor unless you intend load-balancing with identical behavior. For the Stripe path specifically, Keybrake's per-FlowFile vault keys — issued before the billing loop and scoped to a daily USD cap — mean that even if two InvokeHTTP processors both receive the same FlowFile due to a flow configuration error, the second charge attempt against the same vault key either hits the spend cap or is flagged in the audit log before your next reconciliation run.

# ExecuteScript: proper FlowFile cloning for parallel API calls (Python / Jython)
# Use this pattern instead of multiple connections from the same relationship

flowFile = session.get()
if not flowFile:
    return

# Create explicit copies — one per target API
stripe_ff = session.create(flowFile)
twilio_ff = session.create(flowFile)
resend_ff  = session.create(flowFile)

stripe_ff = session.putAttribute(stripe_ff, "target.api", "stripe")
twilio_ff = session.putAttribute(twilio_ff, "target.api", "twilio")
resend_ff  = session.putAttribute(resend_ff,  "target.api", "resend")

# Remove the original FlowFile
session.remove(flowFile)

# Transfer all three to a single RouteOnAttribute processor
# which routes by ${target.api} to the correct InvokeHTTP
session.transfer(stripe_ff, REL_SUCCESS)
session.transfer(twilio_ff, REL_SUCCESS)
session.transfer(resend_ff,  REL_SUCCESS)

The two-layer governance pattern for all three failure modes

Each failure mode above has a specific structural fix in the NiFi flow. But the same two-layer pattern closes all three simultaneously and provides defense in depth for failure modes you haven't encountered yet.

Layer 1: content-hash idempotency keys. Before any InvokeHTTP processor that calls Stripe, run an ExecuteScript processor that computes sha256(customer_id:amount_cents:billing_period:nifi-billing)[:32] and sets it as a FlowFile attribute. Configure InvokeHTTP to send this attribute as the Idempotency-Key HTTP request header. The key must be derived from the billing inputs, not from the FlowFile UUID or a timestamp — NiFi can create new FlowFiles on retry paths, and the idempotency key must be stable across every path that might re-invoke the Stripe call for the same billing event.

Layer 2: per-FlowFile vault keys via a spend-cap proxy. Instead of reading your Stripe secret key from a NiFi Sensitive Property (which is shared across all FlowFiles and all processor threads), issue a vault key from Keybrake before the billing loop begins — scoped to the expected billing amount per customer per period — and set it as a flow variable or a FlowFile attribute. Configure InvokeHTTP's Remote URL as https://proxy.keybrake.com/stripe/v1/charges and Basic Authentication Username as ${keybrake.vault.key}. The proxy enforces the per-customer daily USD cap, forwards the real Stripe key server-side, and logs every call with its idempotency key, FlowFile attributes, and HTTP status — giving you a queryable audit trail that your NiFi provenance view doesn't provide for cross-node and cross-run correlation.

NiFi-specific audit gaps that make Stripe billing failures hard to detect

NiFi's provenance view tracks every FlowFile's lineage — which processors it passed through, what attributes it had at each step, how long each processing step took. This is excellent for debugging data routing issues. But it has three gaps that make Stripe billing failures hard to detect without an external audit log.

First, NiFi provenance does not correlate across cluster nodes. After a Primary Node failover, the provenance records on the old node (now offline) and the new node are separate repositories. A FlowFile that was processed on the old node (charge ch_A created) and then re-processed on the new node (charge ch_B created) appears as two independent successful FlowFile flows in two separate provenance views, with no cross-reference between them. You need a billing-period dedup table in your external database to catch this.

Second, NiFi provenance retention is typically short — 24 to 72 hours — because the repository is local disk and large flows generate enormous provenance volumes. A double-charge discovered from a customer support ticket 4 days after the billing run will not have provenance data to explain which flow path caused it. An external audit table with the charge ID, idempotency key, and FlowFile attributes written at the time of the Stripe call is the only reliable post-incident record.

Third, NiFi's kill-switch story for live billing is coarse-grained. You can stop a processor, which halts new FlowFiles from being claimed, but FlowFiles already claimed by a processor thread and in mid-execution are not interrupted — they complete. In a billing pipeline with 10,000 FlowFiles in the InvokeHTTP queue, stopping the processor stops new Stripe calls from starting but does not prevent the 40 FlowFiles already executing across NiFi's thread pool from completing their charges. A Keybrake vault key revocation takes effect in under a second across all in-flight requests — the proxy returns a 401 immediately, InvokeHTTP routes those FlowFiles to its Failure relationship, and no further Stripe calls succeed for that vault key regardless of what is in the NiFi queue.

FAQ

Does the NiFi Sensitive Properties Key protect Stripe API keys from other NiFi users?

Yes — NiFi Sensitive Properties are encrypted at rest using the Sensitive Properties Key configured in nifi.properties. But encryption at rest does not prevent authorized NiFi users with READ permission on the processor from seeing the configured key value in the NiFi UI (it is displayed as ***** in the UI but can be extracted via the NiFi API). More importantly, one Stripe secret key shared across all billing FlowFiles means a misconfigured retry loop or a bad flow topology can fire that key against Stripe without any per-FlowFile spend cap — the key has no ceiling on what it can charge. Vault keys issued per billing run from a spend-cap proxy scope each run's Stripe access to its expected billing amount, regardless of how many times the key is used in that run.

Can I use NiFi's built-in retry processor instead of a custom retry loop?

Yes — NiFi 1.16+ has a RetryFlowFile processor that manages retry count as a FlowFile attribute and routes to retries_exceeded when the limit is hit. This is better than a hand-drawn loop because it prevents infinite retry cycles. But it does not prevent the retry from re-invoking InvokeHTTP against Stripe unless you restructure the flow to only retry the database write step, not the entire Stripe-call-plus-database-write chain. The idempotency key on InvokeHTTP is the correct fix for the Stripe call itself; RetryFlowFile is the correct fix for the database write retry.

Does NiFi's back-pressure mechanism prevent billing FlowFiles from piling up?

Back-pressure on a connection (configured via the connection's object count or data size threshold) halts the upstream processor from adding more FlowFiles to that connection until the queue drains below the threshold. This prevents queue overflow but does not prevent double-charging. If the InvokeHTTP processor is running and the downstream database connection is back-pressured, InvokeHTTP continues claiming FlowFiles from its input queue and calling Stripe — the Stripe calls succeed, and the response FlowFiles accumulate in the InvokeHTTP-to-PutDatabaseRecord connection until back-pressure releases. No double-charging occurs from back-pressure alone. Double-charging occurs from the retry-loop pattern, not from back-pressure.

How do I issue a Keybrake vault key from within a NiFi ExecuteScript processor?

Use an ExecuteScript processor at the start of your billing flow — before the per-customer fan-out — to call the Keybrake API once and set the vault key as a flow variable or a FlowFile attribute. The API call is a POST to https://api.keybrake.com/v1/vault-keys with your Keybrake account key and the policy parameters (vendor, daily_usd_cap, billing_period label). The response contains a vault_key that is then set as a flow variable via context.getVariableRegistry().setVariable("keybrake.vault.key", vault_key) — all downstream FlowFiles in the same flow inherit it. For per-customer vault keys with tighter spend caps, call the API once per customer FlowFile in the fan-out layer instead.

Does using a reverse proxy (Keybrake) add latency to each Stripe API call through NiFi?

Keybrake's proxy adds approximately 5-15ms of overhead per Stripe call — a policy lookup (SQLite, in-process), a spend cap check (one read + one write to the audit table), and a standard HTTPS forward to Stripe's API endpoint. For NiFi billing pipelines that call Stripe once per customer per month, 10ms of additional latency per call is negligible compared to Stripe's own API response time (typically 150-400ms) and the NiFi thread scheduling and InvokeHTTP processor overhead (typically 20-50ms). The audit log and spend cap enforcement that the proxy provides are worth the latency premium, which is invisible against the background of a billing pipeline's total runtime.