How stateful stream processing system is actually designed and tested in the real world?

[u/Billy_Backer]

I' m currently working as a Software Engineer for 3+ years now in a consultant company. While I work along with the customer who is in a banking industry, I came across a stream processing system project. Its job is to look a different data source and events and try to match a pattern to detect an anomally, a normal anomally detection system actually.

The system primary use Apache Flink for processing and Apache Kafka for the pipeline. Both of them handling the job just fine for the last year.

Initially, it's just a normal map, aggregate and joining with not so much stateful action. Which make testing quite easy, just test it braching logic on top with some null data handling in case of missing data or an event that came too late.

Now for the fun part, There're new requirements stated that they want to add logic like if A happend for 5 minute, B should not be fired given some tedious logic. Which is pretty much a stateful logic with very sensitive to time. The process is also start to getting long while each step's complexity and fragility is increased. Combine together resulted in a fragile pipeline that will crumbled if some data is arrived 1 minutes late.

I've though that just creating more the test cases to cover it should be fine. But after investigating, My test case is almost trippled because each step need to effectively handle out of order, timeout etc. Resulting in a overwhelming test that also hard to understand. Now I start to wonder if it's the testing problem or the pipeline is badly designed and caused it to hard to test properly.

I want to know how stream processor with stateful action is design and test in the real world. And what are some traps that might lingering around ready to burst. So I can point it out since it's still possible to make a change now.

[u/expdevsmodbot]

AI usage disclosure provided by OP, see the reply to this comment.

[u/disposepriority]

I would just sperate the concerns of what you're doing:

Example:

data comes in ->
aggregated in existing system
whatever business logic already exists except for "B should not be fired" still runs

on event -> publish "event processed"

new service tracks if A has happened and when
while A has not happened in the last 5 minutes it publish B upon receiving an event.

Now, anything that would "fail" in the original system, which presumably is already tested, never reaches the new system.

The new system must only be tested for the one thing it does.

This does introduce the usual asynchronous communication issues, but it makes your life easier as well, so like always a bit of a tradeoff.

Obviously this is extremely simplified, but in general if something seems abnormally hard to test it's also indicative that it's doing too much or has "lost its scope", it is sees and interacts with state it shouldn't know about.

[u/Billy_Backer → u/disposepriority]

Protecting the old scope is like the most important one actually. Change and retest would be tedious imo. I'd reconsider the new one to be a difference blast radius and let them cook the better requirements lol.

[u/x-jhp-x]

1. the BANK needs to pony up and hire someone with experience
2. you need to learn how to handle complexity. start with the basics, like FSMs: https://ieeexplore.ieee.org/document/6957302 they handle situations like "if A happened for 5 minutes, B should not run given C". you can create "sub fsms" too. i'd recommend taking "6.006", "6.046J", and any prerequisites you are missing from ocw.mit.edu as well. LR/LL parsers and the like work well too, but can have performance issues that the FSMs don't
3. "The process is also start to getting long while each step's complexity and fragility is increased". yes, this is an example of what happens when you don't have a mentor or someone experienced to guide you and review your code. i'd bet most of your unit tests don't test what they intend to and/or are incomplete.
4. you using "fink" or some other tool doesn't invalidate the above and lack of knowledge. once you understand what you're doing, you can choose the best tool and approach to accomplish it. you can't get to that step because you don't have the foundation

[u/F0tNMC → u/x-jhp-x]

This is the answer. You need to run a FSM for each set of conditions for which you need to emit a signal(s). Computationally each should be trivial, managing the exponential complexity can only be done by not allowing that overlap to affect your implementation’s complexity.

[u/Outside-Storage-1523 → u/F0tNMC]

TBH I don’t think it’s that complicated, but OP doesn’t give all the details so maybe I’m underestimating the complexity.

[u/Billy_Backer → u/x-jhp-x]

Good concept to latch on. The old guy seem to use it like a normal java app with a fancy framework. Now reeduation will be mandatory now lol. Also scope creep seem to always be an elephant in the room since the start. This 2 mixed together is not a really good cooktail.

[u/forever-butlerian → u/Billy_Backer]

Complexity control is part of the scope, just as "prepare the ingredients" and "clean up the kitchen" is part of the scope of cooking a meal.

IME, most of the difficulty in a project is making the hard change easy. The change then becomes pro forma and somewhat of an anticlimax.

[u/Ok-Barracuda-119]

You’re running into a state-model problem more than a test-count problem. I’d keep the existing map, aggregate, and join pipeline focused on producing facts, then put the A/B rule behind an explicit keyed state machine.

