DEV Community: amaendeepm

I Fell for `?`. Then I Couldn't Debug Anything.

amaendeepm — Fri, 13 Mar 2026 03:48:37 +0000

The ? operator in Rust is genuinely beautiful. You go from a mess of match arms to something that reads like a sentence:

async fn process_report(payload: &[u8]) -> Result<(), AppError> {
    let envelope = parse_envelope(payload)?;
    let member   = resolve_member(&envelope.member_id)?;
    let positions = extract_positions(&envelope)?;
    persist_positions(&member, positions).await?;
    Ok(())
}

Clean. Compositional. I used it everywhere.

Then one day, records stopped being written to the database. No panic. No error log. No trace of anything going wrong.

The problem was obvious in hindsight. Any one of those four ?s could have triggered an early return. The error was propagating up to a caller that wasn't logging it properly. I had no idea which step was the culprit without manually re-running the pipeline with test fixtures.

The ? operator had silently swallowed my debugging surface.

The fix that worked:

let envelope = parse_envelope(payload)
    .map_err(|e| { tracing::error!(step = "parse_envelope", error = ?e, "failed"); e })?;

let member = resolve_member(&envelope.member_id)
    .map_err(|e| { tracing::error!(step = "resolve_member", error = ?e, "failed"); e })?;

Every potential exit point now announces itself. The error still propagates, the caller still gets it, but the point of propagation is now observable.

The lesson: ? is ergonomics for writing code. Logs are ergonomics for operating it. In any pipeline where a silent early return would cause missing data or unnoticed failures, wrap every ? with a log. The verbosity is worth it.

My Quest for Speed: How a Clickhouse Type Improvement Led Me Down a Caching Rabbit Hole in Rust

amaendeepm — Fri, 07 Nov 2025 01:44:35 +0000

It all started with a wonderful quality-of-life improvement in the clickhouse crate. The recent 0.14.0 release introduced support for the RowBinaryWithNamesAndTypes format. This was a game-changer!

Gone were the days of fetching data as untyped strings and manually converting them into Rust types like Decimal. Now, I could get strongly-typed rows directly, making my code safer, cleaner, and more pleasant to write.

But with great power comes great...greed.

Seeing my data pipeline become so elegant lit a fire in me. My Axum web application was now efficiently talking to ClickHouse, but I thought, "What if we could be even faster? Why hit the clickhouse database for every single request for spot prices?" The answer seemed obvious: cache it.

The Naive Beginning: Infinite Growth

My first instinct was to reach for the trusty std::collections::HashMap.
The plan was simple: store the fetched ClickHouse data in an in-memory cache.

use std::collections::HashMap;
use std::sync::RwLock;
use indexmap::IndexMap;

struct ApplicationContext {
    // Cache mapping symbols to their historical spot prices
    spot_price_cache: RwLock<HashMap<String, IndexMap<String, Decimal>>>,
}

I used IndexMap specifically to preserve the timestamp-order of price data, which was crucial for time-series analysis. I wrapped it in an RwLock for thread-safety, and... immediately saw a problem. This cache had no memory. It would grow indefinitely, consuming more and more RAM as we tracked more symbols and time periods. A cache with no eviction policy is just a memory leak with extra steps.

The Crossroads: lru vs. moka
It was time for a proper caching solution. A quick search led me to two popular contenders in the Rust ecosystem:

lru: A classic, bare-bones LRU (Least Recently Used) cache implementation.

moka: A feature-rich caching library inspired by Caffeine from the Java world.

I was confused, but after some evaluation, my argument for moka was compelling:

Simplicity: moka's constructor-based configuration felt intuitive. I could set up a cache with a maximum capacity in one line.

Thread-Safety: moka handles concurrency internally. No need to mess with Arc> semantics myself. This is a huge win for reducing boilerplate and potential deadlocks.

Observability: This was the clincher. moka provides built-in metrics like entry_count, hit_count, and miss_count. Being able to see how my cache was performing was invaluable for debugging and tuning.

The choice was clear. I was going with moka.

The Implementation: A Seemingly Robust Solution

I refactored my cache to use a moka::sync::Cache.


use moka::sync::Cache;
use indexmap::IndexMap;

#[derive(Clone)]
pub struct ApplicationContext {
    pub clickhouse_client: clickhouse::Client,
    pub spot_price_cache: Cache<String, IndexMap<String, Decimal>>,
}

impl ApplicationContext {
    pub fn new() -> Self {
        Self {
            clickhouse_client: create_clickhouse_client(),
            spot_price_cache: Cache::builder()
                .max_capacity(1000) // Hold up to 1000 symbol histories
                .build(),
        }
    }
}

The integration was smooth. I would check spot_price_cache for a symbol; on a miss, I'd query ClickHouse (using the lovely new typed interface), populate the cache with the IndexMap of timestamp-price pairs, and return the data. The cache metrics showed a beautiful story: a high hit rate after the warm-up period. The cache was working! The data provider was robust, and the logic was sound.

The Cold, Hard Truth: The Performance Letdown
But something was wrong. The overall response times for my spot price API endpoints hadn't improved. In fact, they had slowed down slightly.

Confused, I broke out the profiler and did rust flamegraph. The culprit wasn't the cache logic itself, nor the ClickHouse queries. It was the cloning.

Each time I served spot price data of a day from the cache, I was returning a clone of the entire IndexMap of intervals of that day. While IndexMap gives us ordering guarantees crucial for time-series data, it's generally less performant for cloning than a regular HashMap due to its additional internal structures for maintaining order. When you're dealing with hundreds of price points per day (96 entries in each IndexMap for 15-min intervals of the day), the allocation and data copying for every single request was creating significant overhead.

The cost of cloning the IndexMap was rivaling—and in some cases exceeding—the cost of a quick network round-trip to ClickHouse. The cache was "winning" logically with its high hit count, but losing the performance battle.

The Pragmatic Retreat: Feature-Flagging a Strategic Retreat

I did optimize the cache first by using Arc> to avoid the deep clone, but no IndexMaps are built differently and communicated across processes

Faced with this reality, I had to make a call. I could:

Consider smaller granularity: Cache individual price points instead of entire histories

Abandon the cache: Admit that for this specific use case, it wasn't the right tool

For now, I chose option two. The simplicity of the direct ClickHouse call was hard to beat. To keep my new, clean code but disable the caching overhead, I did what any pragmatic rust engineer would do: I feature-flagged it.

Lessons Learned

Strong Typing is a Pure Win: The clickhouse crate's new type support is fantastic and didn't cause my problem. It improved my code quality.

Caching is a Trade-off: It's not free. The overhead of storing and retrieving data from the cache must be significantly less than the cost of fetching the data fresh.

Data Structure Choice Matters: Using IndexMap over HashMap has real performance implications, especially when cloning entire structures. Preserving order has its cost!

