DEV Community: Rahul Reddy Talatala

I Got Tired of Filling Out the Same Form 50 Times, So I Built an AI to Do It

Rahul Reddy Talatala — Fri, 06 Mar 2026 06:44:27 +0000

Every time I applied for a job, I faced the same ritual. Open the application form. Type my full name. Type my email. Paste my LinkedIn URL. Type my phone number. Select my country from a dropdown. Answer "Are you authorized to work in the US?" for the fifteenth time that week.

The entire process takes about ten minutes per application, and roughly eight of those minutes are spent on fields I have answered hundreds of times before. The two minutes that actually matter, the cover letter, the portfolio link, the thoughtful answers to specific questions, get squeezed into whatever mental energy I have left.

I am a GenAI engineer. I spend my days building systems that make computers do repetitive cognitive work. The irony of manually typing my zip code into yet another Greenhouse form at midnight was not lost on me.

So I built ApplyAI: a Chrome extension that reads a job application form, sends the fields to an AI agent, gets back a fill plan, and applies it to the page in under ten seconds.

🚀 Try it on the Chrome Web Store | 🌐 Web App | 🔗 Github

🎥 Full product demo

This post walks through how it works, the actual architecture, the hard problems, and the design decisions that shaped the final system.

The Stack

Chrome Extension: Vanilla JS with a React popup (Vite + Tailwind)
Backend: FastAPI (Python)
AI Agent: LangGraph StateGraph with Gemini 2.5 Flash
Database and Auth: Supabase (PostgreSQL + Auth + Storage)
Frontend Dashboard: Next.js 14 with TypeScript

Three components. One clear job each. The extension is the hands. The backend is the brain. The frontend is the face.

The Architecture

Here is the data flow for a single autofill run:

User navigates to a job application form (Lever, Ashby, Greenhouse, or any careers page)
The Chrome extension extracts every form field from the live DOM, including dropdown options
The extension sends those fields plus the raw DOM HTML to the FastAPI backend
The backend runs a LangGraph agent that calls Gemini 2.5 Flash with the user's profile, resume, and the extracted job description
Gemini returns a structured JSON answer for every field
The backend assembles a fill plan and sends it back to the extension
The extension applies each value to the correct form element using CSS selectors

The key insight that made the whole thing feasible: the browser is the only environment that can see a fully rendered React form. A server-side scraper sees the HTML skeleton. A real browser running JavaScript sees the actual dropdown options, the dynamic field states, and the ARIA attributes that React Select generates at runtime. So the extraction has to happen inside the browser, and the AI reasoning has to happen on the server where I have access to the user's data.

Part 1: Getting Form Fields Out of a Live Page

This turned out to be the hardest non-AI problem in the whole project.

Job application forms are not plain HTML. They are React components. The dropdowns are typically React Select, which renders options into a floating portal that only exists in the DOM when the dropdown is actually open. If you scrape the page while all dropdowns are closed, you get combobox elements with no options. The AI has no idea what values are valid.

My solution: open every dropdown programmatically before scraping anything.

The extension injects a script into the active tab using chrome.scripting.executeScript. Before touching a single field, it finds every [role="combobox"] element and simulates a real user opening it by dispatching mouse and keyboard events in sequence, then waits 300ms for React to render the options into the DOM.

combobox.dispatchEvent(new MouseEvent('mousedown', { bubbles: true }));
combobox.dispatchEvent(new MouseEvent('click', { bubbles: true }));
combobox.dispatchEvent(new KeyboardEvent('keydown', { key: 'ArrowDown', bubbles: true }));

await sleep(300); // wait for React to paint the listbox

The 300ms wait is not arbitrary. I tested against Lever, Ashby, and Greenhouse. At 200ms the options were missing about 30% of the time. At 300ms the failure rate dropped to near zero.

Once the dropdown is open, options are extracted from the ARIA-controlled listbox. Each field is then serialized into a structured object containing its type, label, CSS selector, required status, and available options. That object is what gets sent to the backend.

The CSS selector is computed at extraction time using the element's id or name attribute. This same selector is used later during the apply step to locate the exact element on the page, so precision matters here.

Label detection uses multiple fallback strategies in order: the for attribute on an associated <label>, a parent <label> element, aria-label, and finally placeholder. When none of those exist, the field name or id becomes the label.

Part 2: The LangGraph Autofill Agent

Once the extension sends the extracted fields to the backend, a LangGraph StateGraph takes over. The autofill pipeline has four clearly separated concerns and a DAG maps to that structure naturally.

START -> initialize -> extract_form_fields -> generate_answers -> assemble_autofill_plan -> END

Each node receives the full shared state, does one job, returns its updates, and passes control to the next node.

Node 1: Initialize

Sets the run ID, page URL, and initial empty collections. Straightforward bookkeeping.

Node 2: Extract Form Fields

Converts the JavaScript field objects from the extension into typed Python FormField dictionaries. This handles type mapping (React Select comboboxes become select type), deduplication by field signature, and one non-obvious enrichment.

If a select field has "country", "nationality", or "citizenship" in its label and has zero extracted options, the backend automatically injects the full list of 196 standard country names. This is a safety net for the cases where the browser-side dropdown opening fails. Some Greenhouse forms use a custom country component that does not respond to standard mouse events. The backend catches this gap and fills in the options so the LLM still has something to work with.

Node 3: Generate Answers

This is where Gemini does the work.

I spent a lot of time on the prompt design. The format that worked best is structured JSON as the prompt body rather than prose. The task description, rules, context, and output format are all keys in a JSON object. This consistently outperformed plain English paragraphs for precision tasks.