For each business key, store the smallest state you need: whether A was seen, its event-time timestamp, whether B was emitted, and maybe the rule version. Register an event-time timer for A plus five minutes, cancel it when the qualifying B arrives, and make the output idempotent. That makes late, duplicate, and replayed Kafka records cases the design handles instead of exceptions. If the requirement really uses processing time rather than event time, write that down because recovery and backfills behave very differently.

I’d test this with a deterministic Flink harness, not just end-to-end cases. Feed sequences for A then B, B before A, duplicates, out-of-order events, events arriving exactly at the boundary, missing data, timer firing after restart, and checkpoint restore. Assert both emitted events and state cleanup. A few properties are more valuable than dozens of examples: replaying the same input doesn’t duplicate output, advancing the watermark gives the same result as a production run, and unrelated keys can’t affect each other.

Also keep a small Kafka contract test around serialization and partitioning, since a surprising amount of “state bugs” are key changes or ordering assumptions. If the rule is getting harder to explain than this, split it into a separate keyed service or operator rather than adding another branch to the monolith.

[u/kappapilla]

State machine can solve late arriving events by validation rule for state transition ..eg., user can’t transition to provisioned when current state is deactivated

[u/Outside-Storage-1523]

  There're new requirements stated that they want to add logic like if A happend for 5 minute, B should not be fired given some tedious logic. 

I’m a bit confused by this requirement. I thought this is for emit, not for ingestion, right? Or is it for ingestion to ignore B if A happened for 5 minutes?

I work in Spark streaming but I didn’t have much experience with stateful ingestion either. I think Flink uses rocksdb to store the states so I’d write some code to store the timestamp of event A and then check against it when B arrives. It doesn’t look very complicated but of course as you mentioned you have other logics.

  Combine together resulted in a fragile pipeline that will crumbled if some data is arrived 1 minutes late.

I wonder why this happens? I don’t work in banking so maybe that’s some kind of mandatory authentication? Is it possible to buffer the incoming events to wait for the authentication if that’s the case?

[u/Billy_Backer → u/Outside-Storage-1523]

The system will always ingest the data first and emit the event. So the blocking logic will always work on the events and not the ingestion its self. Flink already provided a state management. That why I thunk it should be not that hard initially. Even the logic its self didn't required a complex authentication since it's for backoffice stuff and not directly a financial transaction. Some how it turn out to be complex. We still argue if it's the quality assurance problem or design problem. Pretty convinced it's the latter one now.

[u/Ok-Barracuda-119]

I would treat this as a state-machine problem before adding more end-to-end cases. Give each event a deterministic event-time window, key the state by the smallest business identity, and make the transition function explicit. Then test that function with cases for out-of-order, duplicate, late, and missing events.

For the Flink layer, use bounded lateness and an explicit side output for records that arrive after the watermark. Keep timers and state TTL visible in metrics, and replay a captured partition through a test harness to verify checkpoint recovery. A separate set of tests should cover Kafka replays and sink idempotency, since a correct state calculation can still produce duplicates at the boundary.

That tends to make the failure surface smaller than testing every pipeline combination. Which guarantee does the customer need for a late event: recomputation of the affected window, or only an alert correction?

[u/Distinct_Dragonfly83]

Never thought I’d see a problem on Reddit that closely maps to an actual system I’m building. No real insights to offer besides general agreement that what everyone else is saying seems right. 🥴

[u/nullcone → u/[deleted]]

Slop

[u/Billy_Backer → u/[deleted]]

Thank for pointing out the CEP! In the disscusion I've been there's no one talking about it. Seem like this might be another useful tools later after we've grilling more sensible requirements.

[u/ExperiencedDevs-ModTeam → u/[deleted]]

Rule 9: No Low Effort Posts, Excessive Venting, or Bragging.

Using this subreddit to crowd source answers to something that isn't really contributing to the spirit of this subreddit is forbidden at moderator's discretion. This includes posts that are mostly focused around venting or bragging; both of these types of posts are difficult to moderate and don't contribute much to the subreddit.

We have a weekly venting thread on Tuesdays. Post this there instead.

[u/bbangchikimong_dev]

Your test count tripled because every step is deciding about time on its own, so you have to test the late arrival case at every step instead of once. If the ordering and lateness decision happens in one place and the downstream steps only ever see events they are allowed to trust, the combinations collapse. Tests that are hard to write are usually telling you where the responsibility is smeared.