Measure, Don't Assume: The cache metrics told me one story (high hits = good), but the overall application profiling told the real story (cloning overhead = bad). Always profile with real-world data sizes.

moka is Excellent: My initial assessment stands. For my next caching need where the value type is cheap to clone or can be wrapped in Arc, moka will be my first choice. Its API and observability features are top-notch.

My quest for ultimate speed hit a snag, but it was a valuable learning experience. The journey to optimize is often iterative, and sometimes the bravest move is to know when to step back and try a different path.

Why Composability (and Event Streaming Ingestion) Is the Backbone of Financial Reporting Systems for Metered Energy Data

amaendeepm — Wed, 18 Jun 2025 19:55:16 +0000

In the evolving landscape of energy markets, the reporting and financial settlement systems for metered data are expected to be agile, auditable, and automation-friendly. Yet too often, they’re rigid — siloed batch publishings of meter readings and there on stitched together with brittle ETL data pipelines

While at first glance it might look that we need a better or another analytics dashboard, what we actually need is a composable architecture, one that can ingest metered data as streams, apply domain-driven transformations, and support replayable, traceable pipelines that keep up with real-world complexity — including different resolution time intervals time series data, cancellations updates, other market updates, and shifting regulations plus time & volume dependent tariffs applications.

The Streaming-First Mindset for Metered Data

Financial settlement in energy isn't just about summing numbers. It’s about events — energy usage events, billing revisions, market conditions, regulatory pricing shifts of tariffs. A streaming-first architecture offers:

Low-latency ingestion of time-series meter data and market signals
Decoupled processing of diverse domains like credit notes, asset lifecycles, tradebooks, and exceptions
Replayable and audit-compliant pipelines, essential for dispute resolution and regulation
Composable microservices that act on streams, not snapshots

Traditional batch ETLs often fail in this landscape due to poor observability, latency, and lack of idempotency. With Apache Kafka and similar platforms, we can treat each meter read, each trade, and each regulatory trigger as a first-class event.

Composability Is Not a Buzzword

In my experience designing financial settlement systems, composability enables both performance and explainability!

Imagine a flow like this:

[Meter Reading Stream] → [Meter Mapper] → [Area Pricing Engine] → [Tradebook Join] → [Market Tariffs]→ [Invoice Generator]

Each stage is an independent component consuming and emitting events, capable of being redeployed, updated, or audited in isolation.

Add to this:

Kafka topics as source-of-truth logs instead of intermediate tables
Event-driven triggers to handle late data, cancellations, or re-rating without overhauling pipelines
Stateful stream processors (like Kafka Streams or Flink) that can persist intermediate state while remaining replayable
The result is a system that’s not just resilient but living and evolvable — a necessity in the post-renewables energy market where contracts, baselines, and obligations can all shift monthly.

Beyond ETL: Streaming as System Integration

Many teams treat Kafka like a glorified message bus. But in composable financial reporting systems,

It is the spine of integration.
Every topic is a contract, while schema evolution tells a story.
Every consumer lag is a performance metric.

In such systems:

Trade amendments can trigger partial invoice recalculations automatically.
Historical reruns (e.g. regulation backdates) are handled by replaying topics into stateful consumers.
Cross-domain data correlation happens in motion, not by nightly joins.

Why This Matters at Scale

When you’re dealing with:

Thousands of smart meters
Dozens of energy trading partners
Frequent retrospective adjustments, and
Strict auditing and compliance requirements;

composability powered by event streaming is not optional. It’s the only scalable way to keep reporting explainable, up-to-date, and testable.

Size Does Matter: Fixing Kafka BrokerTransportFailure in Redpanda

amaendeepm — Mon, 24 Mar 2025 15:29:58 +0000

If you're using Redpanda (or Kafka) and hitting mysterious consumer errors like:

Message consumption error: BrokerTransportFailure

You might be dealing with a message size limit issue! I ran into this recently, and increasing the fetch.message.max.bytes limit solved it instantly.

The Problem
My Rust consumer, built with rdkafka, was randomly failing with BrokerTransportFailure. Debugging didn’t reveal much rather going back n forth as everything otherwise with broker connection, topic, network reachability was all good; but after experimenting with different configurations, I realized the issue was message size constraints as this particular topic hosted quite large size of messages being received

Kafka/Redpanda has default limits on how much data can be fetched per request. If a message is too big, the broker refuses to send it, causing consumer failures.

The Fix
To resolve this, I had to increase the message size limits on both the the consumer. And that does not require any change in cluster or broker - but only in client code

The key setting here is fetch.message.max.bytes, which allows the consumer to fetch larger messages.


pub struct RPMessageProcessor {
    consumer: StreamConsumer,
}

impl RPMessageProcessor {
    pub fn new(group_id: &str, topic_name: &str) -> Self {
        let broker = env::var("REDPANDA_BROKERS").unwrap_or_else(|_| "localhost:9092".to_string());

        let consumer: StreamConsumer = ClientConfig::new()
            .set("bootstrap.servers", broker)
            .set("group.id", group_id)
            .set("enable.auto.commit", "false")
            .set("auto.offset.reset", "earliest")
            .set("fetch.message.max.bytes", "262144000") // Increased to 250 MB
            .create()
            .expect("Consumer creation failed");

        tracing::debug!("Streaming Consumer Created");

        consumer
            .subscribe(&[topic_name])
            .expect("Subscription failed");

        tracing::debug!("Subscribed to {}", topic_name);

        Self { consumer }
    }

