Apache Storm and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance
Apache Storm is a distributed real-time stream processing system built around at-least-once delivery: every tuple is guaranteed to be processed, but not exactly once. When a billing topology reads usage events and calls stripe.charges.create() in a bolt, three Storm-specific behaviors — topology.message.timeout.secs tuple replay, JVM worker kill with bolt-local state loss, and Trident’s misunderstood exactly-once scope — introduce Stripe double-charge failure modes that standard retry backoff and Kafka offset tracking alone cannot prevent.
This post covers those three failure modes with Python (streamparse) and Java (Trident) code examples, content-hash idempotency keys that exclude all Storm metadata, pre-flight PostgreSQL checks with ON CONFLICT DO NOTHING, per-billing-period vault keys via a spend-cap proxy, and write-before-ack ordering — the two-layer governance pattern that closes all three failure modes without changing Storm topology configuration or Kafka retention settings.
Failure mode 1: topology.message.timeout.secs expiry during stripe.charges.create() — root spout replays the tuple with a new Storm-assigned ID; billing bolt processes the replayed tuple and calls stripe.charges.create() again; idempotency key derived from Storm tuple metadata creates ch_B
Storm’s reliability mechanism is built around the acker task. When a spout emits a tuple, it registers a messageId and Storm assigns a random 64-bit tuple ID to the emitted tuple. Downstream bolts must call collector.ack(input_tuple) for every tuple they successfully process. The acker tracks the XOR of all tuple IDs in each processing tree; when the XOR reaches zero (all downstream bolts have acked their tuples), the root is considered complete and the spout’s ack(messageId) is called. If the XOR does not reach zero within topology.message.timeout.secs (default: 30 seconds), the acker marks the root as failed and calls the spout’s fail(messageId) method.
A Kafka spout (using the storm-kafka-client connector or the older storm-kafka library) stores the last successfully committed Kafka partition offset in ZooKeeper. On fail(messageId), the spout re-emits the record at the failed Kafka offset. Storm assigns a new tuple ID to this re-emitted tuple — tuple IDs are random longs generated independently per emission event.
The failure scenario: stripe-python’s default max_network_retries=2 with exponential backoff can produce a worst-case round-trip time of approximately 90 seconds across three network attempts (3 × 30 s). With the default topology.message.timeout.secs=30, any stripe.charges.create() call that takes longer than 30 seconds causes the tuple to time out. The spout re-emits the original Kafka record. The billing bolt receives the replayed tuple and calls stripe.charges.create() a second time — while the original call may still be in-flight on the first bolt instance.
# streamparse / Storm multilang protocol (Python bolt)
import stripe
from streamparse.bolt import Bolt
class BillingBolt(Bolt):
def process(self, tup):
billing = tup.values[0]
customer_id = billing['customer_id']
billing_period = billing['billing_period']
# UNSAFE: idempotency key includes Storm tuple ID.
# Storm assigns a NEW tuple ID to every replayed tuple.
# topology.message.timeout.secs=30 fires before stripe.charges.create()
# completes if Stripe latency exceeds the timeout window.
# Replayed tuple has a different tup.id → different idempotency key
# → Stripe creates ch_B alongside ch_A for the same customer+period.
idempotency_key = f"storm-{tup.id}:{customer_id}:{billing_period}"
charge = stripe.Charge.create(
amount=billing['amount_cents'],
currency='usd',
customer=billing['stripe_customer_id'],
description=f'Usage billing for {billing_period}',
idempotency_key=idempotency_key,
)
self.ack(tup)
The tuple ID in the example is a Storm-internal 64-bit random long. It is not the Kafka offset, not the Kafka message key, and not any business-level field. The replayed tuple has a completely different tuple ID. The two idempotency keys are distinct. Stripe processes both requests and creates ch_A and ch_B for the same customer and billing period.
This failure mode is correlated with Stripe network latency and topology timeout configuration. It is most likely during a high-load billing run when Stripe API latency is elevated and the 30-second default timeout is tight. A developer who increases topology.message.timeout.secs to 120 or 300 seconds reduces (but does not eliminate) the replay window — Stripe P99 latency with retries can reach 90 seconds, and network timeouts can be longer. The fix must close the window regardless of timeout tuning.
A second variant of this failure mode occurs when the bolt processes a batch of tuples. A Storm bolt can call emit_many (streamparse) or process multiple input tuples before acking. If a timeout fires during mid-batch processing, the entire unacked batch is replayed. Records already charged in the first pass are charged again on the replay pass, with each replayed tuple carrying a new Storm-assigned ID.
import hashlib
import psycopg2
import stripe
from streamparse.bolt import Bolt
def make_idempotency_key(customer_id: str, billing_period: str) -> str:
# Content-hash from stable business fields only.
# Must NOT include: tup.id (new random long on every replay),
# tup.task_id (Storm task assignment — can change on reassignment),
# any Storm context timestamp (set at tuple emission time, changes per replay).
raw = f"{customer_id}:{billing_period}:storm-billing"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
class BillingBolt(Bolt):
def initialize(self, stormconf, context):
self.db_conn = psycopg2.connect(host='db', dbname='billing', user='app')
def process(self, tup):
billing = tup.values[0]
customer_id = billing['customer_id']
billing_period = billing['billing_period']
idempotency_key = make_idempotency_key(customer_id, billing_period)
# Pre-flight check against PostgreSQL — external, durable, shared
# across all Storm workers. Closes the replay race.
with self.db_conn.cursor() as cur:
cur.execute("""
INSERT INTO billing_records (customer_id, billing_period, idempotency_key)
VALUES (%s, %s, %s)
ON CONFLICT (customer_id, billing_period) DO NOTHING
RETURNING id
""", (customer_id, billing_period, idempotency_key))
inserted = cur.rowcount
self.db_conn.commit()
if inserted == 0:
# Already billed — replayed tuple, skip Stripe call.
self.ack(tup)
return
charge = stripe.Charge.create(
amount=billing['amount_cents'],
currency='usd',
customer=billing['stripe_customer_id'],
description=f'Usage billing for {billing_period}',
idempotency_key=idempotency_key,
)
self.ack(tup)
The safe pattern: content-hash idempotency key derived from business fields only, combined with a pre-flight PostgreSQL check using ON CONFLICT DO NOTHING. If the tuple is replayed, the pre-flight check finds the existing row and the bolt acks without calling Stripe again. The content-hash key provides a second layer via Stripe’s 24-hour idempotency window if the pre-flight row was written but PostgreSQL is briefly unavailable on the replay pass.
Failure mode 2: JVM GC pause triggers supervisor worker kill; billing bolt’s ack() call is in the JVM stack when the worker process is killed; bolt-local dedup state is destroyed; replacement worker has no record of prior charges
Storm workers are JVM processes. A long stop-the-world GC pause — common with CMS concurrent mode failure on large heap allocations or G1GC humongous object promotions during a high-throughput billing run — prevents the worker’s heartbeat thread from sending heartbeats to the supervisor daemon. The supervisor’s heartbeat timeout (supervisor.worker.timeout.secs, default: 3 minutes in most Storm distributions) causes the supervisor to kill the worker process via SIGKILL and restart it.
All tuples in-flight on the killed worker are orphaned from the acker’s perspective. The acker task (which runs in a separate JVM, on a different worker) marks those root tuples as failed after topology.message.timeout.secs passes without receiving the ack. The root spout calls fail(messageId) for each and re-emits the original records.
The specific timing that creates the duplicate charge: the billing bolt calls stripe.charges.create() and receives ch_A. The call returns. The bolt is about to write to the billing database and call self.ack(tup). At this moment, the stop-the-world GC pause freezes all JVM threads. The supervisor times out and kills the process. self.ack(tup) was never called. The acker marks the tuple as failed. The spout re-emits it. A new billing bolt instance on the replacement worker processes the replayed tuple.
If the billing bolt used a bolt-local set or dict as its dedup guard, that guard is gone:
class BillingBolt(Bolt):
def initialize(self, stormconf, context):
# UNSAFE: bolt-local dedup store. Lives in JVM heap.
# Destroyed when the worker process is killed by supervisor.
# Replacement worker starts with an empty set.
self._billed = set()
def process(self, tup):
billing = tup.values[0]
customer_id = billing['customer_id']
billing_period = billing['billing_period']
key = (customer_id, billing_period)
if key in self._billed:
# Skip — already billed this period.
# DOES NOT PROTECT across worker restarts.
self.ack(tup)
return
idempotency_key = make_idempotency_key(customer_id, billing_period)
charge = stripe.Charge.create(
amount=billing['amount_cents'],
currency='usd',
customer=billing['stripe_customer_id'],
idempotency_key=idempotency_key,
)
self._billed.add(key) # lost on JVM kill
self.ack(tup)
The replacement worker starts fresh. self._billed is empty. The pre-flight check on line 11 passes. stripe.charges.create() is called again for the customer that was already charged by the killed worker.
Within Stripe’s 24-hour idempotency window, the content-hash idempotency key returns ch_A from Stripe’s idempotency cache — no visible duplicate on the customer’s statement. This creates a false sense of safety during normal operations and routine chaos tests that complete within hours. The failure emerges in practice when:
- A Storm topology is taken offline for maintenance (Nimbus upgrade, ZooKeeper rolling restart) for more than 24 hours, then brought back up. Replayed tuples from the pre-maintenance run now hit Stripe outside the idempotency window.
- A billing run is paused by a Nimbus failover that takes more than 24 hours to recover. The backlog of replayed tuples hits Stripe fresh.
- A Kafka spout commits offsets infrequently, and a large replay batch from a prior billing run takes more than 24 hours to drain through the topology after the worker restart.
The pre-flight PostgreSQL check closes this failure mode regardless of how much time has elapsed since the original charge:
class BillingBolt(Bolt):
def initialize(self, stormconf, context):
# Durable external store — survives worker restarts, Nimbus failovers,
# topology kills and restarts. Shared across all bolt instances
# on all workers in the Storm cluster.
self.db_conn = psycopg2.connect(host='db', dbname='billing', user='app')
def process(self, tup):
billing = tup.values[0]
customer_id = billing['customer_id']
billing_period = billing['billing_period']
idempotency_key = make_idempotency_key(customer_id, billing_period)
with self.db_conn.cursor() as cur:
cur.execute("""
INSERT INTO billing_records
(customer_id, billing_period, idempotency_key, created_at)
VALUES (%s, %s, %s, NOW())
ON CONFLICT (customer_id, billing_period) DO NOTHING
RETURNING id
""", (customer_id, billing_period, idempotency_key))
inserted = cur.rowcount
self.db_conn.commit()
if inserted == 0:
# billing_records row already exists — this is a replayed tuple.
# The original worker charged the customer. Skip Stripe call.
self.ack(tup)
return
try:
charge = stripe.Charge.create(
amount=billing['amount_cents'],
currency='usd',
customer=billing['stripe_customer_id'],
description=f'Usage billing for {billing_period}',
idempotency_key=idempotency_key,
)
self.ack(tup)
except stripe.error.CardError:
# Write card decline to billing_records and ack — prevents
# infinite Storm replay of an unchargeable billing event.
with self.db_conn.cursor() as cur:
cur.execute("""
UPDATE billing_records SET status = 'card_declined'
WHERE customer_id = %s AND billing_period = %s
""", (customer_id, billing_period))
self.db_conn.commit()
self.ack(tup)
except stripe.error.Timeout:
# Do NOT ack on Timeout — Stripe may have created the charge
# server-side before the timeout response was sent.
# Let the tuple timeout and replay. On replay, the pre-flight
# check finds the billing_records row (written before the
# Stripe call) and skips without a second Stripe call.
raise
Note the write-before-Stripe ordering: billing_records is inserted before stripe.charges.create() is called. If the worker is killed between the INSERT and the Stripe call, the replayed tuple’s pre-flight check finds the row and skips the Stripe call. If Stripe returned Timeout after creating the charge server-side, the content-hash idempotency key returns ch_A on the replay (within 24 hours) and the pre-flight row is already in place to skip the replay at the database layer.
Failure mode 3: Storm Trident’s exactly-once guarantee is scoped to TridentState updates, not to BaseFunction.execute() side effects; stripe.charges.create() inside a Trident function runs on every batch replay; idempotency key derived from Trident partition metadata creates ch_B when partition assignment changes after worker failure
Storm Trident provides exactly-once processing semantics for stream operations, but this guarantee has a precise and narrow scope: each source tuple contributes to each TridentState update exactly once. Trident achieves this by dividing the stream into microbatches, assigning each batch a monotonically increasing transaction ID (txid), and requiring TridentState backends (Redis, Cassandra, PostgreSQL via OpaqueTridentJdbcState, etc.) to implement transactional or opaque-transactional commit protocols that detect and skip previously-committed batches by txid.
stripe.charges.create() cannot be wrapped in a TridentState backend — there is no Trident state abstraction for external API calls. It must be called inside a BaseFunction (Java) or a function applied via tup.apply() in a Trident stream. BaseFunction.execute() runs on every attempt of the batch, including replays.
The Trident batch replay scenario: a billing topology processes a stream of usage events in microbatches of 100 records each. Batch txid=42 contains usage events for 100 customers. The BillingFunction processes records 1–80, calling stripe.charges.create() for each. At record 81, the topology’s TridentState commit for an upstream aggregation fails — a Cassandra timeout, a Redis WRONGTYPE error on the opaque value comparison, or a Nimbus batch coordinator failure. Trident marks the batch as failed and schedules a replay with the same txid=42. Records 1–80 are processed again.
In Java, the unsafe pattern (Trident’s partition metadata in the idempotency key):
// Java — Storm Trident BaseFunction
public class BillingFunction extends BaseFunction {
@Override
public void execute(TridentTuple tuple, TridentCollector collector) {
String customerId = tuple.getStringByField("customer_id");
String billingPeriod = tuple.getStringByField("billing_period");
// Trident context: txid is stable across replays of the same batch.
// partition is the Trident partition index — assigned by Storm's
// task scheduling. After a worker failure and task reassignment,
// partition.index may differ for the same source records on replay.
long txid = this.txInfo.txid;
int partitionIndex = this.txInfo.partitionIndex;
// UNSAFE: includes partitionIndex.
// On replay after worker reassignment, partitionIndex changes.
// Same customer, same billing period → different idempotency key → ch_B.
String idempotencyKey = "trident-" + txid + "-" + partitionIndex
+ ":" + customerId + ":" + billingPeriod;
Map<String, Object> params = new HashMap<>();
params.put("amount", tuple.getIntegerByField("amount_cents"));
params.put("currency", "usd");
params.put("customer", tuple.getStringByField("stripe_customer_id"));
params.put("idempotency_key", idempotencyKey);
Charge charge = Charge.create(params);
collector.emit(new Values(charge.getId()));
}
}
Trident assigns the same txid to all retry attempts of the same batch — this is the documented stable property. But when a worker failure during the batch causes Storm to reassign tasks, the partitionIndex (the Trident partition’s position in the task assignment ordering) can change. A billing batch originally processed with partitionIndex=2 may be replayed on a new worker as partitionIndex=0. The idempotency key suffix changes. Stripe sees a new unique key and creates ch_B for every customer in that partition.
A subtler variant: the developer uses txid alone as the idempotency key prefix, correctly making it stable across replays. But since all 100 records in the batch share the same txid, the developer appends a record-level identifier to ensure per-record uniqueness. If that identifier is tuple.getInteger("offset_in_partition") (the record’s position within the Trident partition input) and the Trident spout is re-partitioned during replay (partition count changed, task slots reassigned), the offset-in-partition for the same source record changes — again, a different idempotency key, a different Stripe charge.
The safe pattern for Trident (Java) is to use a content-hash key from business fields only and add the pre-flight database check inside execute():
// Java — Storm Trident BaseFunction (safe version)
import java.security.MessageDigest;
import java.nio.charset.StandardCharsets;
public class BillingFunction extends BaseFunction {
private Connection dbConn;
@Override
public void prepare(Map conf, TridentOperationContext context) {
this.dbConn = DriverManager.getConnection(
(String) conf.get("billing.db.url"),
(String) conf.get("billing.db.user"),
(String) conf.get("billing.db.password")
);
}
private String makeIdempotencyKey(String customerId, String billingPeriod) {
// Content-hash from stable business fields only.
// Must NOT include: txid (batch ID — correct for batch dedup but
// wrong for per-record dedup across txids), partitionIndex (changes
// on reassignment), offset-in-partition (changes on repartitioning),
// Storm task ID, or any Trident topology metadata.
String raw = customerId + ":" + billingPeriod + ":trident-billing";
try {
MessageDigest md = MessageDigest.getInstance("SHA-256");
byte[] hash = md.digest(raw.getBytes(StandardCharsets.UTF_8));
StringBuilder hex = new StringBuilder();
for (int i = 0; i < 16; i++) {
hex.append(String.format("%02x", hash[i]));
}
return hex.toString();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@Override
public void execute(TridentTuple tuple, TridentCollector collector) {
String customerId = tuple.getStringByField("customer_id");
String billingPeriod = tuple.getStringByField("billing_period");
String idempotencyKey = makeIdempotencyKey(customerId, billingPeriod);
try {
// Pre-flight check — external, durable, shared across all
// Trident workers. Closes Trident batch replay failure mode
// regardless of partition assignment changes.
PreparedStatement ps = dbConn.prepareStatement(
"INSERT INTO billing_records (customer_id, billing_period, idempotency_key) " +
"VALUES (?, ?, ?) ON CONFLICT (customer_id, billing_period) DO NOTHING"
);
ps.setString(1, customerId);
ps.setString(2, billingPeriod);
ps.setString(3, idempotencyKey);
int inserted = ps.executeUpdate();
dbConn.commit();
if (inserted == 0) {
// Already billed. Trident batch replay — skip Stripe call.
collector.emit(new Values("already_billed"));
return;
}
RequestOptions options = RequestOptions.builder()
.setIdempotencyKey(idempotencyKey)
.build();
Map<String, Object> params = new HashMap<>();
params.put("amount", tuple.getIntegerByField("amount_cents"));
params.put("currency", "usd");
params.put("customer", tuple.getStringByField("stripe_customer_id"));
Charge charge = Charge.create(params, options);
collector.emit(new Values(charge.getId()));
} catch (Exception e) {
dbConn.rollback();
throw new RuntimeException(e);
}
}
}
The Trident exactly-once guarantee does not prevent this function from running twice on the same source record. It only ensures that if you write to a TridentState backend (using the opaque or transactional commit protocol), the state change is applied exactly once per source tuple. A billing architecture that relies on Trident’s exactly-once guarantee to prevent Stripe double charges is based on a misunderstanding of where the guarantee applies. The pre-flight check is the only mechanism that prevents re-execution of the Stripe API call.
The governance pattern that closes all three failure modes
1. Content-hash idempotency key from stable business fields only
The idempotency key must be derived exclusively from fields that are stable across all Storm replay scenarios — tuple replays, worker restarts, topology kills, Trident batch replays, and partition reassignments:
import hashlib
def make_idempotency_key(customer_id: str, billing_period: str) -> str:
# Stable across all Storm replay scenarios.
# Does NOT include: tup.id (new random long per emission),
# tup.task_id (changes on task reassignment), Storm context timestamp
# (set at emission time, changes on replay), Kafka partition (changes
# if producer routes to different partition on retry), Kafka offset
# (different for each message copy on KafkaSpout replay from fail()),
# Trident txid (batch-level — wrong scope for per-record Stripe keys
# because all records in one batch share the same txid),
# Trident partitionIndex (changes on partition reassignment).
raw = f"{customer_id}:{billing_period}:storm-billing"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
2. Pre-flight PostgreSQL check with ON CONFLICT DO NOTHING
Write to billing_records before calling stripe.charges.create(). This closes all three failure modes regardless of Storm topology configuration:
- Tuple timeout replay (failure mode 1): The pre-flight
INSERTwas written by the original bolt instance before calling Stripe. The replayed tuple’s bolt finds the row and skips without a second Stripe call. - Worker kill (failure mode 2): If the worker is killed before the
INSERT, no Stripe call was made — nothing to dedup. If the worker is killed after theINSERTbut beforeack(), the replayed tuple’s bolt finds the row and skips. If the worker is killed after the Stripe call but beforeack(), the pre-flight row is already in place and the replay skips cleanly. - Trident batch replay (failure mode 3): The pre-flight check is inside
BaseFunction.execute(), which runs on every batch attempt. Records billed in a prior attempt have rows inbilling_records. TheON CONFLICT DO NOTHINGreturns zero rows inserted, and the function emits without calling Stripe.
-- DDL: billing_records table with unique constraint on (customer_id, billing_period).
-- ON CONFLICT DO NOTHING ensures that concurrent bolt instances processing
-- the same replayed tuple (e.g., two workers racing on reassignment) are
-- safe: only one INSERT succeeds; the other silently no-ops.
CREATE TABLE billing_records (
id BIGSERIAL PRIMARY KEY,
customer_id TEXT NOT NULL,
billing_period TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'billed',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (customer_id, billing_period)
);
3. Per-billing-period vault key capped at expected_total × 1.10
A Keybrake vault key per customer per billing period, capped at the expected billing amount times 1.10, provides a hard backstop for three cases that the pre-flight check cannot close:
- Stripe’s 24-hour idempotency window closes before the replay is processed (topology offline for maintenance, Nimbus failover longer than 24 hours). The content-hash key no longer deduplicates. The pre-flight check closes this, but the vault cap is the final backstop if PostgreSQL was also restored from backup without the
billing_recordsrow. - Concurrent bolt instances both pass the pre-flight check in the same millisecond (race between the
SELECTand theINSERT). TheON CONFLICT DO NOTHINGunique constraint handles this at the database layer, but the vault cap is the backstop if both bolt instances pass and the constraint is not configured. - A Trident batch replay that completed after Stripe’s idempotency window closed and the
billing_recordstable was cleared by a data retention policy.
Protection level comparison
| Protection level | Timeout replay (FM1) | Worker kill / local state loss (FM2) | Trident batch replay (FM3) |
|---|---|---|---|
| No protection | ch_B on every replay past 30 s | ch_B on every replay past worker restart | ch_B on every batch replay past partition reassignment |
| Tuple-ID-based idempotency key only | ch_B (new tuple ID on replay) | ch_B (new tuple ID on replay) | ch_B (partitionIndex changes on reassignment) |
| Content-hash key only (no pre-flight) | Safe within 24 h; ch_B after idempotency window | Safe within 24 h; ch_B after 24 h gap | Safe within 24 h; ch_B after idempotency window |
| Bolt-local dedup (in-memory set) only | Safe within JVM lifetime; ch_B after worker restart | ch_B on every worker restart | ch_B on every new Trident worker instance |
| Full pattern: content-hash + pre-flight + vault cap | Safe always | Safe always | Safe always |
Gap analysis: four additional Storm failure modes
1. topology.message.timeout.secs tuned too low for high-latency billing bolts
A default timeout of 30 seconds is appropriate for low-latency stream processing (parsing, filtering, aggregating) but is too short for bolts that make synchronous external HTTP calls. A billing bolt that calls stripe.charges.create() with max_network_retries=2 needs at least 120–300 seconds of timeout budget. Setting topology.message.timeout.secs to 300 reduces but does not eliminate the replay window — the content-hash key and pre-flight check remain necessary.
2. Storm tick tuples fire billing bolts on every worker
Storm tick tuples (Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS) deliver a periodic tuple to every bolt instance running on every worker in the topology. A billing bolt that triggers a Stripe billing cycle on receiving a tick tuple fires on every task instance simultaneously, not once per topology. A 4-worker Storm cluster with 8 billing bolt tasks fires 8 simultaneous billing cycles per tick interval. A Redis SET NX EX distributed lock or a PostgreSQL advisory lock is required to ensure only one task executes the billing cycle per tick.
3. Spout offset commit failure leaves gap in replay coverage
The Kafka spout commits the last successfully acked offset to ZooKeeper. If ZooKeeper is briefly unavailable during offset commit (a transient ZooKeeper leader election), the spout may continue emitting past the uncommitted offset. After a topology restart, the spout replays from the last committed offset — which predates the ZooKeeper outage. Records emitted during the outage window are processed again. The billing bolt’s pre-flight check against PostgreSQL handles these replays identically to ack-timeout replays: ON CONFLICT DO NOTHING returns zero rows for already-billed customers, and the bolt acks without calling Stripe.
4. Storm DRPC topology fires Stripe calls via remote procedure call clients
Storm DRPC (Distributed Remote Procedure Call) topologies expose synchronous RPC endpoints backed by real-time stream processing. A billing DRPC function that calls stripe.charges.create() inside a LinearDRPCTopologyBuilder bolt is subject to the same ack-timeout replay and worker-kill failure modes as a regular topology. Additionally, DRPC clients can retry the remote call if the initial request times out, creating a duplicate RPC invocation that triggers a second bolt execution and a second Stripe call for the same billing request. The content-hash idempotency key and pre-flight PostgreSQL check close this failure mode at the bolt layer.
FAQ
How often does tuple replay actually happen in a production Storm billing topology?
Tuple replay frequency depends on Stripe API latency relative to topology.message.timeout.secs and on JVM GC characteristics. In a high-throughput billing run with 30-second timeout and Stripe P99 latency of 8–15 seconds (normal range), replay is rare. Under Stripe load spikes that push P99 past 30 seconds, replay becomes frequent and is correlated with billing load — exactly when you can least afford duplicate charges. A content-hash key and pre-flight check cost a single PostgreSQL round-trip per tuple: worth paying unconditionally.
Can I use Storm Trident’s txid as the Stripe idempotency key to get exactly-once Stripe calls?
No. The txid is a batch-level identifier that covers all records in the batch, not individual records. Using txid alone as the idempotency key would cause Stripe to process only the first record and return the same charge object for all subsequent records in the batch — silently skipping billing for all customers except the first. A per-record key derived from business fields (customer_id + billing_period) is required. The txid is not part of the key because it would require that every billing period spans exactly one Trident batch, which is not true in practice.
Does increasing topology.message.timeout.secs to match Stripe’s worst-case latency eliminate tuple replay?
It reduces the frequency of replay events but does not eliminate them. JVM GC pauses can stop heartbeat threads for longer than the configured timeout regardless of timeout value. Supervisor kills fire after supervisor.worker.timeout.secs, which defaults to 3 minutes in most Storm distributions. A billing bolt that is killed mid-Stripe-call (after stripe.charges.create() returns but before ack()) produces a replayed tuple regardless of timeout configuration. The pre-flight PostgreSQL check is the authoritative dedup gate.
Can I use Storm’s at-least-once guarantee as a feature by idempotent-side-effecting the Stripe call?
Stripe’s idempotency key mechanism provides this: within 24 hours of the original request, the same key returns the same charge object from cache without hitting Stripe’s payment network again. This makes Stripe calls safe to replay within the 24-hour window, provided the idempotency key is stable. The failure modes in this post all arise from idempotency keys that are not stable across Storm’s replay mechanisms (new tuple ID, new partition index, new worker). The fix is a stable content-hash key, with the pre-flight check as the backstop for replays outside the 24-hour window.
Does ON CONFLICT DO NOTHING handle concurrent bolt instances racing on the same replayed tuple?
Yes. PostgreSQL evaluates the unique constraint under serialization: if two billing bolt instances both attempt to insert a row for the same (customer_id, billing_period) concurrently — which can happen when Storm’s task migration during a worker restart temporarily assigns the same partition to two workers — one INSERT succeeds and the other no-ops silently. The winning instance proceeds to call stripe.charges.create(); the losing instance finds zero rows returned and acks the tuple without a Stripe call. The content-hash idempotency key provides a second layer if both instances pass the pre-flight check in the same millisecond before either INSERT commits.
Put the brakes on your agent’s Stripe key
Keybrake proxies your agent’s Stripe calls with per-period spend caps, endpoint allowlists, and a full audit log. If a billing bolt fires twice, the vault key cap absorbs the second charge before it reaches Stripe.