I went backwards and forwards for some time on whether or not to in any respect. The work is enterprise knowledge integration: wiring the information from a number of separate enterprise methods collectively by means of a pipeline. Orders, stock, finance, logistics, buyer data, plus a pile of legacy FTP batch channels no one needs to the touch. Greater than twenty methods on the 2 ends of it. Just a few million occasions a day, a number of occasions that at month-end shut and through massive gross sales pushes.<\/p>\n
It sounds easy. System A calls system B\u2019s API, what\u2019s the large deal. Anybody who has truly achieved this is aware of the annoying half isn\u2019t getting A to speak to B. It\u2019s protecting it appropriate after it\u2019s speaking. Of the twenty-odd methods, some are new and converse REST, some had been outsourced ten years in the past and solely converse SOAP, and at the least one solely is aware of find out how to drop a file over FTP. The stacks are far and wide and the reliability is far and wide, and when one thing breaks it lands on you, since you\u2019re the layer within the center.<\/p>\n
This text is concerning the third of three issues that pipeline compelled me to unravel, and the one individuals often attain for first and get flawed: throughput. The pipeline has to maintain day-to-day latency below about half a second and soak up roughly ten occasions regular quantity at peak, which in apply means tens of 1000’s of occasions a second on an strange peak and greater than that in a sale. The lure is that just about the whole lot you do to go quicker can also be a option to silently break the information, and as soon as the information is flawed you discover out about it weeks later, from finance, throughout a reconciliation, which is the worst attainable time. So I can\u2019t speak about pace with out first being clear concerning the ground I wasn\u2019t allowed to drop beneath.<\/p>\n
A observe on the place the numbers come from<\/h2>\n
Earlier than any of the figures beneath, it\u2019s value being sincere about what sort of numbers they’re. The whole lot I quote is a consumer-side runtime metric taken from the stay pipeline throughout regular operation, not a managed benchmark on a clear cluster. Throughput is occasions processed per second measured on the client, learn off throughout strange business-hour site visitors quite than at peak; once I say a charge was \u201csecure\u201d I imply it held inside regular variance throughout full enterprise cycles, not that I pinned it in a single run. The batch-size comparability later (50, 100, 200, 500) was run in opposition to actual manufacturing load, not artificial knowledge, which is why the reply is particular to this workload and never a common fixed. The place a determine is softer than it appears to be like, I say so. These numbers had been collected throughout a number of month-end shut and peak-sales cycles of regular operation, not in a single benchmark run. I\u2019m reporting an expertise, not a examine, and the worth of it’s within the failure modes and the trade-offs, not in a benchmark you may rerun.<\/p>\n
The ground: what scaling is just not allowed to interrupt<\/h2>\n
Two ensures sat beneath each throughput change, and each one of many optimizations later on this article is constructed so it could possibly\u2019t violate them.<\/p>\n
The primary is {that a} later model of an entity\u2019s state can by no means be overwritten by an earlier one. In a distributed pipeline the identical logical replace arrives greater than as soon as and out of order, on a regular basis. Community retransmits, queue redelivery, a client restart mid-flight, an upstream timeout-and-resend. You may\u2019t cease any of that from taking place, so the one transfer is to make the write path detached to it. Each entity carries a model quantity that the supply system owns (not one the pipeline invents, as a result of the pipeline has no thought when the supply truly modified one thing), and the write rejects something stale:<\/p>\n
public void upsertWithVersionCheck(EntitySync sync) {
\n int up to date = jdbcTemplate.replace(
\n “UPDATE entity_store SET knowledge = ?, model = ?, updated_at = NOW() ” +
\n “WHERE entity_id = ? AND entity_type = ? AND model < ?”,
\n sync.getData(), sync.getVersion(),
\n sync.getEntityId(), sync.getEntityType(), sync.getVersion()
\n );
\n if (up to date == 0) {
\n \/\/ both a brand-new row to INSERT, or an older model we should always drop
\n strive {
\n jdbcTemplate.replace(
\n “INSERT INTO entity_store (entity_id, entity_type, knowledge, model) ” +
\n “VALUES (?, ?, ?, ?)”,
\n sync.getEntityId(), sync.getEntityType(),
\n sync.getData(), sync.getVersion());
\n } catch (DuplicateKeyException e) {
\n \/\/ a more moderen model already landed; dropping this one is appropriate
\n }
\n }
\n}<\/p>\n
It\u2019s mainly a stripped-down last-write-wins the place \u201cfinal\u201d means highest model, not most up-to-date arrival. That one rule is what lets me be aggressive about parallelism later with out mendacity awake about ordering.<\/p>\n
The second assure is that \u201cdid we already course of this?\u201d can by no means be flawed. Each accepted document writes its dedup-log entry and its enterprise knowledge in the identical database transaction, in order that they commit collectively or by no means. The dedup log is the one supply of fact for what was accepted, and it isn\u2019t allowed to float from the information it claims to explain. Early on we did the dedup test up within the enterprise code, question first then write, and at excessive concurrency the hole between the 2 let duplicates slip by means of. The repair was to push it right down to a primary-key constraint and let the database inform us. (That log desk grows without end should you let it; a nightly job trims entries older than thirty days, which is generously previous the window the place redeliveries truly occur.)<\/p>\n
I\u2019m spending these few paragraphs on correctness as a result of the whole lot beneath trades in opposition to it, and the trades are solely protected as a result of this ground holds.<\/p>\n
Partitioning, and the entity that\u2019s 100 occasions louder than the remainder<\/h2>\n
Extra partitions means extra parallelism, however it additionally means extra probabilities for occasions to be processed out of order throughout partitions. The rule I settled on is that each occasion for a similar entity goes to the identical partition, keyed by entity ID. Identical entity, identical partition, naturally so as, no cross-consumer coordination to motive about.<\/p>\n
That works proper up till one entity isn\u2019t just like the others. We had a single giant account producing updates at one thing like 100 occasions the speed of a traditional one. The whole lot for that account hashed to at least one partition, so one client was buried whereas its neighbors sat idle, and including shoppers did nothing, as a result of the bottleneck was one partition, not whole capability.<\/p>\n
The repair was to sub-partition the new ones. For entities we all know are sizzling, the important thing will get a second element so their site visitors spreads throughout partitions as an alternative of piling onto one:<\/p>\n
public class AdaptivePartitioner implements Partitioner {<\/p>\n
non-public last Set hotEntities; \/\/ maintained within the background<\/p>\n
@Override
\n public int partition(String subject, String key, byte[] worth, Cluster cluster) {
\n int numPartitions = cluster.partitionCountForTopic(subject);
\n String entityId = extractEntityId(key);
\n if (hotEntities.accommodates(entityId)) {
\n \/\/ sizzling entity: break up it finer by entityId + eventType
\n String fineKey = entityId + “:” + extractEventType(key);
\n return Math.abs(fineKey.hashCode()) % numPartitions;
\n }
\n \/\/ regular entity: key by entityId so its occasions keep ordered
\n return Math.abs(entityId.hashCode()) % numPartitions;
\n }
\n}<\/p>\n
The hotEntities set isn\u2019t hard-coded. A background job samples per-entity charges each hour and strikes an entity in when it crosses a threshold and again out when it cools off. Spreading a sizzling entity throughout partitions does reintroduce some out-of-order danger for that entity, however that\u2019s precisely what the model test from the earlier part is there to soak up. If v1 reveals up after v2 as a result of they took completely different partitions, the write drops v1 and the ultimate state remains to be proper. That is the sample for the entire article: I\u2019m allowed to loosen up ordering right here solely as a result of correctness is enforced one layer down.<\/p>\n
Micro-batching, which is the place the pace truly comes from<\/h2>\n
Processing one document at a time is sluggish, and it\u2019s sluggish in two particular locations: a community round-trip to the database or a downstream API for each single occasion, and a separate database transaction per occasion with the commit value that suggests. Neither is CPU. You may throw shoppers at it without end and never transfer the quantity.<\/p>\n
So we batch. Accumulate a small group, 100 data or fifty milliseconds, whichever comes first, then deal with the group in a single shot:<\/p>\n
public class MicroBatchConsumer {<\/p>\n
non-public static last int BATCH_SIZE = 100;
\n non-public static last Period BATCH_TIMEOUT = Period.ofMillis(50);<\/p>\n
non-public void processBatch(Checklist> batch) {
\n \/\/ 1) dedup the entire batch in a single question, not N queries
\n Set keys = batch.stream()
\n .map(r -> r.worth().getIdempotentKey())
\n .acquire(Collectors.toSet());
\n Set current = dedupRepository.findExistingKeys(keys);<\/p>\n
Checklist newEvents = batch.stream()
\n .map(ConsumerRecord::worth)
\n .filter(e -> !current.accommodates(e.getIdempotentKey()))
\n .toList();<\/p>\n
\/\/ 2) one transaction, with a savepoint per document so one dangerous
\n \/\/ document would not take the opposite ninety-nine down with it
\n jdbcTemplate.execute((Connection conn) -> {
\n conn.setAutoCommit(false);
\n for (IntegrationEvent occasion : newEvents) {
\n Savepoint sp = conn.setSavepoint();
\n strive {
\n processOne(conn, occasion);
\n } catch (Exception e) {
\n conn.rollback(sp);
\n dlqProducer.ship(occasion, e);
\n }
\n }
\n conn.commit();
\n return null;
\n });
\n }
\n}<\/p>\n
The impact is just not delicate. Single-record processing held round 500 occasions a second. Micro-batched, the identical pipeline held round 8,000, name it a sixteen-fold bounce, and the reason being virtually totally {that a} hundred round-trips collapsed into one or two.<\/p>\n It prices you two issues. One is as much as fifty milliseconds of additional latency whereas the batch fills, which for second-scale workloads is nothing. The opposite is that batch failure is now an actual query: if one document within the batch blows up, what occurs to the remainder? Rolling again the entire batch and retrying it’s wasteful, so every document sits in its personal savepoint, and a failure rolls again solely that document, ships it to the dead-letter queue, and lets the remainder commit. That solely works as a result of the dedup-log write and the enterprise write rewind collectively contained in the savepoint; in the event that they didn\u2019t, a rollback would depart a dedup entry with no knowledge behind it, or the reverse, and the subsequent retry would make the flawed resolution.<\/p>\n The batch dimension and timeout are tuned, not guessed. We tried 50, 100, 200, and 500. 100 gained. Previous that the throughput curve flattens, and worse, the IN clause on the batch dedup question will get lengthy sufficient that the question planner begins making dangerous decisions and the database offers again greater than the round-trips saved. Greater is just not higher right here; it\u2019s higher up to a degree that it’s important to discover in opposition to your personal dedup question, after which it\u2019s worse.<\/p>\n The factor a high-throughput pipeline ought to truly be afraid of isn\u2019t falling behind. It\u2019s falling behind with out figuring out it. If the upstream stays quicker than the downstream, the backlog grows with out sure till a disk fills or a client runs out of reminiscence. So consumption has to have the ability to push again, in three tiers, every for a special method it goes flawed.<\/p>\n The primary tier is the buyer slowing itself down. It watches its personal processing latency and throttles its personal ballot charge when it sees itself getting slower:<\/p>\n public class AdaptiveRateLimiter {<\/p>\n non-public last MovingAverage latencyAvg = new MovingAverage(100); public void recordLatency(lengthy ms) { public Period getPollDelay() { The second tier watches client lag per partition from outdoors the buyer and feeds a charge restrict again to the producers by means of the config service. It isn\u2019t a well mannered request: producers test the restrict earlier than sending and buffer regionally after they\u2019re throttled, so the brake truly holds.<\/p>\n The third tier is for when the downstream is genuinely in bother and the backlog can\u2019t be labored off. Occasion sorts are ranked by enterprise precedence after they\u2019re first onboarded, not in the midst of an incident, and below actual downstream failure the low-priority sorts are suspended (saved within the queue, simply not consumed) so the entire fleet\u2019s capability goes to the occasions that matter. Order-state and stock writes are high precedence; overview syncs and historic backfills aren’t. The rating has to exist earlier than the outage, as a result of the one factor you’ll be able to\u2019t do reliably at 2 a.m. is determine what\u2019s vital.<\/p>\n One throughput downside value singling out, as a result of it didn\u2019t begin within the pipeline in any respect. A client had an HTTP connection pool of fifty connections to at least one downstream. The downstream later break up learn and write onto two hostnames. We up to date the code and forgot the pool config, so fifty connections obtained divided throughout two hosts, twenty-five every. At peak the pool ran dry, requests queued ready for a connection, and latency went by means of the roof.<\/p>\n It took a very long time to seek out, and the explanation it took a very long time is the symptom lied. The error wasn\u2019t \u201cconnection refused,\u201d it was \u201crequest timed out,\u201d as a result of each request was sitting within the pool\u2019s wait queue till it gave up. Tail latency spiked whereas the error charge stayed flat, and when you\u2019ve seen that signature when you acknowledge it: a downstream that’s itself sluggish raises errors too, however pool hunger raises latency with no errors, as a result of nothing has failed but, it\u2019s all simply ready.<\/p>\n We added pool monitoring after that, utilization and wait-queue depth and an alert when utilization sits above eighty p.c, and made it a rule that downstream structural modifications (a hostname break up, a load-balancer change) must be advised to the mixing workforce, as a result of to us they don’t seem to be an implementation element, they\u2019re a capability occasion.<\/p>\n Right here\u2019s the entire thing in a single actual incident, as a result of the three considerations are by no means truly separate when one thing breaks.<\/p>\n Two within the afternoon, an alert: order-domain client lag climbing from a number of hundred milliseconds previous 5 minutes and nonetheless rising, and on the identical time the ERP API error charge going from below one p.c to forty.<\/p>\n For the primary two minutes no one touched something. The circuit breaker noticed the error charge cross its threshold and opened, slicing requests to ERP; occasions that couldn\u2019t be processed went to the retry queue, and backpressure dropped the buyer ballot charge by about sixty p.c by itself. That was the primary line of protection and it was imagined to be computerized.<\/p>\n Minutes two by means of ten had been prognosis. The on-call engineer logged in, noticed the order-domain breaker open and ERP\u2019s well being checks all purple, and obtained affirmation from the ERP workforce: a database migration, about thirty minutes to restoration.<\/p>\n Thirty minutes meant an actual backlog, so minutes ten by means of fifteen had been the deliberate half: the on-call triggered the order-domain shedding coverage, suspended the non-core sorts (overview sync, historic backfill), and let the shoppers consider order-state and stock. The core occasions waited within the retry queue for ERP to return again.<\/p>\n When ERP recovered, the breaker went half-open, tried a number of requests, confirmed they had been effective, and closed. The retry-queue backlog replayed, and since each processing path is idempotent, replaying it was protected, no particular dealing with for the duplicates that replay inevitably produces. Backpressure eased off and the ballot charge got here again to regular.<\/p>\n That night the offline reconciliation put numbers on it: 23,000 occasions affected, 22,987 replayed and processed robotically, 13 within the dead-letter queue from soiled knowledge written throughout ERP\u2019s migration window, dealt with by hand the subsequent morning. Core enterprise noticed at most two minutes of interruption, the 2 minutes earlier than the breaker tripped. Non-core was suspended about forty minutes. Zero knowledge misplaced. The one two human choices in the entire sequence had been confirming the trigger and selecting to shed; the whole lot else the pipeline did itself.<\/p>\n Not one of the particular person items listed below are new, and it\u2019s value saying what they descend from, as a result of the contribution isn\u2019t anybody mechanism. The new-entity downside particularly has an actual literature. Partial Key Grouping [1] confirmed you’ll be able to steadiness a skewed key stream by giving sizzling keys a selection of two staff as an alternative of 1, and the follow-up work [2] identified that for the very heaviest hitters two decisions aren\u2019t sufficient and it is advisable to unfold them wider. Later work folded skew-aware key splitting straight into micro-batch stream processing [3]. My adaptive sub-partitioning is a blunter, operations-driven cousin of that line of labor: I\u2019m not computing an optimum break up, I\u2019m keying off a background hot-set with a charge threshold and accepting some reordering as a result of the model test downstream makes that reordering protected. The tutorial schemes optimize steadiness; I\u2019m optimizing for \u201cadequate with out a coordination protocol I\u2019d must function at 2 a.m.\u201d<\/p>\n The bigger framing, that \u201cexactly-once\u201d in a distributed pipeline is admittedly effectively-once and rests on idempotency quite than on never-deliver-twice, is Helland\u2019s [4], and it\u2019s the idea your entire correctness ground leans on. The survey literature catalogs the remainder of the transferring elements: out-of-order dealing with, state administration, fault tolerance, and cargo administration are specified by the stream-processing evolution survey [5], and the still-open query of bolting transactional ensures onto streaming is surveyed in [6], which is kind of the issue this pipeline solves by hand with a model column and a savepoint quite than with a normal mechanism. Backpressure as a first-class sign quite than an afterthought traces to the Reactive Streams line of pondering [7], and the foundational therapy of why all of that is onerous sits in Kleppmann [8].<\/p>\n The place this differs from the papers is the setting. The analysis principally assumes one streaming engine you management finish to finish. Enterprise integration doesn\u2019t offer you that. Half your upstreams are methods you’ll be able to\u2019t change, the model numbers must be generated by sources that predate the pipeline by a decade, and \u201cload shedding\u201d must be a business-priority resolution made earlier than the incident, not a sampling technique chosen by the engine throughout it. The worth right here, if there’s any, is in how these recognized strategies compose below a tough correctness ground while you don\u2019t personal the methods on both finish.<\/p>\n Throughput is the third requirement, not the primary. Correctness is what makes the enterprise belief the pipeline in any respect, resilience is what permits you to sleep whereas it\u2019s operating, and pace solely issues as soon as these two maintain. The onerous a part of integration work was by no means choosing a partitioning scheme or a batch dimension. It was discovering the steadiness between the three, as a result of pushing any one among them to its restrict prices you the opposite two: confirm each message 5 methods and you don’t have any throughput, skip the breaker checks for latency and you don’t have any resilience. Engineering right here is discovering the purpose that\u2019s adequate for the amount you even have and the methods you even have to speak to. Not the optimum one. The one that matches.<\/p>\n Yuelin Ou is a Knowledge & AI Engineer whose work focuses on idempotent write paths, distributed pipeline resilience, and scaling enterprise integration methods with out breaking correctness ensures. She holds a B.A. in Arithmetic with a minor in Pc Science from the College of Rochester. Web site: yuelinou.com.<\/p>\n [1] M. A. U. Nasir, G. De Francisci Morales, D. Garc\u00eda-Soriano, N. Kourtellis, G. M. Serafini, The Energy of Each Selections: Sensible Load Balancing for Distributed Stream Processing Engines (2015), Proc. thirty first IEEE Worldwide Convention on Knowledge Engineering (ICDE)<\/p>\n [2] M. A. U. Nasir, G. De Francisci Morales, N. Kourtellis, M. Serafini, When Two Selections Are Not Sufficient: Balancing at Scale in Distributed Stream Processing (2016), Proc. thirty second IEEE Worldwide Convention on Knowledge Engineering (ICDE)<\/p>\n [3] A. S. Abdelhamid, A. R. Mahmood, A. Daghistani, W. G. Aref, Immediate: Dynamic Knowledge-Partitioning for Distributed Micro-batch Stream Processing Programs (2020), Proc. 2020 ACM SIGMOD Worldwide Convention on Administration of Knowledge<\/p>\n [4] P. Helland, Idempotence Is Not a Medical Situation (2012), ACM Queue, vol. 10, no. 4<\/p>\n [5] M. Fragkoulis, P. Carbone, V. Kalavri, A. Katsifodimos, A Survey on the Evolution of Stream Processing Programs (2024), The VLDB Journal, vol. 33, no. 2<\/p>\n [6] S. Zhang, J. Soto, V. Markl, A Survey on Transactional Stream Processing (2024), The VLDB Journal, vol. 33, no. 2<\/p>\n [7] R. Kuhn, B. Hanafee, J. Allen, Reactive Design Patterns (2017), Manning<\/p>\n [8] M. Kleppmann, Designing Knowledge-Intensive Functions (2017), O\u2019Reilly<\/p>\n<\/div>\n![\"Bar](\"https:\/\/contributor.insightmediagroup.io\/wp-content\/uploads\/2026\/08\/throughput_before_after.png\")
Backpressure: the half that retains it from consuming itself<\/h2>\n
\n non-public unstable double throttleFactor = 1.0;<\/p>\n
\n latencyAvg.add(ms);
\n double avg = latencyAvg.get();
\n if (avg > 200) { \/\/ getting sluggish: again off
\n throttleFactor = Math.max(0.1, throttleFactor * 0.8);
\n } else if (avg < 50) { \/\/ loads of headroom: pace up
\n throttleFactor = Math.min(1.0, throttleFactor * 1.1);
\n }
\n }<\/p>\n
\n lengthy delayMs = (lengthy)((1.0 – throttleFactor) * 500);
\n return Period.ofMillis(delayMs);
\n }
\n}<\/p>\nThe bug that hid as a timeout<\/h2>\n
Placing all three collectively: one afternoon<\/h2>\n
![\"Timeline](\"https:\/\/contributor.insightmediagroup.io\/wp-content\/uploads\/2026\/08\/incident_timeline.png\")
How this strains up with the analysis, and the place it doesn\u2019t<\/h2>\n
What I truly take away from this<\/h2>\n
Concerning the writer<\/h2>\n
References<\/h2>\n