    pub async fn run(&self, message_type: &str) {
        loop {
            match self.consumer.recv().await {
                Ok(msg) => {
                    let should_process = msg.headers().and_then(|headers| {
                        headers.iter().find(|header| {
                            header.key == "message-type"
                                && header
                                    .value
                                    .map(|v| v == message_type.as_bytes())
                                    .unwrap_or(false)
                        })
                    }).is_some();

                    if should_process {
                        if let Err(err) = Self::process_message(&msg).await {
                            tracing::error!("Error processing message: {}", err);
                        }
                    }

                    // Commit offset
                    if let Err(e) = self.consumer.commit_message(&msg, CommitMode::Sync) {
                        tracing::error!("Failed to commit offset: {}", e);
                    }
                }
                Err(err) => tracing::error!("Kafka error: {}", err),
            }

            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    }

    async fn process_message(msg: &BorrowedMessage<'_>) -> Result<()> {
        if let Some(payload) = msg.payload_view::<str>().transpose()? {
            tracing::info!("Processing message: {}", payload);
        } else {
            tracing::error!("Received empty or invalid message");
        }
        Ok(())
    }
}

How it is invoked?

RPMessageProcessor::new("CRE-RP-CG", &CONFIG.edx.input_topic) .run("CRE_HEADER_VALUE") .await;

Lesson Learned: Size DOES Matter!

If you’re running into Kafka or Redpanda consumer errors, check your message size limits! The broker, producer, and consumer need to be in sync to avoid failures

How Flamegraph Helped Me Optimize a Rust Application for Intensive Data Transformation and Migration

amaendeepm — Fri, 31 Jan 2025 14:41:29 +0000

Rust is having its own reputation for performance and speed. Therefore when working on performance-critical Rust applications, especially those dealing with intensive data transformation and migration, pushing for faster performance by identifying bottlenecks can be challenging. I faced this exact challenge with a Rust application responsible for processing and migrating large datasets. The initial implementation was slow, and I needed a way to pinpoint the performance issues. Flamegraph—a visualization tool that helped me optimize my code effectively though. And here's how I used Flamegraph to transform my application's performance

The Problem: Slow Data Migration
My Rust application was designed to migrate data from one system to another, involving complex transformations and large datasets. Despite Rust's reputation for performance, the process was taking hours to complete. I suspected inefficiencies in the code, but without concrete data, optimizing it felt like guesswork.

That's when I decided to use Flamegraph to visualize where the CPU cycles were being spent. Flamegraph provided a clear, hierarchical view of my application's execution, making it easy to identify hotspots.

What is Flamegraph?
Flamegraph is a profiling tool that generates an interactive SVG graph representing your application's call stack. The width of each stack frame corresponds to the amount of time spent in that function, allowing you to quickly identify performance bottlenecks.

For Rust, tools like cargo-flamegraph make it easy to generate flamegraphs. It integrates seamlessly with Cargo and provides a straightforward way to profile your application.

Setting Up Flamegraph for Rust
Here’s how I set up Flamegraph for my Rust application:

Install cargo-flamegraph:

cargo install flamegraph

Profile Your Application:
Run your application with cargo flamegraph. This generates a flamegraph SVG file.

cargo flamegraph --bin my_app

Analyze the Flamegraph:

Open the generated SVG file in your browser. The flamegraph shows a hierarchical view of your program's execution, with the most time-consuming functions at the top.

What the Flamegraph Revealed
When I first generated the flamegraph for my application, the results were dazzling to the eye, and my underlying curosity as well. Here's what I discovered:

Inefficient Database Queries:
The flamegraph showed that a significant amount of time was spent executing database queries. This was due to fetching large datasets in a single query without proper batching.

Expensive Data Transformation:
A hotspot was identified in the transmute_response function, which was responsible for transforming data. The function was processing data sequentially, leading to unnecessary delays.

Unnecessary Cloning:
The flamegraph highlighted that a lot of time was spent cloning data structures, particularly in the persist_identifiable_message function. This was due to passing owned data instead of references.

Optimizing the Code
With this information, I made the following changes to my application:

Batched Database Queries:
I refactored the database queries to fetch data in smaller batches. This reduced memory usage and improved query performance.

Parallel Data Transformation:
I parallelized the transmute_response function using futures::future::join_all to process multiple records concurrently. This significantly reduced the time spent on data transformation.

Reduced Cloning:
I refactored the code to avoid unnecessary cloning by using references wherever possible. This reduced memory allocations and improved overall performance.

Here’s a snippet of the optimized code:

let batch_size = 5;
for chunk in records.chunks(batch_size) {
    let futures = chunk.iter().map(|rsm34| {
        let transmuted_message = transmute_response(rsm34.complete_payload.clone().into()).unwrap();
        persist_identifiable_message(
            db_conn_pool.clone(),
            rsm34.peek_trace_id.clone().into(),
            transmuted_message.clone(),
        )
    });

    let results = join_all(futures).await;
    for result in results {
        if result.is_err() {
            panic!("Error {:?}", result.err());
        }
    }
}

The Results

After applying these optimizations, I regenerated the flamegraph to verify the improvements. The new flamegraph showed a much more balanced distribution of CPU time, with the previously identified hotspots no longer dominating the execution.

The most satisfying part? The data migration process, which previously took days, now seems to get complete much sooner. The application is not only faster but also more efficient in terms of memory usage.

Lessons Learned

Profiling is Essential:
Without profiling tools like Flamegraph, optimizing performance is a guessing game. Profiling gives you concrete data to work with.

Rust's Performance is Only as Good as Your Code:
While Rust is a high-performance language, it's still possible to write inefficient code. Tools like Flamegraph help you stay on track.

Optimization is Iterative:
Profiling and optimization are iterative processes. Each round of profiling can reveal new bottlenecks to address.

Give Flamegraph a Try!

If you're working on performance-critical Rust applications, I highly recommend giving Flamegraph a try. It's easy to set up, and the insights it provides are invaluable. Whether you're dealing with data transformation, migration, or any other CPU-intensive task, Flamegraph can help you identify and eliminate bottlenecks.

Axum in Rust: Flexibility, CORS Control, and Tower Power

amaendeepm — Mon, 02 Dec 2024 15:48:16 +0000

I started my Rust journey in August 2022, and it’s been an eye-opening experience. Rust is a language designed for performance and safety, but its web development ecosystem was still growing when I first encountered it. After experimenting with different frameworks and approaches, Axum stood out to me as a modern, intuitive solution.

With Axum, I’ve found building APIs not only productive but also enjoyable. Whether it’s implementing selective CORS controls, tracing requests with ease, or setting up API rate limits, the combination of Axum and Tower middleware just clicks.

CORS Control: Fine-Tuning Access

One of the first issues I had to tackle was implementing flexible CORS policies. I had an API with some public endpoints but also internal routes that needed restricted access. In many frameworks, managing this kind of fine-grained control can get messy. But Axum made this simple.

Using tower_http::cors, I could apply a global CORS policy while exempting specific routes, such as the /public route being open to all and /internal requiring strict access. Here’s a quick snippet of how I handled it:

use axum::{Router, routing::get};
use tower_http::cors::{CorsLayer, Any};
use tower::ServiceBuilder;

let cors_layer = CorsLayer::new()
    .allow_origin(Any)  // Open access to selected route
    .allow_methods(vec!["GET", "POST"]);

let app = Router::new()
    .route("/public", get(public_handler))
    .route("/internal", get(internal_handler))
    .layer(ServiceBuilder::new().layer(cors_layer));

Axum allowed me to keep my security intact without complicating the setup, which I hadn’t found in other frameworks with this level of ease.

Tower Middleware: Tracing and Logging Done Right

I quickly became a fan of Tower, which integrates effortlessly with Axum. Setting up logging, request tracing, and metrics collection is as simple as adding layers to the router.

With tower_http::trace, I could capture detailed logs of requests and responses, all in a non-intrusive way:

use tower_http::trace::TraceLayer;
use tracing::Level;

let app = Router::new()
    .route("/api", get(api_handler))
    .layer(TraceLayer::new_for_http().with_level(Level::INFO));  
// Enable logging

In a few lines of code, I had a fully instrumented API that logged every interaction. It made debugging and performance optimization much easier, and it felt like Axum was purpose-built for this kind of composability.

Rate Limiting and Protecting APIs

Rate limiting is another essential feature for any API-driven project. With tower’s middleware, I could easily cap the number of concurrent requests using the ConcurrencyLimitLayer. Again, the integration with Axum was smooth:

use tower::limit::ConcurrencyLimitLayer;

let app = Router::new()
    .route("/api/limited", get(limited_handler))
    .layer(ConcurrencyLimitLayer::new(100));  // Limit to 100

This setup ensures that my server remains responsive under load, which is crucial when handling high-traffic APIs.

Why Axum Outshines Raw Tokio and Actix

During my journey, I spent some time exploring both raw Tokio and Actix as alternatives. While both are powerful in their own right, I found Axum to be a step ahead in terms of developer experience and intuitiveness.

Axum vs Raw Tokio: Raw Tokio is fantastic when you need granular control over asynchronous operations, and its event-driven model is second to none. But when it comes to building web APIs, managing everything at the Tokio level can quickly become complex and verbose. Axum, built on top of Tokio, abstracts much of the boilerplate while maintaining full async capabilities. It’s like having the power of Tokio without the steep learning curve. You can focus on building your application rather than handling the low-level details. With Axum, you don’t need to manage HTTP requests manually or write tons of code to set up routing, middleware, or async logic. Axum does all of this in a much cleaner, modular way, making it the next-generation framework over raw Tokio for web applications.
Axum vs Actix: Actix Web was one of the first Rust web frameworks to gain serious traction, and it’s known for its high performance. However, I found Actix’s actor-based model to be more complex than necessary for most web use cases. While Actix shines in certain scenarios, the actor pattern can be an overkill for basic API services. Axum, in contrast, uses a more familiar and intuitive routing model. Its design follows the simplicity of frameworks like Express in JavaScript or Flask in Python, but with all the benefits of Rust’s type safety and performance. And yet, it doesn’t compromise on advanced features like middleware stacking, error handling, or async support. Plus, Axum’s reliance on Tower middleware gives you access to a wide range of ready-made tools (like CORS handling, rate limiting, and tracing) that are easy to plug into your API.
Tokio and Tower at its Core: Axum is also deeply integrated with Tokio and Tower, which are two of the most battle-tested and high-performance libraries in the Rust ecosystem. This means that Axum is not only fast but also scalable and production-ready. Whether it’s handling high concurrency with Tokio’s async runtime or layering on functionality with Tower middleware, Axum is designed to scale without getting in your way.

Why I’m Loving Axum

Axum has made my Rust web development journey far smoother than I anticipated. The fact that it allows me to easily control CORS per route, integrate logging and tracing, and enforce API rate limits—without over-complicating things—proves it’s a framework designed with developers in mind.

Here’s what I love most:

Elegant design: Routing, middleware, and error handling all feel natural. You don’t need to fight the framework to build complex APIs.
Composable and modular: Tower middleware makes it easy to layer in functionality like logging, rate limiting, and CORS control without cluttering your application logic.
Power of async: Axum’s seamless integration with Tokio gives me full control over asynchronous operations without adding unnecessary complexity.

Anyone looking to build web APIs in Rust, Axum can be easily your first choice. It offers the right balance of simplicity, power, and flexibility, making it a worthy successor to both raw Tokio setups and older frameworks like Actix. Whether building a small personal project or scaling up an enterprise service, Axum has the tools and developer experience worth exploring.

Financial Calculations in Rust: Building a Reference Portfolio App from Firm Trades to Customer Earnings

amaendeepm — Wed, 06 Nov 2024 09:40:19 +0000

When precision matters, especially in finance, the smallest decimal places can have a big impact on the bottom line. In this post, I will show a portfolio app in Rust that starts with reconciling the firm’s income from multiple sources, then allocates earnings to customers based on their portfolios, and finally calculates payouts or reinvestments based on their preferences. All while ensuring that our sums add up, from the firm level down to each individual customer.

The Use Case

Our app models a portfolio that deals with income from multiple sources. The steps are:

Reconcile the firm’s income from various sources (trade returns, dividends, etc.) and validate that the firm has received the expected amounts.
Calculate the firm’s overall earnings after reconciliation.
Allocate earnings to each customer based on their share of the firm’s portfolio.**
Handle payouts or reinvestments for each customer, applying specific rounding rules.**
Validate that the sum of all customer payouts matches the reconciled firm earnings, ensuring accuracy across our calculation chain.

By using Rust’s Decimal type and PostgreSQL’s NUMERIC, it can maintain high precision and consistency for our data.

Step 1: Reconciling the Firm's Income from Multiple Sources

Before calculating the firm’s total earnings, first we need to reconcile income from different sources. Let’s assume the firm has several streams of income, such as:

Market Trades
Dividends
Bond Interest Payments

Each income source is expected to contribute a certain amount, and we want to validate that we’ve received what was anticipated.

Firm Income Struct and Reconciliation

To model this, let’s define a FirmIncome struct that records both expected and received income for each source. Then, we’ll sum these up and check if they match.

use rust_decimal::Decimal;
use std::collections::HashMap;
use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize, Clone, Debug)]
struct FirmIncome {
    source: String,
    expected: Decimal,
    received: Decimal,
}