prompt_obj = {
    "task": f"Generate answers for ALL {len(fields_spec)} form fields.",
    "critical_rules": [
        "MANDATORY: Set action='autofill' for ALL fields. Never use 'skip' or 'suggest'.",
        "If you don't know an answer, use action='autofill' with value='' and low confidence.",
    ],
    "context": {
        "user_ctx": user_ctx,    # profile fields
        "job_ctx": job_ctx,      # extracted job description
        "resume_ctx": resume_ctx # parsed resume data
    },
    "form_fields": fields_spec,
}

The model treats JSON keys as hard constraints, not suggestions. Prose prompts produced more hedging and more "I cannot fill this" responses. JSON prompts produced precise, consistent output.

Gemini is called with response_mime_type: "application/json" and a response_json_schema derived directly from a Pydantic model:

response = llm.client.models.generate_content(
    model="gemini-2.5-flash",
    contents=prompt,
    config={
        "response_mime_type": "application/json",
        "response_json_schema": LLMAnswersResponse.model_json_schema(),
    },
)

This eliminates an entire class of failure modes. No regex extraction. No JSON fence stripping. No trying to parse Gemini's explanation text alongside the output. The response is always valid JSON that maps directly to the Pydantic model.

The aggressive autofill rule

Early versions would skip EEO fields ("Race", "Veteran Status") and optional demographics because the LLM would flag them as sensitive. Users found this confusing: they had no idea why some fields were left blank.

The fix was a two-layer no-skip guarantee. The prompt says action='autofill' is mandatory. Then a _normalize_answer() post-processing step converts any skip or suggest action to autofill at plan assembly time, even if the LLM ignored the instruction. If the LLM has no idea what value to use, it returns an empty string with low confidence. The field gets pre-filled with nothing and the user can fix it manually. This is a much better experience than leaving fields blank silently.

Option matching

The LLM might return "United States" when the dropdown says "USA". Or it might get the case right but add extra whitespace. After receiving the LLM response, each answer for a select, radio, or checkbox field goes through a normalizer that strips both the LLM value and each available option down to lowercase alphanumeric characters, then tries an exact match first and falls back to substring containment. This handles most real-world mismatches without any hardcoded synonym maps.

File inputs are handled separately, outside the LLM entirely

File upload fields are detected by input_type == "file". The code checks the label for "cover letter" keywords. If matched, the field is skipped because no one should submit an AI-generated cover letter without reviewing it. Everything else gets value: "resume" with maximum confidence. No LLM call needed.

Node 4: Assemble Autofill Plan

Combines form_fields and answers into the final plan structure, generates a summary with field counts, and writes the completed plan to the database. The plan contains each field's selector, value, action, and confidence score.

Part 3: Applying the Plan Back in the Browser

The backend returns the plan. Now the extension has to actually fill the form.

Another injected script iterates through every field in the plan and uses a different fill strategy based on input type:

Text and textarea: Set value through React's native property setter to trigger synthetic events, then dispatch input and change events so the React component knows the value changed.
Native <select>: Find the matching option by text content. Set selectedIndex. Fire a change event.
React Select (combobox): Type the value into the input, wait for the listbox, find the matching option, click it.
Radio and checkbox groups: Find the label whose text matches the target value and click it.
File inputs: This one deserves its own explanation.

The browser's security model prevents setting input.value on a file input for obvious reasons. The only legitimate way to attach a file programmatically is through the DataTransfer API:

async function fillFileInput(el, fileUrl) {
    const response = await fetch(fileUrl);
    const arrayBuffer = await response.arrayBuffer();
    const file = new File([arrayBuffer], "resume.pdf", { type: "application/pdf" });

    const dataTransfer = new DataTransfer();
    dataTransfer.items.add(file);
    el.files = dataTransfer.files;
    el.dispatchEvent(new Event('change', { bubbles: true }));
}

The fileUrl is a Supabase Storage signed URL generated by the backend at plan time with a one-hour TTL. The extension fetches the user's resume as an ArrayBuffer, wraps it in a File object, and attaches it to the input through the DataTransfer API. This works reliably across Greenhouse, Lever, and Ashby.

Part 4: The Authentication Design

The system runs two completely separate auth systems side by side.

The web frontend uses Supabase's standard JWT (email/password and Google OAuth). The token lives in a cookie and is sent as a Bearer token on API calls.

The Chrome extension cannot participate in cookie-based sessions. It runs in a sandboxed context with no access to the frontend's cookies. So I built a separate auth flow using custom JWTs.

The connection process works like a one-time code exchange. When the user clicks "Connect" in the popup, the frontend generates a 32-character urlsafe random code, stores its SHA-256 hash in the database with a 10-minute expiry, and sends the plaintext code to the extension via window.postMessage(). The extension exchanges this code at POST /extension/connect/exchange for a 7-day JWT signed with a custom secret.

The JWT carries an audience claim set to applyai-extension. The backend validates this on every extension endpoint, so the extension token cannot be replayed against any other part of the API. It also carries an install_id (a UUID stored in chrome.storage.local) for device-level tracking without building a separate device registry.

Part 5: Three Hard Problems I Did Not Expect

1. Chrome extension popups throttle CSS animations

I built the popup with loading spinners and skeleton loaders. They worked fine during development in the browser. Once I loaded the extension and opened the actual popup, every animation was frozen solid.

Chrome throttles JavaScript execution in extension popup contexts to save battery. Tailwind's default animate-spin and animate-pulse classes get paused by this throttling.

The fix is to redefine the keyframes explicitly in CSS and force the play state:

@keyframes spin {
  to { transform: rotate(360deg); }
}

.animate-spin {
  animation: spin 0.85s linear infinite;
  animation-play-state: running !important;
}

The !important on animation-play-state overrides whatever Chrome tries to set. Every animation class in the popup needs this treatment.

2. DOM hashing breaks plan caching

The first version of plan caching computed a SHA-256 hash of the full DOM HTML and used it as part of the cache key. If the hash matched a previous run, return the cached plan.

This broke constantly. Every Greenhouse page load includes a fresh CSRF token embedded in the HTML. Every page view produces a different hash even for the exact same form. The cache was useless.

