We’ve all experienced that moment of relief when an ATM successfully dispenses our cash. But behind that simple transaction lies a complex dance of systems working together. In this post, I’ll walk you through how I tackled the challenge of processing cash dispense requests at scale using Apache Kafka and Python’s asyncio.

This approach grew out of a real problem I faced at work recently - how to allow multiple payment transactions to happen simultaneously across different locations while making sure we never had conflicting operations at any single cash machine. The solution I developed turned out to be both elegant and robust (at least I think so 😅) combining message streaming with a simple concurrency control mechanism that I’ll try to explain in this blog post.

My Journey to a Solution

My first approach was simply implementing multi-threading in Python, but knowing about multithreading in Python and the Global Interpreter Lock (GIL) deemed it not a good solution. The GIL prevents true parallel execution of Python threads, which would defeat the purpose of trying to scale our transaction processing.

That approach quickly fell apart when I tested it. With multiple workers processing messages independently, I ended up with redundant hardware operations - the same cash machine trying to dispense money twice for the same transaction! Not exactly what you want in a financial system.

So I swung to the opposite extreme. “What if I just make everything strictly synchronous?" Sure, it would be slower, but at least transactions would be recorded with proper balances and we wouldn’t have machines dispensing phantom cash.

But forcing everything to be sequential created a massive bottleneck. Transactions at Location A had to wait for transactions at Location B to complete, even though they had nothing to do with each other. Users were waiting unnecessarily, and the system couldn’t scale.

That’s when I realized I needed something in between - parallel processing where it’s safe, but strict controls where it matters.

My Solution: Location-Based Semaphores with Asyncio

The heart of our solution combines Kafka’s distributed messaging with Python asyncio’s concurrency primitives to create a system that processes transactions in parallel while maintaining safety constraints.

In simpler terms, we built a system that can handle many cash transactions at once (like a multi-lane highway) but makes sure that no two transactions try to access the same ATM at the same time (no lane crossing). This gives us speed without sacrificing safety.

Let me walk through the individual components that make this possible:

Key Components

1. Location-Based Concurrency Control

1
2
3
4
5
6

# Dictionary to track location-specific tasks
location_tasks = {}
# Dictionary to track location-specific semaphores
location_semaphores = {}
# Default maximum concurrent transactions per location
MAX_CONCURRENT_PER_LOCATION = 1

This approach creates a dedicated semaphore for each location, allowing the system to process transactions from different locations in parallel while ensuring only one transaction executes at a time for any given location. Like having independent checkout lines at different momo stalls, customers can get tasty momos served simultaneously across the network, but each individual stall carefully manages its own line to prevent confusion.

2. Dynamic Task Management

1
2
3
4
5
6
7
8

# Create a semaphore for this location if it doesn't exist yet
if location not in location_semaphores:
    location_semaphores[location] = asyncio.Semaphore(MAX_CONCURRENT_PER_LOCATION)
    location_tasks[location] = []

# Create and track the task
task = asyncio.create_task(process_with_semaphore(data))
location_tasks[location].append(task)

Think of this part as an organized filing system that automatically creates a new folder the first time we encounter a new ATM location. Then, whenever a customer wants to use an ATM, their request gets placed in the appropriate location’s folder and tracked until completion. Once their transaction is done, we remove their paperwork from the folder to keep things tidy - no clutter, no lost transactions.

3. Semaphore-Protected Processing

1
2
3
4
5
6
7
8
9

async def process_with_semaphore(msg_data):
    try:
        async with location_semaphores[location]:
            return await process_message(msg_data)
    finally:
        # Clean up - remove this task from the location's task list when done
        task = asyncio.current_task()
        if location in location_tasks and task in location_tasks[location]:
            location_tasks[location].remove(task)

Here’s where the magic happens!🪄Think of this like a security gate that only allows one person through at a time. When a transaction arrives, it waits its turn until it gets the green light to proceed. Once it’s done, it cleans up after itself by removing its record from our tracking system.

The benefit of this approach? Transactions from different locations run simultaneously, but transactions for the same ATM politely wait their turn. And if something goes wrong, we still clean up properly - no loose ends!

Kafka Integration and Resilience

Our solution leverages Kafka’s robust message queuing capabilities, which forms the backbone of our cash dispensing architecture. When a transaction comes in, Kafka ensures it’s not lost and can be properly tracked even if something goes wrong. I implemented a consumer group that processes these messages asynchronously:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

async def consume_dispense_messages():
    while True:
        try:
            async with current_app.app_context():
                current_app.kafka_consumer = await create_consumer(group_id="cash-dispense-service-group")

                async for message in current_app.kafka_consumer:
                    # Extract location from message data
                    data = json.loads(message.value) if isinstance(message.value, str) else message.value
                    location = str(data["location"])

                    # Create semaphore & process with location protection
                    if location not in location_semaphores:
                        location_semaphores[location] = asyncio.Semaphore(MAX_CONCURRENT_PER_LOCATION)
                        location_tasks[location] = []

                    # Create and track the task
                    task = asyncio.create_task(process_with_semaphore(data))
                    location_tasks[location].append(task)

The system handles connection failures gracefully with retry mechanisms and uses a combination of try/except blocks for error handling. If something breaks mid-transaction, our error recovery kicks in:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

except ConsumerStoppedError:
    current_app.logger.warning("Consumer stopped, attempting to restart...")
    await cleanup_consumer(current_app.kafka_consumer)
    current_app.kafka_consumer = None
    await asyncio.sleep(5)  # Wait before trying again
finally:
    # Cancel all pending tasks and clean up
    for loc, tasks in location_tasks.items():
        for task in tasks:
            if not task.done():
                task.cancel()

    # Wait for tasks to complete or be cancelled
    all_tasks = []
    for loc_tasks in location_tasks.values():
        all_tasks.extend(loc_tasks)

    if all_tasks:
        await asyncio.gather(*all_tasks, return_exceptions=True)

This comprehensive approach means our system automatically restarts after failures, cleans up resources properly during errors, and maintains detailed logs for troubleshooting. Kafka’s distributed nature complements our semaphore approach perfectly - it handles the reliable message delivery while our concurrency controls manage the safe execution of those messages.

Conclusion

So there you have it - by combining Kafka’s distributed messaging with asyncio’s concurrency controls, I built a system that handles cash operations at scale while keeping transactions fast and accurate (till date 🙏)

Could I have solved this with a simpler approach? Probably. But since Apache Kafka was already the backbone of our microservice architecture, it made sense to leverage what we had rather than introduce yet another technology into the mix. Does Kafka add its own complexity? Absolutely - the learning curve is steep, and operational overhead is real. But for our needs, this complexity was a necessary evil when weighed against the performance and scalability benefits.

This pattern has already proven useful beyond just cash dispensing. We’ve applied similar techniques to activity logs generation and storage where we needed that careful balance between parallelism and safety constraints.

I’m pretty satisfied with how this turned out. Through use of semaphores, careful task tracking, and robust error handling, I built something that’s both fast and reliable. And when you’re dealing with people’s money, that’s exactly what you need.😉

Let me know your thoughts in the comments below if you spot any error-prone regions in my implementation. I’m curious - how would you approach this problem? Would you go with a similar Kafka + asyncio solution, or take a completely different path? 🤔