fn reconcile_income(incomes: &Vec<FirmIncome>) -> Result<Decimal, String> {
    let total_expected: Decimal = incomes.iter().map(|i| i.expected).sum();
    let total_received: Decimal = incomes.iter().map(|i| i.received).sum();

    if total_expected == total_received {
        Ok(total_received)
    } else {
        Err(format!("Income mismatch! Expected: {}, Received: {}", total_expected, total_received))
    }
}

Example Reconciliation

Here’s how we’d create and reconcile incomes from various sources:

let incomes = vec![
    FirmIncome { source: "Market Trades".to_string(), expected: Decimal::new(200_000, 2), received: Decimal::new(200_000, 2) },
    FirmIncome { source: "Dividends".to_string(), expected: Decimal::new(50_000, 2), received: Decimal::new(50_000, 2) },
    FirmIncome { source: "Bond Interest".to_string(), expected: Decimal::new(20_000, 2), received: Decimal::new(20_000, 2) },
];

let reconciled_income = reconcile_income(&incomes).expect("Income reconciliation failed");
println!("Reconciled Income: {}", reconciled_income); // Reconciled Income: 270,000.00

Now we know the total amount received (270,000.00) and can proceed with further calculations, confident that our firm’s income is accurately accounted for.

Step 2: Calculating the Firm’s Total Earnings Post-Reconciliation