The fix: cache by job_application_id + normalized_page_url. The DOM hash is still stored in the database but ignored for cache lookups. The same form at the same URL always returns the same cached plan regardless of what changed in the page source between visits.

3. Lever and Ashby split one job across two URLs

On Lever, the job description lives at jobs.lever.co/company/slug. The application form lives at jobs.lever.co/company/slug/apply. These look like two different pages but they represent one job.

If a user extracts the job on the description page and then navigates to the apply page, the backend needs to match the apply URL back to the original job record. The extract_job_url_info() utility detects the /apply or /application suffix and strips it to get the canonical base URL before doing any database lookups.

On the extension side, when the popup detects the user is on a job description page with a saved job, it shows an amber banner telling them to navigate to the application form. For Lever and Ashby, the banner includes a direct link constructed entirely client-side by appending /apply or /application to the current URL. No extra backend call needed.

Part 6: What Is Coming Next

Streaming the autofill plan

Right now the user sees a spinner for 5 to 10 seconds while the full plan generates. Gemini supports streaming and LangGraph supports streaming node outputs. The plan is to emit node completion events to the extension popup as each stage finishes, so users see live progress instead of one long blocking wait.

A retry node in the LangGraph DAG

On very long forms, Gemini occasionally returns fewer answers than there are fields. The current system fills missing fields with empty-string defaults. A better approach is a conditional retry edge in the graph: if the answer count is less than the field count, route back to generate_answers with only the missing field signatures. LangGraph's conditional edges make this a clean addition.

Semantic resume matching

The current resume match feature does a keyword overlap check between the job's required skills and the user's resume skills list. This misses "REST APIs" matching "API development" or "React.js" matching "React". The plan is to replace this with embedding-based similarity using a small vector store per user, which gives a much more accurate match score and surfaces genuinely missing skills rather than just unmatched strings.

Playwright-based extraction fallback

The browser-side dropdown opening works on about 90% of forms. The remaining 10% use heavy custom focus trapping, animated dropdowns with delays over 300ms, or React Select versions that ignore standard mouse events. A Playwright-powered extraction service on the backend would handle these edge cases by running a full headless browser with complete control over the page lifecycle, without requiring any changes to the extension.

Closing Thoughts

The most interesting thing I learned building this is that the Chrome extension is not just a UI layer. It is the only component in the system with access to the live, JavaScript-executed version of the page. That makes it the source of truth for form structure. The AI cannot do its job without what the browser extracts first.

The second thing: structured JSON prompts beat prose prompts for precision tasks. When I need Gemini to return exactly N answers with specific fields and constrained action values, a JSON prompt with rules expressed as an array performs better than a paragraph of instructions. The model treats it like a spec, not a suggestion.

The combination of LangGraph for agent orchestration, Gemini's native JSON schema output mode, and the DataTransfer API for file uploads turned out to be a surprisingly complete toolkit for this problem.

If you are building something similar or have questions about any part of the architecture, drop a comment below.

Built with ❤️ by Rahul

LinkedIn | GitHub | Portfolio

Spring Core Fundamentals: A Beginner Guide

Rahul Reddy Talatala — Wed, 12 Mar 2025 05:48:22 +0000

Spring is one of the most popular frameworks in the Java ecosystem, known for its powerful dependency injection capabilities and extensive ecosystem. Whether you're just starting with Spring or looking to refresh your knowledge, this guide will walk you through the core concepts that make Spring such a powerful framework.

Understanding Spring vs Spring Boot

Before diving into the core concepts, let's clarify the difference between Spring and Spring Boot:

Feature	Spring Framework	Spring Boot
Definition	A comprehensive framework providing infrastructure support for Java applications	An extension of Spring that simplifies development with pre-configured defaults
Configuration	Requires extensive manual configuration	Provides auto-configuration, reducing manual setup
Standalone Apps	Requires external servers	Comes with embedded servers for standalone applications
Boilerplate Code	Requires more configuration code	Reduces boilerplate with opinionated defaults
Dependency Management	Manual dependency management	Uses Spring Boot Starter dependencies

At its core, Spring framework is primarily a dependency injection container with additional convenience layers for database access, proxies, aspect-oriented programming, and web MVC.

Dependency Injection: The Heart of Spring

Dependency Injection (DI) is a design pattern where a class receives its dependencies from an external source rather than creating them itself. This promotes:

Loose coupling between components
Improved testability
Better code maintainability

Spring's core functionality is implementing this pattern effectively through its IoC container.

Inversion of Control: Shifting Responsibility

Inversion of Control (IoC) is a design principle where the control of object creation and lifecycle management is transferred from your application code to the Spring framework.

Traditional Approach vs IoC

Without IoC:

class Car {
    private Engine engine;
    public Car() {
        this.engine = new Engine(); // Object creation inside the class (tight coupling)
    }
}