With the reconciled income, we can calculate the firm’s total earnings using a market rate (or another performance metric). The firm’s earnings are calculated based on its market performance multiplied by the total assets.


#[derive(Serialize, Deserialize, Clone, Debug)]
struct Firm {
    total_assets: Decimal,
    market_rate: Decimal,
    reconciled_income: Decimal,
}

fn calculate_firm_earnings(firm: &Firm) -> Decimal {
    firm.reconciled_income + (firm.total_assets * firm.market_rate)
}

Example Usage

Assume the firm has 10,000,000.00 in total assets with a market rate of 2.5%:

let firm = Firm {
    total_assets: Decimal::new(10_000_000, 2),
    market_rate: Decimal::new(25, 3),
    reconciled_income,
};

let firm_earnings = calculate_firm_earnings(&firm);
println!("Firm Earnings after reconciliation: {}", firm_earnings); // Firm Earnings after reconciliation: 520,000.00

This calculation gives us the final amount for distribution across the customer portfolios.

Step 3: Allocating Earnings to Customer Portfolios

With the firm’s earnings calculated, let’s allocate these earnings to each customer based on their portfolio balance.

Customer Struct and Allocation Logic

#[derive(Serialize, Deserialize, Clone, Debug)]
struct Customer {
    id: i32,
    name: String,
    balance: Decimal,
    earnings: Decimal,
    reinvest: bool,
}

fn allocate_customer_earnings(firm_earnings: Decimal, customer_balance: Decimal, total_assets: Decimal) -> Decimal {
    firm_earnings * (customer_balance / total_assets)
}

Example Allocation

If a customer has 100,000.00 in their balance, the allocated earnings are calculated as follows:

let customer_balance = Decimal::new(100_000, 2);
let total_assets = firm.total_assets;

let customer_earnings = allocate_customer_earnings(firm_earnings, customer_balance, total_assets);
println!("Customer Earnings: {}", customer_earnings); // Customer Earnings: 5,200.00

Step 4: Processing Payouts and Reinvestments

Depending on each customer’s preference, we either round the earnings to two decimal places for a payout or add them to their balance for reinvestment.

Handling Payouts

fn payout(customer: &mut Customer) -> Decimal {
    let payout_amount = customer.earnings.round_dp(2);
    customer.balance -= payout_amount;
    customer.earnings = Decimal::ZERO;
    payout_amount
}

Handling Reinvestment

fn reinvest_earnings(customer: &mut Customer) {
    customer.balance += customer.earnings;
    customer.earnings = Decimal::ZERO;
}

Example Usage

let mut customer = Customer {
    id: 1,
    name: "Alice".to_string(),
    balance: Decimal::new(100_000, 2),
    earnings: customer_earnings,
    reinvest: false,
};

if customer.reinvest {
    reinvest_earnings(&mut customer);
} else {
    let payout_amount = payout(&mut customer);
    println!("Payout Amount: {}", payout_amount); // Payout Amount: 5,200.00
}

println!("Updated Balance: {}", customer.balance);

Step 5: Ensuring Bottom-Up Validation of Total Payouts

For consistency, we need to check that the total payouts match the firm’s reconciled earnings.

Validation Function

fn validate_total_payouts(customers: &Vec<Customer>, firm_earnings: Decimal) -> bool {
    let total_payouts: Decimal = customers.iter().map(|c| c.earnings.round_dp(2)).sum();
    total_payouts == firm_earnings.round_dp(2)
}

This validation adds an integrity check, ensuring that the sum of individual payouts aligns with the firm’s overall earnings, preventing discrepancies.

Precision Preservation with PostgreSQL’s NUMERIC

Using Decimal alongside PostgreSQL’s NUMERIC type keeps our data precise, ensuring our calculations maintain integrity across application and database layers.

PostgreSQL Schema Example

CREATE TABLE firm (
    id SERIAL PRIMARY KEY,
    total_assets NUMERIC(20, 2),
    market_rate NUMERIC(20, 4),
    reconciled_income NUMERIC(20, 2)
);

CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    balance NUMERIC(20, 2),
    earnings NUMERIC(20, 2),
    reinvest BOOLEAN
);

Final Thoughts

This Rust reference application demonstrates a robust, precision-oriented approach to managing financial calculations. By starting with income reconciliation and layering on accurate payout allocation and validation, we are required to maintain precise accounting of every decimal, aligning the payouts with the firm’s total earnings.

Rust’s Decimal and PostgreSQL’s NUMERIC together make a winning combination, ideal for any business application where precision is not just preferred but mandatory.

Kafka with Rust using rdkafka

amaendeepm — Mon, 21 Oct 2024 14:23:54 +0000

This Rust binding for the librdkafka C library allows you to harness Kafka's powerful messaging capabilities while benefiting from Rust's safety and performance.

First, you'll want to add the rdkafka dependency to your Cargo.toml file:

[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build", "naive-runtime", "tracing", "tokio","zstd"] }
tokio = { version = "1", features = ["full"] }
anyhow = "1.0"

You’ll also need to have librdkafka installed on your system. If you're on macOS, you can easily install it using Homebrew:

brew install librdkafka

For Ubuntu or other Debian-based systems, use:

sudo apt-get install librdkafka-dev

*Setting Up the Kafka Client
*
Here’s a sample implementation:

use rdkafka::{
    message::OwnedHeaders,
    producer::{BaseRecord, Producer, ProducerContext, ThreadedProducer},
    ClientConfig, ClientContext, Message,
};
use std::{env, time::Duration};

/// Kafka Client
pub struct Client {}

impl Client {
    /// Publish on the Kafka client
    pub async fn publish(
        headers: OwnedHeaders,
        message: &str,
        key: &str,
        topic: &str,
    ) -> anyhow::Result<()> {
        tracing::debug!(event="publish-start", key=?key, topic=?topic);

        let brokers = env::var("KAFKA_BROKERS").unwrap_or_else(|_| "localhost:9092".into());

        let producer: ThreadedProducer<ProducerCallback> = ClientConfig::new()
            .set("bootstrap.servers", brokers)
            .set("acks", "all")
            .set("enable.idempotence", "true")
            .set("compression.type", "zstd")  // Enable zstd compression
            .set("queue.buffering.max.ms", "0")
            .set("retries", "10")
            .set("request.timeout.ms", "10000")
            .create_with_context(ProducerCallback {})
            .expect("unable to create kafka producer");

        // Send the message
        let _ = producer.send(
            BaseRecord::to(topic)
                .payload(message)
                .key(key)
                .headers(headers),
        );

        // Flush the producer to ensure the message is sent
        let _ = producer.flush(Duration::from_secs(5));

        Ok(())
    }
}

Implementing Retry Logic

Network issues or broker unavailability can cause message publishing to fail. To handle this, we can add a function for publishing messages with an exponential backoff strategy:

/// Publish on Kafka with an exponential backoff
pub(crate) async fn publish_with_retry(
    headers: OwnedHeaders,
    message: &str,
    key: &str,
    topic: &str,
    interval: u64,
) -> anyhow::Result<()> {
    const MAX_ATTEMPTS: u64 = 10;
    let mut publish_attempts = 1;

    while let Err(err) = Client::publish(headers.clone(), message, key, topic).await {
        tracing::error!(event="publish", msg="failed", err=?err);

        if publish_attempts > MAX_ATTEMPTS {
            tracing::error!(event="publish", msg="exceeded retries", err=?err);
            return Err(KafkaError::RetryExceeded.into());
        } else {
            publish_attempts += 1;
            tracing::error!(event="publish", msg="retrying", err=?err, retry_count=publish_attempts);

            // Note: exponential backoff happens here
            tokio::time::sleep(Duration::from_secs(interval + (2 * publish_attempts))).await;
        }
    }

    Ok(())
}

This function retries sending the message, waiting longer between attempts, and gives up after a set number of retries.

Error Handling
For better error management during publishing, defining a custom error type can be useful:

/// Kafka Error
#[derive(Debug, thiserror::Error)]
pub enum KafkaError {
    #[error("Retry exceeded on trying to publish message")]
    RetryExceeded,
}

Producer Callback for Delivery Reports
To handle delivery reports for messages sent, you can implement a ProducerCallback struct. This allows you to manage responses to successful deliveries and failures:

/// Producer Callback for Kafka
struct ProducerCallback {}

impl ClientContext for ProducerCallback {}

impl ProducerContext for ProducerCallback {
    type DeliveryOpaque = ();

    /// Handle delivery report from the producer
    fn delivery(
        &self,
        delivery_result: &rdkafka::producer::DeliveryResult<'_>,
        _delivery_opaque: Self::DeliveryOpaque,
    ) {
        let dr = delivery_result.as_ref();

        match dr {
            Ok(msg) => {
                let key: &str = msg.key_view().expect("unable to get key view").unwrap();

                tracing::debug!(
                    event = "publish",
                    key = key,
                    offset = msg.offset(),
                    partition = msg.partition()
                );
            }
            Err(err) => {
                tracing::warn!(event="publish", state="fail",
                    content=?err);
            }
        };
    }
}

This callback provides insights into the delivery status of your messages, allowing you to react appropriately to successes and failures.

With this setup, you can effectively publish messages to Kafka using Rust’s rdkafka library while utilizing zstd compression.

API Development in Rust: CORS, Tower Middleware, and the Power of Axum

amaendeepm — Mon, 21 Oct 2024 13:40:54 +0000

CORS Control: Fine-Tuning Access

use axum::{Router, routing::get};
use tower_http::cors::{CorsLayer, Any};
use tower::ServiceBuilder;

let cors_layer = CorsLayer::new()
    .allow_origin(Any)  // Open access to selected route
    .allow_methods(vec!["GET", "POST"]);

let app = Router::new()
    .route("/public", get(public_handler))
    .route("/internal", get(internal_handler))
    .layer(ServiceBuilder::new().layer(cors_layer));

Axum allowed me to keep my security intact without complicating the setup, which I hadn’t found in other frameworks with this level of ease.

Tower Middleware: Tracing and Logging Done Right

I quickly became a fan of Tower, which integrates effortlessly with Axum. Setting up logging, request tracing, and metrics collection is as simple as adding layers to the router.

With tower_http::trace, I could capture detailed logs of requests and responses, all in a non-intrusive way:

use tower_http::trace::TraceLayer;
use tracing::Level;

let app = Router::new()
    .route("/api", get(api_handler))
    .layer(TraceLayer::new_for_http().with_level(Level::INFO));  
// Enable logging

Rate Limiting and Protecting APIs

use tower::limit::ConcurrencyLimitLayer;

let app = Router::new()
    .route("/api/limited", get(limited_handler))
    .layer(ConcurrencyLimitLayer::new(100));  // Limit to 100

This setup ensures that my server remains responsive under load, which is crucial when handling high-traffic APIs.

Why Axum Outshines Raw Tokio and Actix

Axum vs Raw Tokio: Raw Tokio is fantastic when you need granular control over asynchronous operations, and its event-driven model is second to none. But when it comes to building web APIs, managing everything at the Tokio level can quickly become complex and verbose. Axum, built on top of Tokio, abstracts much of the boilerplate while maintaining full async capabilities. It’s like having the power of Tokio without the steep learning curve. You can focus on building your application rather than handling the low-level details. With Axum, you don’t need to manage HTTP requests manually or write tons of code to set up routing, middleware, or async logic. Axum does all of this in a much cleaner, modular way, making it the next-generation framework over raw Tokio for web applications.
Axum vs Actix: Actix Web was one of the first Rust web frameworks to gain serious traction, and it’s known for its high performance. However, I found Actix’s actor-based model to be more complex than necessary for most web use cases. While Actix shines in certain scenarios, the actor pattern can be an overkill for basic API services. Axum, in contrast, uses a more familiar and intuitive routing model. Its design follows the simplicity of frameworks like Express in JavaScript or Flask in Python, but with all the benefits of Rust’s type safety and performance. And yet, it doesn’t compromise on advanced features like middleware stacking, error handling, or async support. Plus, Axum’s reliance on Tower middleware gives you access to a wide range of ready-made tools (like CORS handling, rate limiting, and tracing) that are easy to plug into your API.
Tokio and Tower at its Core: Axum is also deeply integrated with Tokio and Tower, which are two of the most battle-tested and high-performance libraries in the Rust ecosystem. This means that Axum is not only fast but also scalable and production-ready. Whether it’s handling high concurrency with Tokio’s async runtime or layering on functionality with Tower middleware, Axum is designed to scale without getting in your way.

Why I’m Loving Axum

Here’s what I love most:

Building a Robust Peek-Dequeue Engine in Rust for a General Message Portal

amaendeepm — Mon, 14 Oct 2024 13:44:57 +0000

In modern distributed systems, reliable message handling is crucial. Rust, with its emphasis on safety and concurrency, is an excellent choice for building robust systems. Here is an example of Peek-Dequeue engine implemented in Rust that interacts with a general message portal. Further there is explaining each component to help understand.

Introduction
Message portals are essential components in distributed architectures, acting as intermediaries for message exchange between different services. Building a reliable engine to interact with such portals requires careful handling of network communication, error management, and asynchronous operations.

This articulation of code below presents a Rust implementation of a Peek-Dequeue engine that utilizes some of best possible practices in asynchronous programming, error handling, and most importantly system resilience.

Overview of the Peek-Dequeue Engine
The engine's primary responsibilities are:

Peeking Messages: Regularly checking the message portal for new messages.
Handling Responses: Parsing and validating the responses, including error handling.
Processing Messages: Transmuting and persisting valid messages.
Dequeueing Messages: Removing processed messages from the portal.

Here's the high-level structure of the MessagePortalConnection:

pub struct MessagePortalConnection {
    interval: u64,
}

impl Default for MessagePortalConnection {
    fn default() -> Self {
        Self { interval: 1 }
    }
}

impl MessagePortalConnection {
    pub async fn run(&self) {
        // Implementation details...
    }
}

Setting Up the Connection

The MessagePortalConnection struct holds the configuration for the connection, such as the peeking interval.

pub struct MessagePortalConnection {
    interval: u64,
}

impl Default for MessagePortalConnection {
    fn default() -> Self {
        Self { interval: 60 } // Default peeking every minute
    }
}

Using the Default trait allows for easy instantiation with default settings while providing flexibility for customization.

The Main Run Loop
The run method contains the main loop that continuously peeks for new messages.

impl MessagePortalConnection {
    pub async fn run(&self) {
        let portal_endpoint: Arc<str> = env::var("PORTAL_URL")
            .unwrap_or_else(|_| "https://portal.example.com/".to_owned())
            .into();

        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(self.interval));

        loop {
            tokio::select! {
                _ = interval.tick() => {
                    // Peeking logic...
                }
            }
        }
    }
}

Interval Setup: Uses tokio::time::interval for periodic execution without blocking the runtime.

Asynchronous Loop: The tokio::select! macro allows the loop to wait on multiple asynchronous operations.

Peeking Messages:
Within the loop, the engine attempts to peek messages from the portal.

let peek_trace_id = Ulid::from_datetime(SystemTime::now()).to_string(); // Unique ID for tracing

tracing::debug!(event="peek-portal", trace=peek_trace_id);

let peek_result = retry_with_backoff(
    || async {
        peek_from_portal(portal_endpoint.clone().into()).await.map(|arc_str| arc_str.to_string())
    },
    5,  // Max retries
    2,  // Initial delay (seconds)
    1.5 // Backoff factor
).await;

Unique Trace ID: Generates a ULID based on the current time for traceability.

Retry Logic: Utilizes a custom retry_with_backoff function to handle transient failures.

Handling Responses
After attempting to peek, the engine processes the response.

if peek_result.is_err() {
    tracing::error!(event="portal_peek_result_error", trace=peek_trace_id, err=?peek_result.err());
    continue;
}


let peek_response = peek_result.unwrap();
let doc_result = Document::parse(&peek_response);

if doc_result.is_err() {
    tracing::error!(event="peeked-invalid-response", trace=peek_trace_id, err=?doc_result.err(), msg="peek-response-error");
    continue;
}

Error Handling: Checks for errors at each step and logs them appropriately.
XML Parsing: Uses roxmltree::Document::parse to parse the XML response.
Processing Valid Messages
Once a valid XML response is confirmed, the engine processes the message.

let root_node = doc_result.unwrap();
let fault_string = root_node
    .descendants()
    .find(|node| node.has_tag_name("faultstring"))
    .map(|node| node.text())
    .unwrap_or_default();

if fault_string.is_some() {
    tracing::error!(event="peeked-fault-string", msg="peek-SOAP-fault", trace=peek_trace_id, response=?peek_response);
    continue;
}

// Determine the document type
let doc_type_opt = find_document_type_from_peek_response((&peek_response.clone()).to_string());

if doc_type_opt.is_none() {
    tracing::warn!(event="peek", msg="nothing-to-peek", trace=peek_trace_id);
    continue;
}

let doc_type = doc_type_opt.unwrap();
tracing::info!(event="peek", msg="returned document", doc_type=?doc_type, trace=peek_trace_id);

SOAP Fault Handling: Checks for fault messages and handles them gracefully.
Document Type Identification: Determines the type of the received document for appropriate processing.

Dequeueing Messages
After processing, the engine attempts to dequeue the message from the portal.

let msg_ref = extract_message_reference(peek_response.clone().into());

if msg_ref.is_none() {
    tracing::error!(event="peek", msg="xml has no message reference included", trace=peek_trace_id);
    continue;
}

let msg_ref_id = msg_ref.unwrap();
tracing::info!(event="peek", msg="extracted message reference", reference_id=?msg_ref_id, trace=peek_trace_id);

let incoming_raw_xml = peek_response.to_string().clone();

let incoming_payload = extract_payload_json(incoming_raw_xml.clone().into());

let handling_status = handle_incoming(
    peek_trace_id.clone().into(),
    incoming_payload.clone().unwrap(),
    incoming_raw_xml.clone().into(),
).await;

if !handling_status {
    tracing::error!(event="peek", msg="response not handled internally; not dequeue or proceed", trace=peek_trace_id);
    continue;
}

if incoming_payload.is_some() {
    let payload = incoming_payload.unwrap();
    let transmuted_message = transmute_response(incoming_raw_xml.clone().into()).unwrap();
    let persist_status = persist_identifiable_message(peek_trace_id.clone().into(), payload, transmuted_message.clone()).await;
    if persist_status.is_err() {
        tracing::error!(event="Persist_Transmuted_Message", msg="Could not persist Transmuted JSON Message", error=?persist_status.err(), trace=peek_trace_id);
    } else {
        tracing::debug!(event="Persist_Transmuted_Message", msg="Successfully Persisted Transmuted JSON Message", trace=peek_trace_id);
    }
}

let msg_ref_str = msg_ref_id.to_string();
tracing::info!(event="dequeue", msg="trying", msg_id=msg_ref_str, trace=peek_trace_id);

loop {
    let dequeue_result = retry_with_backoff(
        || async {
            request_message_dequeue_portal(msg_ref_str.clone().into())
                .await
                .map(|arc_str| arc_str.to_string())
        },
        10, // Max retries
        2,  // Initial delay (seconds)
        1.5 // Backoff factor
    ).await;

    if dequeue_result.is_err() {
        tracing::error!(event="portal_dequeue", msg="failed; retrying", trace=peek_trace_id);
        notify_error(format!("Dequeue Attempts exceeded retries for Message {msg_ref_id} :: Peek is Stuck, PeekTraceID {peek_trace_id}").into());
        break;
    } else {
        tracing::info!(event="dequeue", msg="ok", msg_id=msg_ref_str, trace=peek_trace_id);
        break;
    }
}

Message Reference Extraction:
Retrieves a unique identifier for the message.

Handling Status Check: Ensures the message was handled correctly before proceeding.

Persistence: Attempts to persist the processed message, logging any errors.

Dequeue Loop: Uses a loop with retry logic to ensure the message is dequeued, preventing duplicate processing.

Retry Logic with Exponential Backoff