With IoC (Spring's approach):

@Component
class Car {
    private final Engine engine;
    @Autowired
    public Car(Engine engine) {
        this.engine = engine; // Spring injects the dependency
    }
}

The key difference? With Spring's IoC, your application code doesn't control dependency creation - Spring does.

Types of Dependency Injection in Spring

Spring supports three primary types of dependency injection:

1. Constructor-Based Injection

@Component
class Car {
    private final Engine engine; // immutable
    @Autowired 
    public Car(Engine engine) {
        this.engine = engine;
    }
}

Advantages:

Dependencies are explicit
Encourages immutability
Objects are always created with required dependencies

This is generally the recommended approach in modern Spring applications.

2. Setter-Based Injection

@Component
class Car {
    private Engine engine;

    @Autowired
    public void setEngine(Engine engine) {
        this.engine = engine; // optional dependency
    }
}

Advantages:

Allows optional dependencies
Provides flexibility in modifying dependencies at runtime

Disadvantages:

Objects might be in an incomplete state if dependencies aren't provided
More verbose than constructor injection

3. Field-Based Injection

@Component
class Car {
    @Autowired
    private Engine engine;
}

Advantages:

Less boilerplate code
Simple to implement

Disadvantages:

Harder to test
Cannot enforce immutability
Dependencies aren't explicitly declared

Spring IoC Container: Managing Your Beans

Spring provides two types of IoC containers to manage the lifecycle of beans:

BeanFactory: The Lightweight Container

Provides basic bean management capabilities
Uses lazy initialization - beans are created only when requested
Uses less memory and is suitable for resource-constrained environments

ApplicationContext: The Feature-Rich Container

Extends BeanFactory with enterprise features
Uses eager initialization - beans are created at startup
Provides advanced features like:
- Annotation-based dependency injection
- Event handling
- Internationalization support
- AOP integration

Spring Boot automatically uses ApplicationContext for its enhanced feature set.

Understanding Spring Beans

A bean in Spring is simply an object created and managed by the Spring IoC container. Spring creates instances of classes annotated with @Component (or its specialized annotations), manages their lifecycle, and injects them where needed.

Bean Scopes: Controlling Instance Creation

Spring offers several bean scopes that determine how instances are created and shared:

Scope	Description
singleton (default)	A single shared instance of the bean is created for the entire application
prototype	A new instance is created each time the bean is requested
request	A new instance is created for each HTTP request (web applications only)
session	A new instance is created for each user session (web applications only)
application	A single instance for the entire ServletContext (web applications only)

You can specify the scope using the @Scope annotation:

@Component
@Scope("prototype")
class PrototypeBean {
    // Each request for this bean will create a new instance
}

Essential Spring Annotations

Spring provides a rich set of annotations to simplify configuration:

Component Annotations

@Component: Generic annotation for any Spring-managed component
@Service: For service layer classes containing business logic
@Repository: For data access objects handling database operations
@Controller: For Spring MVC controllers handling web requests
@RestController: For RESTful controllers (combines @Controller and @ResponseBody)

Configuration Annotations

@Configuration: Marks a class as a source of bean definitions
@bean: Explicitly declares a bean method within a @Configuration class
@ComponentScan: Tells Spring where to look for annotated components

Dependency Injection Annotations

@Autowired: Marks a constructor, field, or setter method for automatic dependency injection

@Configuration
@ComponentScan("com.example")
class AppConfig {
    @Bean
    public Engine customEngine() {
        return new TurboEngine();
    }
}

Aspect-Oriented Programming with Spring

Spring AOP allows you to separate cross-cutting concerns (like logging, security, transactions) from your business logic. It does this by using proxies to intercept method calls and apply additional behavior.

@Aspect
@Component
public class LoggingAspect {
    @Before("execution(* com.example.Service.*(..))")
    public void logBefore() { 
        System.out.println("Method execution started..."); 
    }
}

Key AOP concepts include:

Aspect: Module containing cross-cutting logic
Join Point: Execution point where aspect can be applied
Advice: Action performed at a join point (before, after, around)
Pointcut: Expression that determines where advice should be applied
Weaving: Process of applying aspects to target objects

A real-world example is Spring's @Transactional annotation, which uses AOP to handle transaction management without cluttering your business logic.

Conclusion

Spring Core provides a robust foundation for building Java applications by implementing dependency injection and inversion of control. By managing dependencies externally, Spring helps create more modular, testable, and maintainable applications.

Understanding these fundamental concepts - IoC, DI, beans, annotations, and AOP - will give you a solid foundation for working with the Spring ecosystem, whether you're using Spring Framework directly or Spring Boot.

The true power of Spring comes from how these concepts work together, allowing you to focus on your business logic while Spring handles the infrastructure concerns.

Curious about building a rock-solid e-commerce system with Spring Boot microservices? Dive into my latest blog where I break it all down—resilience, scalability, and the magic of Spring! Check it out here.

Happy Springing!🚀

Built with ❤️ by Rahul
LinkedIn | GitHub | Portfolio

Spring Commerce: Building a Resilient E-commerce System with Spring Boot Microservices

Rahul Reddy Talatala — Wed, 12 Mar 2025 05:32:14 +0000

Hola 👋,

I recently completed a comprehensive microservices project using Spring Boot 3, implementing various modern patterns and technologies. Here's an overview of what I built and the key learnings from this journey.

The project code is available here on GitHub

🌱 Before diving deep into the project, make sure to lay a solid foundation! Check out my Spring Boot basics blog first—trust me, it'll make everything click! 🔗 Click Here

Project Overview

I developed a simple e-commerce application with a microservices architecture. The system allows customers to browse products, place orders, and receive notifications. Instead of building a monolithic application, I split functionality into specialized services that communicate with each other.

The project implements several crucial microservice architectural patterns:

Service Discovery
API Gateway
Circuit Breaker
Event-Driven Architecture
Observability

Project Architecture

The architecture consists of multiple services, an API Gateway, and supporting components like Kafka for messaging and Keycloak for security. Services communicate both synchronously (via REST) and asynchronously (via Kafka).

Core Services

Product Service

Acts as a product catalog
Built with Spring Boot, MongoDB
Provides REST APIs to create and view products
Implements CRUD operations for product management
Technical implementation:

@Document(value = "product")
public class Product {    
    @Id    
    private String id;    
    private String name;    
    private String description;    
    private BigDecimal price;
}

Uses records for DTOs (Java 16+ feature) for clean, immutable data transfer:

public record ProductRequest(String name, String description, BigDecimal price) {}

Order Service

Handles customer orders
Uses MySQL via Spring Data JPA with Flyway for database migrations
Communicates with Inventory Service to check product availability via REST
Sends events to Notification Service via Kafka
Implements Resilience4J for circuit breaking and retry mechanisms
Technical implementation includes transaction management:

@Service
@RequiredArgsConstructor
@Transactional
public class OrderService {    
    private final OrderRepository orderRepository;    
    private final InventoryClient inventoryClient;    
    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;        
    public void placeOrder(OrderRequest orderRequest) {        
        // Check inventory        
        // Create and save order        
        // Send event via Kafka 
    }
}

Inventory Service

Manages product stock information
Provides REST API to check if products are in stock
Uses MySQL database with JPA
Uses Flyway for database schema migrations
Implementation example:

@Transactional(readOnly = true)
public boolean isInStock(String skuCode, Integer quantity) {    
    return inventoryRepository .existsBySkuCodeAndQuantityIsGreaterThanEqual(skuCode, quantity);
}

Notification Service

Sends notification emails to customers
Listens to Kafka events from Order Service
Uses Spring Kafka and Avro for message serialization
Implementation with Kafka listener:

@KafkaListener(topics = "order-placed")
public void listen(OrderPlacedEvent orderPlacedEvent) {   
    // Send email notification
}

Key Patterns & Technologies

API Gateway (Spring Cloud Gateway MVC)

I implemented an API Gateway that serves as the entry point for all client requests. This simplified client interactions and provided a centralized point for cross-cutting concerns like security and routing.

The routing configuration uses the functional programming model in Spring:

@Configuration
public class Routes {
    @Bean
    public RouterFunction<ServerResponse> productServiceRoute() {
        return GatewayRouterFunctions.route("product_service")
                .route(RequestPredicates.path("/api/product"), 
                    HandlerFunctions.http("http://localhost:8080"))
                .filter(circuitBreaker("productServiceCircuitBreaker", "/fallbackRoute"))
                .build();
    }

    // More routes for other services...
}

Security with Keycloak

The API Gateway integrates with Keycloak for authentication and authorization, providing OAuth2 security for all services behind it. I configured the API Gateway as an OAuth2 Resource Server.

Security configuration in the API Gateway:

@Configuration
public class SecurityConfig {
    @Bean
    public SecurityFilterChain securityFilterChain(HttpSecurity httpSecurity) throws Exception {
        return httpSecurity.authorizeHttpRequests(authorize -> authorize
                .anyRequest().authenticated())
                .oauth2ResourceServer(oauth2 -> oauth2.jwt(Customizer.withDefaults()))
                .build();
    }
}

Application properties configuration:

spring.security.oauth2.resourceserver.jwt.issuer-uri=http://localhost:8181/realms/springcommerce

Circuit Breaker Pattern (Resilience4J)

To prevent cascading failures between services, I implemented circuit breakers using Resilience4J. This helps maintain system stability when individual services fail.

Resilience4J operates in three states:

Closed: Normal operation, requests flow through
Open: Circuit is broken, requests are blocked after failure threshold is reached
Half-Open: Testing recovery by allowing limited requests

Configuration in application.properties:

# Circuit Breaker configuration
resilience4j.circuitbreaker.configs.default.registerHealthIndicator=true
resilience4j.circuitbreaker.configs.default.slidingWindowType=COUNT_BASED
resilience4j.circuitbreaker.configs.default.slidingWindowSize=10
resilience4j.circuitbreaker.configs.default.failureRateThreshold=50
resilience4j.circuitbreaker.configs.default.waitDurationInOpenState=5s
resilience4j.circuitbreaker.configs.default.permittedNumberOfCallsInHalfOpenState=3
resilience4j.circuitbreaker.configs.default.automaticTransitionFromOpenToHalfOpenEnabled=true

# Timeout configuration
resilience4j.timelimiter.configs.default.timeout-duration=3s

# Retry configuration
resilience4j.retry.configs.default.max-attempts=3
resilience4j.retry.configs.default.wait-duration=2s

In my implementation, if 50% of requests fail in a window of 10 requests, the circuit opens for 5 seconds, then transitions to half-open where it allows 3 test requests.

Asynchronous Communication (Kafka)

For order notifications, I set up async communication between Order Service and Notification Service using Kafka, implementing a true event-driven architecture.

I used Apache Avro for schema definition and Confluent Schema Registry for schema management:

{
  "type": "record",
  "name": "OrderPlacedEvent",
  "namespace": "com.springcommerce.order_service.event",
  "fields": [
    { "name": "orderNumber", "type": "string" },
    { "name": "email", "type": "string" },
    { "name": "firstName", "type": "string" },
    { "name": "lastName", "type": "string" }
  ]
}

Producing events in the Order Service:

// Inside OrderService.java
OrderPlacedEvent orderPlacedEvent = new OrderPlacedEvent(
    order.getOrderNumber(),
    orderRequest.userDetails().email(),
    orderRequest.userDetails().firstName(),
    orderRequest.userDetails().lastName()
);
kafkaTemplate.send("order-placed", orderPlacedEvent);

Consuming events in the Notification Service:

@Service
@RequiredArgsConstructor
public class NotificationService {
    private final JavaMailSender mailSender;

    @KafkaListener(topics = "order-placed")
    public void listen(OrderPlacedEvent event) {
        // Send email notification logic
    }
}

Observability with Grafana Stack

I integrated the complete Grafana stack to implement comprehensive observability across all services:

Grafana: Visualization dashboard for all observability data
Prometheus: Time-series database for collecting and querying metrics
Loki: Log aggregation system (similar to Elasticsearch)
Tempo: Distributed tracing backend

For logging, I configured Loki integration using logback:

<configuration>
    <appender name="LOKI" class="com.github.loki4j.logback.Loki4jAppender">
        <http>
            <url>http://localhost:3100/loki/api/v1/push</url>
        </http>
        <format>
            <label>
                <pattern>application=${appName},host=${HOSTNAME},level=%level</pattern>
            </label>
            <message>
                <pattern>${FILE_LOG_PATTERN}</pattern>
            </message>
        </format>
    </appender>

    <root level="INFO">
        <appender-ref ref="LOKI"/>
    </root>
</configuration>

For metrics, I used Spring Boot Actuator with Micrometer Prometheus registry:

management.endpoints.web.exposure.include=health, info, metrics, prometheus
management.metrics.distribution.percentiles-histogram.http.server.requests=true
management.observations.key-values.application=order-service

For tracing, I implemented distributed tracing with Micrometer Tracing:

@Observed
public class OrderRepository {
    // Repository methods
}

This setup provides a complete picture of system behavior, making debugging and performance analysis much easier.

Testing Strategy

Integration Testing with TestContainers

Rather than mocking databases, I used TestContainers to spin up actual database instances (MongoDB, MySQL) during tests. This provides a more realistic testing environment.

TestContainers implementation for Product Service:

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class ProductServiceApplicationTests {
    @ServiceConnection
    static MongoDBContainer mongoDBContainer = new MongoDBContainer("mongo:7.0.5");

    @LocalServerPort
    private Integer port;

    @BeforeEach
    void setup() {
        RestAssured.baseURI = "http://localhost";
        RestAssured.port = port;
    }

    static {
        mongoDBContainer.start();
    }

    @Test
    void shouldCreateProduct() {
        // Test code using RestAssured
    }
}

WireMock for Service Simulation

For testing the Order Service without depending on the actual Inventory Service, I used WireMock to simulate the Inventory Service's responses, making tests more reliable and independent.

Setting up WireMock for testing:

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@AutoConfigureWireMock(port = 0)  // Dynamic port allocation
class OrderServiceApplicationTests {
    @Test
    void shouldSubmitOrder() {
        // Stub the inventory service call
        stubFor(get(urlEqualTo("/api/inventory?skuCode=iphone_15&quantity=1"))
            .willReturn(aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("true")));

        // Test order submission
    }
}

Test properties:

inventory.url=http://localhost:${wiremock.server.port}

This approach ensures that our Order Service tests aren't affected by the availability or behavior of the actual Inventory Service.

Development Challenges & Solutions

Challenge 1: Implementing Retry Mechanisms in Order Service

When the Inventory Service was temporarily unavailable, the Order Service would fail completely. I needed a way to make it more resilient.

Solution:

Implemented Resilience4J's retry mechanism in the Order Service
Configured optimal retry parameters (3 retries with exponential backoff)
Added fallback mechanisms for when retries were exhausted
Used Circuit Breaker to prevent overwhelming the system with retries

Implementation in the Inventory Client:

@CircuitBreaker(name = "inventory", fallbackMethod = "fallbackMethod")
@Retry(name = "inventory")
boolean isInStock(@RequestParam String skuCode, @RequestParam Integer quantity);

default boolean fallbackMethod(String code, Integer quantity, Throwable exception) {
    log.info("Cannot get inventory for skucode {}, failure reason: {}", 
             code, exception.getMessage());
    return false;
}

Configuration for the retry mechanism:

resilience4j.retry.instances.inventory.max-attempts=3
resilience4j.retry.instances.inventory.wait-duration=5s

The trickiest part was finding the right balance - too aggressive retrying would flood the system, while too cautious an approach would affect user experience. Testing different configurations under various failure scenarios was essential to find the optimal setup.

Challenge 2: Setting Up Distributed Tracing

Tracking requests as they traveled between services was difficult. When issues occurred, I couldn't easily see which service was causing the problem.

Solution:

Integrated Spring Boot's tracing capabilities using Micrometer
Set up Grafana Tempo to collect and visualize traces
Added trace IDs to logs for correlation
Ensured proper propagation of trace context between services

Dependencies added to implement tracing:

<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-tracing-bridge-brave</artifactId>
</dependency>
<dependency>
    <groupId>io.zipkin.reporter2</groupId>
    <artifactId>zipkin-reporter-brave</artifactId>
</dependency>

Configuration for tracing:

@Configuration
public class ObservationConfig {
    @Bean
    ObservedAspect observedAspect(ObservationRegistry registry) {
        return new ObservedAspect(registry);
    }
}

Property configuration:

management.tracing.sampling.probability=1.0

Tempo configuration in docker-compose.yml:

tempo:
  image: grafana/tempo:2.2.2
  command: ['-config.file=/etc/tempo.yaml']
  volumes:
    - ./docker/tempo/tempo.yml:/etc/tempo.yaml:ro
    - ./docker/tempo/tempo-data:/tmp/tempo
  ports:
    - '3110:3100'  # Tempo
    - '9411:9411'  # zipkin

This gave me an end-to-end view of request flows, making it much easier to debug issues across service boundaries. The most challenging aspect was ensuring that trace context properly propagated across all services, especially when using different communication methods (REST vs Kafka).

Key Learnings

This project reinforced that successful microservices architectures need more than just splitting services - they require careful implementation of patterns for resilience, observability, and communication.

Key technical insights gained:

Design for Failure: In distributed systems, failures are inevitable. Implementing circuit breakers, retries, and fallbacks is essential.
Observability is Not Optional: Without proper logging, metrics, and tracing, debugging distributed systems becomes nearly impossible.
Data Consistency Challenges: Managing data consistency across services requires careful design. Use eventual consistency where appropriate.
Test Infrastructure Matters: Using tools like TestContainers and WireMock resulted in more reliable tests that better reflect production behavior.
Security Requires Planning: Implementing OAuth2 with KeyCloak showed the importance of designing security from the beginning.

Happy Springing! 🚀

Built with ❤️ by Rahul
LinkedIn | GitHub | Portfolio

AtmoFlow: Breathing Life into Data - Real Time Weather and Air Quality Insights

Rahul Reddy Talatala — Mon, 20 Jan 2025 07:26:13 +0000

Introduction

In the realm of data engineering, few challenges are as exciting and impactful as harnessing the power of weather and air quality data. The AtmoFlow project embarks on a thrilling journey to create a robust, scalable data pipeline that leverages the capabilities of Google Cloud Platform (GCP) to process, analyze, and visualize this critical environmental data. Join us as we dive deep into the technical intricacies and architectural decisions that bring AtmoFlow to life.

The project code is available here on GitHub

GCP Architecture Overview

At the heart of AtmoFlow lies a carefully crafted GCP architecture that enables seamless data ingestion, processing, and storage. The key components include:

Cloud Functions: Serverless functions that handle data collection from APIs and historical sources.
Pub/Sub: A reliable messaging system for ingesting streaming data.
Cloud Storage: Scalable object storage for storing raw data files.
Dataproc: Managed Hadoop and Spark clusters for distributed data processing.
BigQuery: A serverless, highly-scalable data warehouse for storing processed data.

These GCP services work in harmony to create a robust foundation for AtmoFlow's data engineering pipeline.

Project Architecture

AtmoFlow's architecture is designed to handle both batch and streaming data efficiently. Here's a high-level overview:

1. Data Ingestion:

Cloud Functions collect data from weather and air quality APIs and historical files.
Streaming data is published to Pub/Sub topics.
Batch data is stored in Cloud Storage.

2. Data Processing:

Dataproc clusters are created on-demand using the DataprocCreateClusterOperator in the Cloud Composer DAG.
PySpark jobs are submitted to the cluster using the DataprocSubmitJobOperator.
The PySpark code (dataproc_stream_batch_pyspark.py) handles the core data processing logic.

3. Data Storage:

Processed data is stored in BigQuery tables, partitioned and clustered for optimal querying.
Fact and dimension tables are created to enable efficient analysis.

4. Orchestration:

Cloud Composer, built on Apache Airflow, orchestrates the entire pipeline.
The DAG (cloud_composer.py) defines the tasks and their dependencies.

5. Visualization:

Looker is integrated with BigQuery to create interactive dashboards.

Data Sources and Ingestion

AtmoFlow combines data from multiple sources to create a comprehensive dataset:

Weather data is fetched from the Open-Meteo API using the openmeteo-requests library.
Air quality data is fetched from the Open-Meteo Air Quality API.
Historical data files are stored in Cloud Storage.

Batch Data Collection

def fetch_air_quality_data(start_date, end_date):
    """Fetch historical air quality data"""
    air_quality_params = {
        "latitude": LATITUDE,
        "longitude": LONGITUDE,
        "start_date": start_date.strftime("%Y-%m-%d"),
        "end_date": end_date.strftime("%Y-%m-%d"),
        "hourly": ["pm10", "pm2_5", "carbon_monoxide", "nitrogen_dioxide",
                  "sulphur_dioxide", "ozone", "aerosol_optical_depth",
                  "dust", "uv_index", "us_aqi"]
    }
    # API request and response handling
    # Returns: DataFrame with historical air quality metrics

Streaming Data Collection

def publish_to_topic(topic_path, data):
    """Publish data to a Pub/Sub topic with dead letter handling"""
    try:
        json_string = json.dumps(data)
        data_bytes = json_string.encode('utf-8')
        future = publisher.publish(topic_path, data=data_bytes)
        future.result()
    except Exception as e:
        # Dead letter queue handling
        error_data = {
            "original_data": data,
            "error": str(e),
            "timestamp": datetime.now().isoformat()
        }
        dead_letter_future = publisher.publish(
            DEAD_LETTER_TOPIC, 
            data=json.dumps(error_data).encode('utf-8')
        )

The batch_cloud_function.py and streaming_cloud_function.py scripts handle the data ingestion process. They use the requests-cache library to optimize API requests and implement retry logic for improved reliability.

Data Processing with Dataproc and PySpark

The heart of AtmoFlow's data processing lies in the dataproc_stream_batch_pyspark.py script, which runs on Dataproc clusters. The script performs the following key tasks:

Reads historical data from Cloud Storage using the defined schemas.
Processes streaming data from Pub/Sub topics.
Merges batch and streaming data using sliding windows and watermarks.
Validates data quality using configurable thresholds and required columns.
Creates dimension tables (e.g., time, location, weather condition, air quality status).
Creates fact tables with derived metrics and aggregations.
Writes processed data to BigQuery tables with partitioning and clustering.

Data Quality Monitoring

def validate_data(df, table_name, threshold=10.0, required_columns=None):
    """Enhanced data validation with configurable thresholds"""
    total_rows = df.count()

    # Validate required columns
    if required_columns:
        missing_columns = [col for col in required_columns 
                         if col not in df.columns]
        if missing_columns:
            raise DataQualityError(
                f"Missing columns in {table_name}: {missing_columns}"
            )

    # Check for null values
    null_counts = df.select([
        sum(col(c).isNull().cast("int")).alias(c) 
        for c in df.columns
    ]).collect()[0]

Merging Stream and Batch Data

def merge_batch_and_stream(batch_df, stream_df, 
                          window_duration="1 hour",
                          watermark_delay="10 minutes"):
    """Merge batch and streaming data with deduplication"""
    stream_with_watermark = stream_df.withWatermark(
        "timestamp", watermark_delay
    )

    # Create time windows
    windowed_batch = batch_df.withColumn(
        "window",
        window(col("timestamp"), window_duration)
    )

    # Combine and deduplicate
    merged_df = windowed_batch.unionByName(
        windowed_stream,
        allowMissingColumns=True
    )

The script leverages PySpark's DataFrame API and SQL functions extensively to perform complex transformations and aggregations efficiently.

Dimensional Modeling and Data Warehouse

AtmoFlow employs dimensional modeling techniques to create a structured and optimized data warehouse in BigQuery. The key components include:

Fact Tables:
- AirQualityFact: Stores air quality measurements with time, location, and status dimensions.
- WeatherFact: Stores weather measurements with time, location, condition, severity, and season dimensions.
Dimension Tables:
- TimeDim: Represents time hierarchy with various time attributes.
- LocationDim: Represents location hierarchy with geographic attributes.
- AirQualityStatusDim: Stores air quality status classifications and descriptions.
- WeatherConditionDim: Stores weather condition classifications and descriptions.
- SeasonDim: Represents seasonal characteristics.
- SeverityDim: Represents severity levels and impact information.

The dimension tables are designed to provide rich context and enable efficient querying and analysis of the fact data.

Data Harmonization

To prepare the data for machine learning applications, AtmoFlow creates a harmonized dataset that combines weather and air quality features. The create_harmonized_data function in the PySpark script joins the relevant data sources and performs feature engineering tasks, such as:

Adding derived features like air density and dew point.
Normalizing and scaling numerical features.
Encoding categorical variables.

def create_harmonized_data(weather_df, air_quality_df):
    """Create harmonized dataset combining weather and air quality metrics."""
    try:
        # Add location key to both datasets
        location_key = hash(f"{config['LATITUDE']}_{config['LONGITUDE']}")

        # Select and rename weather features
        weather_features = (weather_df
            .withColumn("location_key", lit(location_key))
            .select(
                ...
            ))

        # Select air quality features
        air_quality_features = (air_quality_df
            .withColumn("location_key", lit(location_key))
            .select(
               ...
            ))

        # Combine features and add derived metrics
        harmonized_df = (weather_features
            .join(air_quality_features, ["timestamp", "location_key"], "outer")
            .withColumn("air_density",  # Calculate air density
                col("surface_pressure") / (287.05 * (col("temperature_2m") + 273.15)))
            .withColumn("dew_point",  # Calculate dew point
                col("temperature_2m") - ((100 - col("relative_humidity_2m")) / 5)))

        return harmonized_df

The resulting harmonized dataset is stored in BigQuery and can be easily consumed by machine learning pipelines.

Orchestration with Cloud Composer

Cloud Composer, based on Apache Airflow, serves as the orchestration layer for AtmoFlow. The cloud_composer.py script defines the DAG that manages the entire pipeline. The key tasks include:

Triggering the batch and streaming data collection Cloud Functions.
Creating a Dataproc cluster with the specified configuration.
Submitting the PySpark job to the Dataproc cluster.
Deleting the Dataproc cluster after the job completion.

Cloud Composer allows for easy scheduling, monitoring, and management of the pipeline, ensuring smooth execution and enabling data freshness.

Data Visualization and Dashboarding

To unlock the full potential of the processed data, AtmoFlow integrates with Looker for data visualization and dashboarding.

Looker connects directly to the BigQuery tables, allowing users to create interactive and insightful visualizations. Key metrics and trends related to weather patterns, air quality levels, and their impact on various factors can be easily explored and analyzed.

This is the link to the Looker Dashboard.

Challenges and Lessons Learned

Building a robust data pipeline like AtmoFlow comes with its fair share of challenges. Some key lessons learned include:

Complex Data Processing Architecture:

Challenge: The system needed to handle both historical weather data (batch) and real-time updates (streaming) while maintaining data consistency and accuracy.
Solution: Built a dual-pipeline system using PySpark that processes both batch and streaming data. Implemented time-based windows to organize data arrival and created a deduplication system to ensure data integrity.

def merge_batch_and_stream(batch_df, stream_df):
    # Add time tracking for late data
    stream_with_watermark = stream_df.withWatermark("timestamp", "10 minutes")

    # Combine streams and remove duplicates
    merged_df = batch_df.unionByName(stream_with_watermark)
    deduplicated_df = merged_df.dropDuplicates(["timestamp"])

Data Quality Management:

Challenge: Weather and air quality data frequently contained missing values, incorrect readings, or arrived late. The system required robust validation to ensure data reliability.
Solution: Implemented configurable validation rules, built comprehensive error tracking, and developed automatic data cleaning procedures with alert systems for quality issues.

def validate_data(df, table_name, threshold=10.0):
    # Monitor data quality with null checks
    null_counts = df.select([
        sum(col(c).isNull().cast("int")).alias(c) 
        for c in df.columns
    ])

    # Raise alerts for quality issues
    if null_percentage > threshold:
        raise DataQualityError(f"Missing value threshold exceeded in {table_name}")

BigQuery Optimization:

Challenge: Growing data volumes led to increased query latency and costs. The system needed optimization for both performance and cost-effectiveness.
Solution: Implemented strategic partitioning and clustering in BigQuery tables. Designed efficient table structures and query patterns to minimize resource usage.

CREATE OR REPLACE TABLE weather_air_quality.WeatherFact
PARTITION BY DATE(timestamp)
CLUSTER BY location_key, severity_key
AS
SELECT * FROM staging_weather_data;

Complex Data Harmonization:

Challenge: The project required combining weather and air quality data from different formats into a unified, analyzable dataset.
Solution: Developed a standardized data format and built a harmonization system that combines multiple data sources while adding derived metrics for enhanced analysis.

def create_harmonized_data(weather_df, air_quality_df):
    # Combine different data sources
    harmonized_df = weather_df.join(
        air_quality_df,
        ["timestamp", "location_key"]
    )

    # Add calculated metrics
    harmonized_df = harmonized_df.withColumn(
        "air_quality_index",
        calculate_aqi(col("pm25"), col("pm10"))
    )

Conclusion

AtmoFlow showcases the power and potential of data engineering in the cloud, specifically leveraging GCP's robust suite of tools and services. By harnessing the capabilities of Cloud Functions, Pub/Sub, Dataproc, BigQuery, and Cloud Composer, AtmoFlow creates a seamless end-to-end pipeline for processing and analyzing weather and air quality data.

Happy data engineering!🚀

Built with ❤️ by Rahul
LinkedIn | GitHub | Portfolio