The retry_with_backoff function implements retry logic with exponential backoff, which is crucial for handling transient errors without overwhelming the system or the portal.

async fn retry_with_backoff<F, Fut, T, E>(operation: F, max_retries: u32, delay_secs: u64, backoff_factor: f64) -> Result<T, E>
where
    F: Fn() -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
{
    let mut retries = 0;
    let mut delay = Duration::from_secs(delay_secs);

    loop {
        let result = operation().await;
        if result.is_ok() || retries >= max_retries {
            return result;
        }
        tokio::time::sleep(delay).await;
        retries += 1;
        delay = Duration::from_secs_f64(delay.as_secs_f64() * backoff_factor);
    }
}

Generic Function: Works with any asynchronous operation that returns a Result.

Exponential Backoff: Increases the delay between retries exponentially, reducing load during failures.

Summarizing

This Peek-Dequeue engine uses Rust's capabilities in building reliable, high-performance system.
The code follows basically:

Asynchronous Efficiency: Leveraging Tokio for non-blocking operations.

Robust Error Handling: Comprehensive checks and logging at each stage.

Resilience: Retry mechanisms with exponential backoff to handle transient failures.

Maintainability: Modular design with clear separation of concerns.

Rust Errors: anyhow Revelation

amaendeepm — Mon, 07 Oct 2024 09:16:30 +0000

How I stopped writing Rust like it was Golang and finally embraced anyhow::Error.

When I first started with Rust, my error handling looked something like this:

fn do_stuff() -> Result<String, String> {
    let step1 = step1()?;
    if step1.is_error() {
        return Err("Step 1 failed".to_string());
    }

    let step2 = step2(step1.unwrap())?;
    if step2.is_error() {
        return Err("Step 2 failed".to_string());
    }

    Ok("All good!".to_string())
}

Or maybe when I learned Rust further my return type was still Result<Arc<str>, Arc<str>> , but still following same style of error checking and handling!

Yikes, right? It looked like I was writing Go, but with extra steps. Having an error as a type provided thru anyhow::Error, and code conditions and sizing came way down smaller and so much more readable:

use anyhow::{Result, Context};

fn do_stuff() -> Result<String> {
    let step1 = step1().context("Step 1 failed")?;
    let step2 = step2(step1).context("Step 2 failed")?;
    Ok("All good!".to_string())
}

So much cleaner! No more explicit checks, and I can add context where it matters. The best part? I can still use Result<String, String> for those rare cases where I need super explicit errors.

fn critical_stuff() -> Result<String, String> {
    // Detailed custom errors here
}

Now, let's talk pros and cons. On the professional side, anyhow has some significant benefits:

It reduces boilerplate, making code more readable and maintainable.
The Context trait allows for rich error messages without custom error types.
It's great for rapid prototyping and applications where detailed error handling isn't critical.

However, it's not all roses. There are some potential drawbacks:

Loss of static type checking for errors. You can't exhaustively match on error types, which can be a problem in libraries or safety-critical code.
Performance overhead. While minimal, anyhow::Error does introduce some runtime cost compared to statically typed errors.
It can make it easier to be less thoughtful about error design. In complex systems, well-designed error types can be invaluable.

Overall, in my experience, the benefits outweigh the drawbacks, especially in application code. For libraries or performance-critical systems, you might want to stick with custom error types.

If you're still writing Rust errors like it's Go, give anyhow::Error a shot

Kafka Data Retention: Symphony of retention.ms & segment.ms

amaendeepm — Tue, 26 Dec 2023 22:10:09 +0000

While grasping to see very old data still on a Kafka topic recently led me to a profound understanding of the often underestimated collaboration between retention.ms and segment.ms. Far from a conventional time-to-live concept, these two configurations work in tandem, significantly impacting Kafka's data retention behaviour.

Going by the documentation around Kafka data orchestration, retention.ms serves as "the" configuration, meticulously shaping data retention. Yet, the true revelation lies in the intricate dance between retention.ms and its counterpart, segment.ms.

Working in tandem, these configurations play a vital role in shaping Kafka's data retention with precision. Basically, each topic partition divides its log into smaller, manageable chunks known as segments. The size of these segments, determining when the log rolls, is controlled by segment.bytes (sized based) and segment.ms (time-based). This dance becomes especially critical in scenarios where slow-paced topics might lead to extended waits for segment.bytes to accumulate sufficient data.

The narrative extends beyond a single Kafka topic for those venturing into tiered storage. The orchestrated dance expands into the cloud, where the influence of retention.bytes and retention.ms orchestrates the symphony of data management. Simultaneously, their local counterpart configrations viz. retention.local.target.bytes and retention.local.target.ms thus take center stage, managing the tempo on the local storage platform.

Example:

Lets maybe take a festive example featuring the topic "ChristmasGreetings." This topic is a lively hub of holiday messages from various channels, partitioned for efficiency, and set to unfold the duality of retention.ms and segment.ms:

Number of Partitions & Segment Rollover Config:
The "ChristmasGreetings" topic boasts 4 partitions, each diligently managing its own log segments.

The size of these segments, governed by segment.bytes, is set to 1 MB, ensuring manageable chunks of holiday cheer.

segment.ms is choreographed to 15 minutes, allowing segments to gracefully age and rollover.

Entering retention.ms - Longer segment.ms than retention.ms:

In this scenario, let's consider a longer segment.ms of 30 minutes, while retention.ms remains at 12 hours.

Log segments gracefully rollover every 30 minutes due to segment.ms, but messages are eligible for deletion only after 12 hours, as dictated by retention.ms.

This can lead to some segments persisting longer than needed, potentially accumulating more holiday messages than necessary. The temporal control enforced by segment.ms ensures a graceful rollover but doesn't necessarily align with the overall retention policy.

Shorter segment.ms than retention.ms:

Now, let's flip the scenario, setting segment.ms to 5 minutes and retaining messages for 12 hours.

Log segments rollover every 5 minutes due to segment.ms, but messages are eligible for deletion only after 12 hours according to retention.ms.

This configuration ensures more frequent rollovers, potentially creating a higher number of smaller log segments. While it aligns with the retention policy, it might introduce additional overhead in terms of segment management.

Balancing Act: Equal segment.ms and retention.ms:

For an optimal configuration, segment.ms is set at 15 minutes, closely aligned with retention.ms of 12 hours.

Log segments gracefully rollover in sync with the overall retention policy, ensuring efficient data management without unnecessary accumulation or overhead.

Conclusion:
In this holiday message capturing performance, segment.ms and retention.ms dance in different tempos, showcasing the nuances of Kafka's orchestration within the festive realm of "ChristmasGreetings." For anyone aiming to optimize and fortify their data streaming platforms, understanding the dynamic interplay between retention.ms and segment.ms becomes an essential therefore. Something to keep in mind always!