Why is it taking so long? Did it crash?

In my previous article in this series (here is a link) there was presented a simple endpoint that returns a list of recipes curated by an LLM based on user input. The result is a nicely structured, but to get it we sometimes have to wait a couple of seconds. Generating a response involves several slower steps, like embedding the user input or waiting for an LLM response. The more complicated the process, the longer the user may wait for the final result. For very complex processes it may even taken hours!

Here is how it looks now:

As you can see, the entire response is returned only after the whole process completes. In the meantime there is no quick feedback to the user about what’s happening, so they may think the application has crashed.

It is better to notify the user about what’s happening — something like “Hey, we got your input and we’re working on it,” followed by “We found some delicious recipes; we’ll show the best ones in a moment”. Moreover if a task is to return multiple items (e.g. list of recipes, rationale, suggested next action) it is good return them once they are produced. There is no need to wait untill all of them are finished. This avoids the impression that the app is stuck.

Streaming chunks

When interacting with most LLM products (ChatGPT, Gemini, etc.), we may notice that responses are returned to us look like they were typed. It’s not just a cosmetic effect, it’s an optimization that informs the user of partial results as soon as they’re available, without waiting for the full response.

There are several solutions to implement streaming chunks (sending small pieces of the entire response). Many variations exist, but they all boil down to two major decisions:

• which communication protocol to choose,
• how to structure each chunk.

Protocols

The first decision we have to make is which communication protocol to select, or in other words, how we’d like to stream chunks from server to client.

Options we have:

• standard HTTP — which can be realized with the following mechanisms: Server-Sent Events (SSE) & Newline Delimited JSON (NDJSON)
• Websockets

All approaches allow a long-lived connection with the server and can send messages in chunks.

We could also consider gRPC, but its strengths shine in cross-agent communication within multi-agent systems, when one agent interacts with another. Therefore I’ll skip it for this post.

Server-Sent Events

Server-Sent Events is a mechanism of unidirectional communication between server and client. That means only the server can send data to the client; the opposite direction is not supported.

To receive an SSE stream the client first needs to make a standard HTTP request with the Accept: text/event-stream header, after which an open connection is established. It will persist until one side (client or server) closes it. During the connection the server pushes messages in plain text; these can be JSON, but they don't have to be.

Technically there are no constraints on how messages are structured. However, conventions most clients and servers follow include:

• a double line ("\n\n") between each message/chunk,
• a message in the format data: <message>,
• an optional event: <event type> field that describes the event (like add, remove, or business-centric types like addedToCart),
• an optional id: <event id> field with a message identifier

So the response, with two events, may look like this:

id: 1 
event: start 
data: Hello 


id: 2 
event: add 
data: there!

SSE is currently the most popular mechanism for AI chats because OpenAI uses it in their streaming API. OpenAI was the first widely used LLM chat provider, and many companies and tools adopted SSE as a result — it has become the de facto standard in the AI chat industry.

Therefore SSE is a good option if you want to integrate your app with popular chat UIs, like Open WebUI, Chainlit, Ollama Desktop App or Jan.ai.

NDJSON

NDJSON stands for Newline Delimited JSON. It’s a format where multiple JSON objects are separated by newline characters (\n). Each line is a valid JSON object and can be treated as a sequence of separate objects/events. This simple format is close to a single JSON array, which makes it easier to integrate with existing tools.

Like SSE, NDJSON relies on an HTTP long-lived connection in which each line (a JSON object) is sent. Unlike SSE, the NDJSON format enforces structured data, which can make it easier to maintain and integrate.

For example, here is how an exemplary response could look (the Content-Type HTTP header would be application/x-ndjson):

{ "id": 1, "event": "start", "data": "Hello"}
{ "id": 2, "event": "add", "data": "there!"}

The NDJSON format is a bit of a niche but it’s used in some systems like Ollama and can be an alternative to SSE.

Websockets

Sometimes one-way communication (server to client only) is too limiting. For instance, systems where an AI agent needs to interact with a human (e.g., to ask for permission or to review plan/task progress — human-in-the-loop) require two-way communication. Such systems benefit from long-lived, bi-directional connections, which avoid a flood of HTTP requests for a single communication channel. WebSocket is useful not only for chats but also for many real-time collaboration systems like Miro or Figma.

Both HTTP and WebSockets are built on the same foundation — TCP (Transmission Control Protocol). WebSocket communication starts with an HTTP handshake in which the client asks the server to upgrade the connection to a long-lived WebSocket. If the server accepts, the same TCP channel remains open and is upgraded to a WebSocket connection; it lasts until either side closes it. In standard HTTP, the client opens a TCP connection, sends a request, and the connection is often closed after the response.

WebSocket supports various data types; the most useful for us is the text type, which allows us to send structured messages (for example JSON) back and forth between client and server. Here is an example of message exchange for a chat application, with messages in JSON format.

In the AI-agent world, WebSockets are used in various products. For example, Perplexity uses WebSockets to give completion suggestions while the user is typing — when I hit a letter, Perplexity often already suggests what I might want to type.

// Client -> Server
{"type":"join", "room":"general", "user":"wojtek"}

// Client -> Server
{"type":"chat", "room":"general", "user":"wojtek", "message":"hello everyone"}

// Server -> Client
{"type":"chat", "room":"general", "user":"alice", "message":"hello Wojtek", "timestamp":"2026-05-12T10:15:00Z"}

// Server -> Client
{"type":"ping"}

// Client -> Server
{"type":"pong"}

The ping/ pong messages are heartbeats - a simple mechanism for client and server to ensure the connection is still alive. The others are the regular messages exchanged between client and server.

Shaping response structure

Once a transport mechanism is chosen, there is one more decision: do we want to send structured responses from the agent? For a simple chat application, markdown may be sufficient. But if you want UI elements (cards, animations, or other components) you need a structured one.

In essence, this is similar to the standard approach: define a response schema (for example in OpenAPI) and the server returns the result. The trick here is how to stream such an object. In most applications the server returns the full JSON object at once; in an agentic application we may want to send only small parts of a large JSON as the LLM produces them.

But first, let’s visualize it with an example. Say the final response for a nutri-chef-ai agent looks like this:

{
    "response": "Thank you for your meal planning request for healthy and fulfilling meals.",
    "suggestedFollowUps": [
        "Prepare a shopping list for the ingredients needed for the selected recipes."
    ],
    "recipes": [
        {
            "recipe": {
                "id": "6464b6f5-17bf-4744-90f6-dbaab3af9983",
                "name": "Mexican Quinoa",
                "description": "...",
                "ingredients": [
                    {
                        "section": "all",
                        "ingredients": [...]
                    }
                ],
                "instructions": [...],
                "sourceUrl": "https://www.bbcgoodfood.com/",
                "source": "bbc_good_food",
                "imageUrl": "https://www.bbcgoodfood.com/quinoa.jpg",
                "servings": "2 portions",
                "tags": [
                    "Lunch"
                ],
                "similarityScore": 0.6430065
            }
        }
    ]
}

This is a large JSON with a lot of information. To provide a seamless experience, you should pick a strategy to split the response into smaller chunks that can be streamed to the client. Here are a few patterns to choose from (these names are informal):

• Snowballing raw response — each chunk re-sends the full accumulated response
• Snowballing structured object — accumulated, valid JSON is emitted token-by-token
• Full-schema delta streaming — stream deltas within a stable schema
• Structured field streaming — one complete JSON field per chunk
• Events streaming — distinguish payload kinds
• Delta patching — each chunk is a typed diff/operation applied to prior state

Snowballing raw response

The first, naive pattern is straightforward. The agentic system emits the response token-by-token. Every new chunk contains the previous content plus the newly added parts, so the message grows with each chunk. Here is an example (I’ll stick to the SSE protocol in all examples):

data: {"response":


data: {"response": "Thank you "


data: {"response": "Thank you for your"

As you can see, the emitted information is not structured in any meaningful way. It is sent token-by-token and it’s up to the client to decide whether the received data forms a valid JSON. This doesn’t differ much from a non-streamed response, since the client often has to wait for all tokens to arrive before deserializing. The next approach addresses this.

Snowballing structured object

Used in: LangGraph (with default streaming mode — values)

A variation of the previous approach is to have a template JSON with empty fields. Each time the LLM generates a token it is inserted into one of the JSON fields. The app sends the accumulated response, containing the newly created parts and the previous ones. Here is an example to visualize it:

data: {"response": "", "suggestedFollowUps": [], "recipes": []}


data: {"response": "Thank you ", "suggestedFollowUps": [], "recipes": []}


data: {"response": "Thank you for your", "suggestedFollowUps": [], "recipes": []}

This approach allows rendering on every incoming token, making the UI feel more responsive. Every chunk is a valid JSON so the UI can update as each chunk arrives. The drawback is that the same data is sent repeatedly while only a small portion changes, making it inefficient as the JSON responses grow.

Full-schema delta streaming

Used in: OpenAI Chat Completions

This problem can be tackled by sending only the tokens that were just generated. Again, each message is a JSON object, but previous responses do not accumulate — only the field that is being generated contains new tokens:

data: {"response": "", "suggestedFollowUps": [], "recipes": []}


data: {"response": "Thank you ", "suggestedFollowUps": [], "recipes": []}


data: {"response": "for your", "suggestedFollowUps": [], "recipes": []}

This way each chunk is smaller and can arrive earlier than the accumulated ones in the previous approach.

The downside is that clients need to cache every chunk and combine them on their end. This approach works well for string fields but can be tricky for complex fields like arrays or objects.

Structured field streaming

The problem of non-string fields can be addressed by sending the entire field content in one chunk. Each chunk contains a complete field of the larger JSON:

data: {"response": "Thank you for your meal planning request for healthy and fulfilling meals."}


data: {"suggestedFollowUps": ["Prepare a shopping list for the ingredients needed for the selected recipes."]}


data: {"recipe": {"id": 1234, "name": "Mexican Quinoa", ... }}

A cost of this approach is that the time between chunks may increase, especially if an object or array contains a lot of data.

Events streaming

Used in: Gemini Interactions API

A variation of the previous approach is to send several fields that are logically connected instead of a single one. The response time for each chunk may increase compared to the previous approach, but in return we get a more consistent response from the agent. We can treat each chunk as an event/message sent to the client, with an id and type, which makes it easier to parse on the client side and to monitor and debug on the server side:

event: rationale
data: {"id": 1, "type":"rationale", "response": "Thank you for your meal planning request for healthy and fulfilling meals."}


event: suggestedFollowUps
data: {"id": 2, "type": "suggestedFollowUps", "suggestedFollowUps": ["Prepare a shopping list for the ingredients needed for the selected recipes."]}


event: recipe
data: {"id": 3, "type": "recipe", "recipe": {"id": "6464b6f5-17bf-4744-90f6-dbaab3af9983","name": "Mexican Quinoa", ... }}

Delta patching

Used in: ChatGPT’s and Perplexity’s web UI

In many cases the previous approach is good enough, but for systems where we don’t want users to wait for each chunk, a variation of Full-schema delta streaming and Events streaming can be applied. This is useful for applications that send a lot of data while keeping the per-chunk size minimal.

As in Events streaming, each chunk is a separate event, but this time the event doesn’t gather related data into a single chunk. Instead, events represent which part of the JSON was generated by providing a field location and the value to add. This way string fields can be sent token-by-token as in Full-schema delta streaming, while complex fields can be sent in one go:

data: {"type":"response.token","payload":"Thank you"}


data: {"type":"response.token","payload":"for your"}

It can get more complicated for operations like adding an item to a list or replacing a previous chunk (e.g., if guard rails detect an incorrect chunk after it was sent). In such cases the app builds a final response using several approaches, one of which is sending events as JSON that consist of three fields describing the operation, location and value:

• o - operation: what action should be applied to the accumulated response, e.g. new, append, replace, add, etc.
• p - path: references a location where the action should be applied (something like JSON Path), e.g. /message/parts/0 (which may translate to the first item of the parts array inside the message object)
• v - value: the change (what should be added, replaced, removed, etc.)

Examples:

data: { "o": "append", "p": "/message/content", "v": "Thank you"}


data: {"o": "add", "p": "/recipes/0", "v": "Crêpe"}


data: {"o": "replace", "p": "/token_count", "v": 1018}

Solution selection

From the earlier sections we can tell there are many options to choose from, and they aren’t limited to the ones described here. So which one did I choose for my nutri-chef-ai project?

My main goal is to learn how to create and tune an AI agent, not to focus on the visual aspects of the project. Of course a nice UI would make working with the plan effortless and fun, but that’s additional work I don’t want to focus on right now. On the other hand, I don’t want to view only raw JSON — I want some visualizations.

After weighing the arguments, I decided to use Chainlit for the UI. It’s an open-source app for building AI conversational solutions. It allows you to create apps similar to ChatGPT or Claude and supports customizations and custom building blocks (views, cards, elements), which mattered most to me.

To integrate with Chainlit, my meal planner had to expose an endpoint compatible with OpenAI’s streaming API. Therefore I chose HTTP and SSE as the transport protocol. Most chunks are sent as whole events; simple text is streamed token-by-token.

Implementation

Agent Flow

Here is how the logic of my agent looks:

Before searching for any recipe, the app acknowledges receipt and starts processing the request. There are two phases — one without an LLM and one with it. The first phase (without the LLM) simply notifies the user that the request was accepted and is being processed. In nutri-chef-ai this appears as a static chunk:

data:{"type":"status","ts":"...","payload":{"phase":"start","message":"Starting meal proposal for: healthy fulfilling meals"}}

The next part involves the LLM in generating a more human-like answer that indicates how the request was understood. This is streamed token-by-token, so it looks like this:

data:{"type":"response.token","ts":"...","payload":"Thank"}

data:{"type":"response.token","ts":"...","payload":" you"}

data:{"type":"response.token","ts":"...","payload":" for"}

data:{"type":"response.token","ts":"...","payload":" your"}

data:{"type":"response.token","ts":"...","payload":" meal"}

data:{"type":"response.token","ts":"...","payload":" planning"}

data:{"type":"response.token","ts":"...","payload":" request"}

// full answer: Thank you for your meal planning request for healthy fulfilling meals. I now search for suitable recipes.

Once that is returned, the agent searches for matching recipes in the RAG database and informs the user about the steps along the way.

data:{"type":"status","ts":"...","payload":{"phase":"search","message":"Searching for matching recipes"}}

data:{"type":"status","ts":"...","payload":{"phase":"search","message":"Found 100 matching recipes"}}

Every step is sent in a single chunk, as is each recipe. The entire structure (shortened below) is sent in one go:

data:{"type":"recipe.selected","ts":"...","payload":{"recipeId":"c01...","name":"Power bowl with sweet potato","ingredients":[...],"instructions":[...], "imageUrl":"https://...jpg", "similarityScore":0.6263648178902887}}

data:{"type":"recipe.selected","ts":"...","payload":{"recipeId":"646..","name":"Mexican Quinoa","ingredients":[...],"instructions":[...],"imageUrl":"https://...jpg","similarityScore":0.6429639599585675}}

Each recipe selection is finalized with a rationale explaining why it was chosen; again, the LLM generates this.

data:{"type":"status","ts":"...","payload":{"phase":"llm","message":"Calling LLM (rationale)"}}

data:{"type":"response.token","ts":"...","payload":"These"}

data:{"type":"response.token","ts":"...","payload":" reci"}

data:{"type":"response.token","ts":"...","payload":"pies"}

data:{"type":"response.token","ts":"...","payload":" ans"}

data:{"type":"response.token","ts":"...","payload":"wer "}

data:{"type":"response.token","ts":"...","payload":"the "}

data:{"type":"response.token","ts":"...","payload":"query"}

//These recipes answer the query "healthy fulfilling meals", because they combine complete sources of protein, ...

The last part of the response is follow-ups — suggestions for next actions or more detailed rationale:

data:{"type":"suggested.follow.ups","ts":"2026-08-06T05:27:39.451857100Z","payload":["Prepare a weekly meal plan and grocery list for a two-person household.","Check other recipes with similar ingredients to enrich my diet."]}

Code structure: callback approach

How the code is organized: I keep core logic in domain services, without framework dependencies. If communication with an external system (service, database, etc.) is required, it happens via interfaces (ports) that can have multiple implementations. This approach is known as Ports & Adapters or Hexagonal architecture.

The proposeMealStreaming(...) method of the MealPlanner class defines the flow for each response. At each step it uses specialized classes to execute a specific task: agents to produce output based on input (from user or another agent), and facades for more complex actions such as searching the vector database.

data class RecipeProposals(
    val response: String,
    val suggestedFollowUps: List<String>,
    val recipes: List<RecipeEntry>,
)

data class RecipeEntry(
    val recipe: Recipe?,
)

@Service
class MealPlanner(
    private val recipeSearch: RecipeSearchFacade,
    private val acknowledgementAgent: AcknowledgementAgent,
    private val recipeSelectionAgent: RecipeSelectionAgent,
    private val rationaleAgent: RationaleAgent,
    private val suggestedFollowUpsAgent: SuggestedFollowUpsAgent,
) {
    companion object {
        private val log = logger {}
        private const val RECIPE_FETCH_LIMIT = 100
    }

    fun proposeMealStreaming(
        userPrompt: String,
        onEvent: (AiAgentEvent) -> Unit,
    ) {
        log.info { "Streaming meal proposals for prompt '$userPrompt'" }
        onEvent(AiAgentEvent.PlanningStarted(userPrompt))

        ...
    }
}

The crucial part of the fun proposeMealStreaming(...) function is the callback onEvent: (AiAgentEvent) -> Unit. It is used to communicate with the outer (controller) layer of the application. Whenever a certain point in the workflow is reached (e.g., recipes have been selected by an agent), the onEvent(...) callback is executed to signal other parts of the application. These signals can then be mapped in the controller to DTOs and sent to the user.

@RestController
@RequestMapping("/api/planner")
class MealPlannerController(
    private val mealPlanner: MealPlanner,
    private val mapper: AiAgentEventMapper,
) {

    @GetMapping(
        path = ["/single"],
        produces = [MediaType.TEXT_EVENT_STREAM_VALUE],
    )
    fun proposeMealSse(
        @RequestParam prompt: String,
    ): ResponseEntity<SseEmitter> {

        mealPlanner.proposeMealStreaming(prompt) { event ->
            val sseEventDto = mapper.toSseEvent(event)
            // send sseEventDto
        }

        return ResponseEntity
            .ok()
            .contentType(MediaType.TEXT_EVENT_STREAM)
            .header("Cache-Control", "no-cache")
            .header("X-Accel-Buffering", "no")
            .body(sseEmitter)
    }
}

This is an elegant approach for handling the asynchronous nature of the communication. This way we can send data to the client as it becomes available, without waiting for the entire workflow to finish.

It’s also worth mentioning that in the above code I’m using Spring MVC as I’m more familiar with it, like many software engineers in the JVM ecosystem. There is however a more performant alternative — non-blocking, reactive approach with Spring WebFlux. For a highly efficient systems I would consider choosing WebFlux over Spring MVC.

Chunks

Focusing on the heart of the meal-planner — the MealPlanner class:

class MealPlanner(
    private val recipeSearch: RecipeSearchFacade,
    private val acknowledgementAgent: AcknowledgementAgent,
    private val recipeSelectionAgent: RecipeSelectionAgent,
    private val rationaleAgent: RationaleAgent,
    private val suggestedFollowUpsAgent: SuggestedFollowUpsAgent,
) {

    fun proposeMealStreaming(
        userPrompt: String,
        onEvent: (AiAgentEvent) -> Unit,
    ) {
        
         log.info { "Streaming meal proposals for prompt '$userPrompt'" }

        // Step 0: stream a brief acknowledgement so the user knows the input was received
        onEvent(AiAgentEvent.PlanningStarted(userPrompt))
        acknowledgementAgent.execute(userPrompt) { token -> onEvent(AiAgentEvent.ResponseToken(token)) }

        // Step 1: fetch candidate recipes
        onEvent(AiAgentEvent.SearchingRecipes())
        val recipes = recipeSearch.findRecipes(userPrompt, RECIPE_FETCH_LIMIT)
        log.info { "Found ${recipes.size} recipes" }
        onEvent(AiAgentEvent.RecipesFound(recipes.size))

        // Step 2: LLM selects the best recipes
        val selectedRecipes = recipeSelectionAgent.execute(userPrompt, recipes)
        if (selectedRecipes.isEmpty()) {
            onEvent(AiAgentEvent.PlanFailed("Failed to find matching recipes. Please try again."))
            return
        }
        selectedRecipes.forEach { entry ->
            entry.recipe?.let { onEvent(AiAgentEvent.RecipeSelected(it.id, it)) }
        }

        // Step 3: LLM streams the rationale for the selected recipes
        val rationaleRecipes = selectedRecipes.mapNotNull { it.recipe }
        rationaleAgent.execute(userPrompt, rationaleRecipes) { token -> onEvent(AiAgentEvent.ResponseToken(token)) }

        // Step 4: LLM suggests follow-up actions
        val suggestedFollowUps = suggestedFollowUpsAgent.execute(userPrompt, rationaleRecipes)
        onEvent(AiAgentEvent.SuggestedFollowUps(suggestedFollowUps))
        onEvent(
            AiAgentEvent.PlanReady(
                RecipeProposals(
                    recipes = selectedRecipes,
                ),
            ),
        )
    }

As mentioned earlier, the agent response is divided into steps. Most steps reduce to:

1. an acknowledgment that indicates the next step in the agent’s “thought” process,
2. executing a specialized agent or class method (e.g., recipe search),
3. sending the result of the step.

Both acknowledgements and results are sent as domain events via the onEvent(...) function. Each event implements the AiAgentEvent interface:

sealed interface AiAgentEvent {
    data class PlanningStarted(
        val prompt: String,
    ) : AiAgentEvent
    
    data class ResponseToken(
        val token: String,
    ) : AiAgentEvent

    class SearchingRecipes : AiAgentEvent

    data class RecipesFound(
        val count: Int,
    ) : AiAgentEvent

    data class RecipeSelected(
        val recipeId: UUID,
        val recipe: Recipe?,
    ) : AiAgentEvent

    data class SuggestedFollowUps(
        val suggestions: List<String>,
    ) : AiAgentEvent

    data class PlanReady(
        val proposals: RecipeProposals,
    ) : AiAgentEvent

    data class PlanFailed(
        val reason: String,
    ) : AiAgentEvent
}

They represent different responses and are sent to the controller layer, which serializes them into the appropriate format.

Server-Sent Events & NDJSON controllers

Let’s look at how events emitted with the onEvent(...) method are consumed and pushed to a client. Here is an implementation for the endpoint that produces SSE:

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter
import java.util.concurrent.Executor
import java.util.concurrent.Executors

@RestController
@RequestMapping("/api/planner")
class MealPlannerController(
    private val mealPlanner: MealPlanner,
    private val mapper: AiAgentEventMapper,
    @Qualifier("mealPlannerExecutor") private val executor: Executor,
) {

    @GetMapping(
        path = ["/single"],
        produces = [MediaType.TEXT_EVENT_STREAM_VALUE],
    )
    fun proposeMealSse(
        @RequestParam prompt: String,
    ): ResponseEntity<SseEmitter> {
        val sseEmitter = SseEmitter(0L)
        executor.execute {
            try {
                mealPlanner.proposeMealStreaming(prompt) { event ->
                    sseEmitter.send(mapper.toSseEvent(event))
                }
            } catch (e: Exception) {
                log.warn(e) { "SSE stream failed" }
                try {
                    sseEmitter.send(mapper.toSseEvent(AiAgentEvent.PlanFailed(e.message ?: "Unknown error")))
                } catch (sendException: Exception) {
                    log.error(sendException) { "Failed to send SSE error event after stream failure" }
                }
            } finally {
                sseEmitter.complete()
            }
        }
        return ResponseEntity
            .ok()
            .contentType(MediaType.TEXT_EVENT_STREAM)
            .header("Cache-Control", "no-cache")
            .header("X-Accel-Buffering", "no")
            .body(sseEmitter)
    }
}

Every AiAgentEvent is consumed, mapped to an SSE event, and then sent using SseEmitter. SseEmitter is a Spring class (a subclass of ResponseBodyEmitter) that allows sending multiple objects. Together with Executor.execute(...), which wraps the controller logic, this enables streaming to run off the HTTP request thread. Here is the Executor bean definition:

@Configuration
class ExecutorConfig {
    @Bean(destroyMethod = "shutdown")
    fun mealPlannerExecutor(): ExecutorService =
        Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors())
}

SSE events should include certain fields — type, ts, and payload. A mapper translates domain events into this envelope, represented as AgentResponseDto:

@Component
class AiAgentEventMapper(
    private val clock: Clock,
) {
    data class StatusEvent(
        val ts: Instant,
        val phase: String,
        val message: String,
    )

    private data class AgentResponseDto(
        val type: String,
        val ts: Instant,
        val payload: Any,
    )

    fun toSseEvent(event: AiAgentEvent): SseEmitter.SseEventBuilder =
        SseEmitter.event().data(toAgentResponseDto(event).toString(Charsets.UTF_8).trimEnd())

    fun toAgentResponseDto(event: AiAgentEvent): ByteArray {
        val now = clock.instant()
        val envelope =
            when (event) {
                is AiAgentEvent.PlanningStarted ->
                    AgentResponseDto(
                        "status",
                        ts = now,
                        payload = StatusEvent(ts = now, phase = "start", message = "Starting meal proposal for: ${event.prompt}"),
                    )
                is AiAgentEvent.ResponseToken -> AgentResponseDto("response.token", ts = now, payload = event.token)
                is AiAgentEvent.SuggestedFollowUps -> AgentResponseDto("suggested.follow.ups", ts = now, payload = event.suggestions)
            }
        return "${envelope.toJson()}\n".toByteArray(Charsets.UTF_8)
    }
}

To keep it short, I limited the number of events to a couple of examples to show how they are mapped.

Besides SSE, the app supports the NDJSON response format. Its implementation is similar to the previous endpoint. Because the application’s domain communicates via domain events, it’s easy to consume events the same way and change only how DTOs are sent:

@GetMapping(
        path = ["/single"],
        produces = ["application/x-ndjson"],
    )
    fun proposeMealNdJson(
        @RequestParam prompt: String,
    ): ResponseEntity<StreamingResponseBody> {
        val body =
            StreamingResponseBody { out: OutputStream ->
                try {
                    mealPlanner.proposeMealStreaming(prompt) { event ->
                        out.write(mapper.toAgentResponseDto(event))
                        out.flush()
                    }
                } catch (e: Exception) {
                    log.warn(e) { "NDJSON stream failed" }
                    out.write(mapper.toAgentResponseDto(AiAgentEvent.PlanFailed(e.message ?: "Unknown error")))
                    out.flush()
                }
            }
        return ResponseEntity
            .ok()
            .contentType(MediaType.parseMediaType("application/x-ndjson"))
            .header("Cache-Control", "no-cache")
            .header("X-Accel-Buffering", "no")
            .body(body)
    }

Unlike SSE, NDJSON is not supported out-of-the-box by Spring MVC, so there is no custom emitter. Fortunately, it can be implemented using the generic StreamingResponseBody interface.

Raw response and Chainlit UI

Here is the end result of how the agent responds. Every chunk is sent to the client as it is produced, so the client doesn’t have to wait for a single large JSON:

curl -N -H "Accept: text/event-stream" "http://localhost:8080/api/planner/single?prompt=chocolate+cake"

And the result:

To make it nicer, I integrated with Chainlit, so I don’t have to read raw chunks. Instead I get a polished UI that can even show recipe images.

I picked Chainlit because it allows easy UI customization. If you’re interested in how I did it, check the project codebase (link in the Summary section).

Summary

I hope this article gave you a clear idea of how to build a responsive Spring MVC endpoint suitable for an agentic application.

If you want the full project code, check my GitHub:

GitHub - wkrzywiec/nutri-chef-ai

References

• Create a model response
• Writing WebSocket servers - Web APIs | MDN
• HTML
• Server-Sent Events: A Comprehensive Guide
• How to Implement Server-Sent Events (SSE) in Spring Boot
• GitHub - Chainlit/chainlit: Build Conversational AI in minutes ⚡️

This article was written by me. ✨ AI was used only to proofread and correct grammar, punctuation, and typos — no content was generated by AI.

Originally published at https://wkrzywiec.is-a.dev on September 6, 2026.

The AI revolution is here. The release of ChatGPT in November 2022 ignited a new revolution in software, where generative AI plays a central role. New tools and patterns have emerged, enabling us as software engineers to build new, exciting projects. It’s hard to keep up because almost every week something new comes out.

At least for me, it’s sometimes hard to keep up with all the new things. To fight off my fear of missing out and to extend my professional toolbox, I have created this series in which I will describe how to build an application powered by AI. But before we start, keep in mind I am not an AI engineer and I have only basic knowledge of machine learning. I am not an expert in this field, but I hope that with this series it will become a little bit clearer to me, and perhaps you, the reader, will also learn something too.

(And I hope that this series won’t share the fate of previous series that I abandoned after one or two entries 😜🤞)

The Project — 👨‍🍳 Nutri Chef AI

I like to learn based on real-life projects. That’s why I’ve decided to build a nutrition assistance application, which will help me plan all my meals for the entire week. I want to keep my meals healthy and balanced, with all needed macronutrients at a good level.

Because I already have a large number of favorite recipes, I don’t want to rely on something from the Internet. I want my AI nutrition assistant to plan my meals based on my prompt and my own cookbook.

This requirement brings us to the first problem — how to provide my entire cookbook to the AI?

Theoretically, we could add all the knowledge to each prompt, but for several reasons, it may not be a good idea:

• it would increase costs — each token sent to AI costs money (no matter if you pay with a subscription or with the electric power of your locally installed AI model),
• it would introduce latency — large inputs usually require the AI to process for much longer,
• it could result in losing the context window — it is not always good to have as big an input as possible because too large inputs may cause some parts to not be taken into account when generating an answer by the AI model; it is always a matter of balancing the size,
• it would not be accepted by the AI model — each AI model has an input token limit which can’t be exceeded.

Instead, we could use only a subset of the knowledge base. We can carefully select only the parts that are most relevant for a task (user prompt). For example, in the Nutri Chef AI app, for a prompt like find all recipes with what is left in my fridge - cheese, ham, egg, paprika, flour, milk, it is not necessary to attach recipes that require other ingredients. We want to receive only those that match the criteria. So we need something smart enough to get only those parts of data that may be relevant before adding them to the system context.

How Retrieval‑Augmented Generation (RAG) works

There are two patterns for how we could achieve that:

1. via Retrieval-Augmented Generation (RAG)
2. via tool calling

As you may guess, this article is focused on the first approach (I’ll do a separate blog post on the latter topic sometime in the future).

RAG is a technique in which an application retrieves the most relevant pieces of information from an external data source (it could be a database, file, or something else) and attaches it to the AI input along with a user prompt. It is up to the application to select data that may be needed to fulfill the task by the AI agent. The app builds the context specific to each user query.

But how an application can select only relevant data?

One way could be based on the keywords included in the user input. This is how Google Search, Elasticsearch, or any other “classic” solution works. They try to find the best matches based on key terms from the input. If the input is low-carb pasta dinner, it may find all documents that have low-carb, pasta, and dinner terms in them. They only analyze the words included in the query and, based on them, prepare the results. They usually do not search for deeper meaning, so in results for the mentioned query, we could get not only low-carb pasta recipes but also regular pasta recipes and low-carb but non-pasta recipes.

There is another approach that addresses this problem — embedding-based retrieval, also known as semantic retrieval.

Embedding-based retrieval

In short, an embedding is a compact representation of data in the form of a multidimensional vector. These vectors represent the (semantic, underlying) meaning of the embedded text. In embedding-based retrieval, we compare a user input, represented as a vector, with the entire knowledge base, also represented as vectors. If, by many measures, two vectors are similar to each other, we consider the data they represent to be relevant.

To visualize it, let’s check the following graph of the x and y axes, which presents three vectors:

Vectors A and B are similar to each other because they are close together, have a similar angle with the x and y axes, and are of similar length. Vector C, however, does not share those similarities with the other two vectors. If vectors A and B represented two pieces of information, we could say that both of them have a similar meaning.

When can we say that two vectors are similar? There are a couple of ways to calculate the distance metric between two vectors, and thus their similarity. Some of them you may already know from math classes:

• l2 (Euclidean) distance
• cosine distance
• l1 distance
• hamming distance
• jaccard distance

Each distance metric has its strengths and is suitable for different data types. For instance, the Euclidean distance is the distance between two endpoints of a position vector, as shown in the picture below:

The larger the differences in length and angle between two vectors, the greater the distance is between them and the greater the dissimilarity.

Each method focuses on different aspects and should be chosen depending on the case. The best approach is to try out each of them and measure which one behaves the best.

From text to vectors: producing embeddings

So the app flow is pretty straightforward. When a user submits a query, we need to transform it into a vector and compare it with the stored vectors to find the most similar ones. But how do we make such a conversion of user input and data? With an AI model, of course!

Embedding models are specialized neural networks designed to convert text into vector representations, and are different from large language models used for chat. For instance, OpenAI is offering three at the moment of writing this — text-embedding-3-small, text-embedding-3-large and text-embedding-ada-002 . We can select whichever model we like, but we need to keep to these rules:

• Tune the number of dimensions based on purpose — a higher number of dimensions is usually good for capturing nuances in data; similar but still different data may have slightly different vectors and hence search may be more accurate. On the other hand, a higher number of parameters may require more resources to compute search and require more storage for larger vectors.
• Model is self-hosted or consumed via API — like for LLMs, we pick from models that we can install locally or use services exposed by AI providers.
• Input and data must be converted using the same model — each model represents space differently, which makes vectors produced not compatible if two different models were used. If you want to compare which embedding model works best for your use case (e.g., via A/B testing), you need to generate embeddings for your data using each model separately.

If you would like to browse available models, go check https://huggingface.co/blog/mteb, which also measures how powerful they all are.

Chunking strategies

One of the problems with data embedding is that the knowledge we would like to embed may be very large, and models may not handle too many input tokens. In fact, each one of them has its own limits.

Therefore, we need to chunk the data into smaller segments. The simple solution would be to split the data by character number. I.e., if a model has a limit of 1000 tokens, we could split the input to be less than that. But it would cause some embeddings to lack the required context, as they would be split in the middle of a sentence or section, which would produce a garbage vector.

Another approach is to chunk input data based on logical units, such as paragraphs, sections, or other meaningful divisions. For instance, the paragraphs of a blog post may be treated as separate chunks, or a section if a paragraph is too granular. This helps preserve context within each chunk, but we still need to ensure that each chunk does not exceed the model’s token limit, which can be checked using available libraries.

Next, a more sophisticated approach could be to logically split the data. If data is already structured, like recipes, chunking could be relatively easy since the data is already broken out into smaller pieces. But what if a knowledge base is not structured? We could use the LLM for that! It could split the data based on certain parameters (e.g., not losing context and not exceeding certain limits), which of course comes with additional cost and may not be accurate (as LLMs may produce nondeterministic results).

As we can see, there are various ways to chunk our data. As always, there is no silver bullet, and sometimes it is better to try out various approaches (different chunking strategies, chunk sizes, etc.) and measure their results.

Storing vectors

Once we’ve got vectors produced by the embedding model, we need to save them somewhere. There are a couple of options for doing that, like storing them in:

1. a file — not efficient but a simple solution
2. an existing database which requires extensions installed. Popular solutions that offer it are, for instance, Postgres, Elasticsearch, Redis, and others.
3. a native vector DB — with the AI RAG revolution, new players have emerged offering native vector storing and search solutions. Examples are: Chroma, Milvus, Qdrant, Weaviate, and others.
4. a cloud provider IaaS — each cloud vendor has its own vector database, e.g. Amazon S3 Vectors, Azure AI Search or Google’s Vertex AI Vector Search.

Each one of them comes with pros and cons. Some are free to use, some are paid but are easy to use, more performant, or do not require you to set up the entire infrastructure.

Using embeddings in applications

Phew, that was a lot of knowledge to be introduced to. Let’s now focus on the application that we’re building. In the Nutri Chef AI app, I want to have two endpoints that would:

• search the best fitting recipes
• propose a meal plan for a day

The workflow for the second one will look like this:

1. User sends a query.
2. The query is transformed into a vector.
3. User input vector is compared with vectors stored in PostgreSQL database. The n most similar chunks are returned each with an id of a different recipe.
4. Full recipes are fetched from db using the recipe ids (this is where the 1st endpoint ends and returns recipes as results).
5. Recipes are formatted and added to the system prompt along with AI instructions and a user prompt.
6. The entire prompt is passed to the LLM.
7. The output is returned to the user as a string

This flow is not specific to the Nutri Chef AI; it is a pretty generic one and could be used in various cases.

The implementation of this entire flow is described in the Enriching prompt with recipe data section of this blog post.

Practical walkthrough

The theoretical introduction is now behind us, so we can move on to implementing a simple RAG in the Nutri Chef AI app. I’ve split this section into two parts:

1. creating embeddings from already stored data
2. implementing RAG flow in the app itself

Embedding recipe data and inserting into pgvector columns

The first step in building the Nutri Chef AI application is to vectorize the recipe data. I already have it in the PostgreSQL database, so the plan is to retrieve it, prepare the request for the embedding process, execute it, and then store the results in the PostgreSQL database.

My data is relatively static. I don’t change it too often, so I’ve decided that vectorizing the data will be a one-off job. The application will use the OpenAI model, so for embedding I’m also using the ChatGPT API.

To achieve this task, I could use a simple endpoint for creating embeddings, which is fast and returns vectors immediately. But in my case, I can wait a little longer for results, and if it’s cheaper, that’s even better. For these reasons, I’ve decided to go with the OpenAI Batch API. This API essentially processes requests in the same fashion as the standard one, but it allows you to combine multiple requests into one. They are then executed asynchronously, and the results are returned after a couple of minutes or hours (depending on how large the dataset is). It may add some overhead to the process (because we need to monitor the async process), but it comes with a lower price per embedding.

The overall process looks like this:

And the steps are:

1. Using Python script the data will be fetched from PostgreSQL database and prepared in the .jsonl format that is accepted by the OpenAI Batch API.
2. Prepared .jsonl file will be sent to OpenAI and then a process of embedding will be started.
3. Using the Python script the batch process will be monitored and once it’s done the resulting file will be downloaded and vectors will be inserted to the database.

This is one of many approaches we could take. If you would like to learn what are the others, go check the Different ways for document embedding process section.

Before moving on to implementation of the scripts first we need to prepare the database. We need to install the pgvector extension by executing the SQL script:

CREATE EXTENSION vector;

Be aware that this extension is not built-in into the standard PostgreSQL version. You need to put the installation files into your PostgreSQL instance first. Or you could use the Docker image provided by creators of the extension:

docker pull pgvector/pgvector:pg17-trixie

More information on how to provide the installation files to the PostgreSQL instance could be found in the GitHub repository of the pgvector extension.

📦 Prepare batch input file

Database is prepared so we could start writing a script for preparing .jsonl file for the batch job. This format is the only accepted by the OpenAI API and it looks somethig like this:

{"custom_id": "61bf11bb_name", "method": "POST", "url": "/v1/embeddings", "body": {"model": "text-embedding-3-small", "input": "Apple pie"}}
{"custom_id": "61bf11bb_ingredients", "method": "POST", "url": "/v1/embeddings", "body": {"model": "text-embedding-3-small", "input": "apples, flour, butter, sugar"}}

As you can see, it is just a file with regular JSONs where each line represents a separate request to the OpenAI API. We’ve got:

• custom_id - which is unique across a file, represents the id of a request,
• method, url & body- which are respectively HTTP method, path and body from "regular" API,

So here is the Python script that generates it:

import json
from pathlib import Path
from typing import Any, Dict, List
import psycopg2

DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "meal_planner",
    "user": "postgres",
    "password": "postgres"
}
EMBEDDING_MODEL = "text-embedding-3-small"

CHUNK_TYPES = ["name", "description", "ingredients", "instructions", "tags"]

def get_db_connection():
    return psycopg2.connect(**DB_CONFIG, client_encoding='UTF8')

def fetch_recipes(conn) -> List[Dict[str, Any]]:
    print("Fetching recipes from database...")
    with conn.cursor() as cur:
        cur.execute("SELECT id, name, description, ingredients, instructions, tags FROM recipe")
        columns = [desc[0] for desc in cur.description]
        recipes = [dict(zip(columns, row)) for row in cur.fetchall()]
        return recipes

def chunk_recipe(recipe: Dict[str, Any]) -> List[Dict[str, Any]]:
    print(f"Chunking recipe {recipe['id']}...")
    chunks = []
    for chunk_type in CHUNK_TYPES:
        content = recipe.get(chunk_type)
        if content:
            chunks.append({
                "recipe_id": recipe["id"],
                "chunk_type": chunk_type,
                "content": content
            })
    print(f"Created {len(chunks)} chunks for recipe {recipe['id']}")
    return chunks

def prepare_jsonl_file(chunks: List[Dict[str, Any]], path: str) -> str:
    record_count = 0
    with open(path, 'w', encoding='utf-8') as f:
        for chunk in chunks:
            record = {
                "custom_id": f"{chunk['recipe_id']}_{chunk['chunk_type']}",
                "method": "POST",
                "url": "/v1/embeddings",
                "body": {
                    "model": EMBEDDING_MODEL,
                    "input": str(chunk["content"])
                }
            }
            f.write(json.dumps(record, ensure_ascii=False) + '\n')
            record_count += 1
    
    print(f"Created file with {record_count} records, path: {path}")
    return path

def main():
    print("Starting embedding process...")
    
    temp_dir = Path("batch-files")
    temp_dir.mkdir(exist_ok=True)
    print(f"Created temporary directory at: {temp_dir}")
    
    try:
        conn = get_db_connection()
        recipes = fetch_recipes(conn)
        
        print("Starting recipe chunking process...")
        all_chunks = []
        for recipe in recipes:
            all_chunks.extend(chunk_recipe(recipe))
        print(f"Created total of {len(all_chunks)} chunks from {len(recipes)} recipes")
        
        base_jsonl_path = temp_dir / "recipes_for_embedding.jsonl"
        jsonl_path = prepare_jsonl_file(all_chunks, str(base_jsonl_path))

    except Exception as e:
        print(f"Error during embedding generation: {str(e)}")
        raise
    finally:
        print("Performing cleanup")
        if 'conn' in locals():
            conn.close()
        print("Database connection closed")

if __name__ == "__main__":
   main()

And after running it we’ve got an output:

Starting embedding process...
Created temporary directory at: batch-files
Fetching recipes from database...
Starting recipe chunking process...
Created total of 25 chunks from 5 recipes
Created file with 25 records, path: batch-files\recipes_for_embedding.jsonl
Performing cleanup
Database connection closed

Simple as that. From the script, you can tell that I’m using the OpenAI API and I’ve picked the text-embedding-3-small embedding model. You can choose from different models, which vary by generation, performance, and price. At the moment of writing this article, there are also text-embedding-ada-002 and text-embedding-3-large. A current list of supported models can be found on the official OpenAI Models website.

Going back to the script, if you look closer you may see that each recipe’s data was split into smaller chunks (name, description, ingredients, instructions, and tags) to make each vector smaller.

🚀Start batch job

An input file is prepared, so we have nothing left but to write a script to first upload it and then start the batch job:

import os
from pathlib import Path
import openai  
from datetime import datetime
from dotenv import load_dotenv


load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
file_path =  Path("batch-files")  / "recipes_for_embedding.jsonl"

def main():
    client = openai.OpenAI(api_key=OPENAI_API_KEY)
    
    file_response = client.files.create(
        file=open(file_path, "rb"),
        purpose="batch"
    )
    file_id = file_response.id
    print(f"File uploaded successfully with ID: {file_id}")
    
    descr = datetime.now().strftime("%Y%m%d_%H%M%S") + " - nutri chef embedding"
    batch_response = client.batches.create(
        input_file_id=file_id,
        endpoint="/v1/embeddings",
        completion_window="24h",
        metadata={
            "description": descr
        }
    )
    job_id = batch_response.id
    print(f"Batch job created with ID: {job_id}")

if __name__ == "__main__":
   main()

The output after running the script:

HTTP Request: POST https://api.openai.com/v1/files "HTTP/1.1 200 OK"
File uploaded successfully with ID: file-HvRA1M3
HTTP Request: POST https://api.openai.com/v1/batches "HTTP/1.1 200 OK"
Batch job created with ID: batch_68c8f2bfTP/1.1 200 OK" File uploaded successfully with ID: file-HvRA1M3 HTTP Request: POST https://api.openai.com/v1/batches "HTTP/1.1 200 OK" Batch job created with ID: batch_68c8f2b

Great! 🎉🎉🎉 The only thing we can do now is sit and relax. The batch job is executed in the background, so how do we know when it finishes? 🤔

🔍 Monitor batch job status

Depending on how large your dataset is, the batch process may take from several minutes to several hours. To verify it, the OpenAI API can be used:

from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
client = OpenAI(api_key=OPENAI_API_KEY)

batch_id = "batch_68c8f2bf"

batch = client.batches.retrieve(batch_id)
print(f"Batch response:\n {batch.to_json()}")

This simple Python script will print out the batch job information:

{
  "id": "batch_68c8f2bf",
  "completion_window": "24h",
  "created_at": 1757999807,
  "endpoint": "/v1/embeddings",
  "input_file_id": "file-HvRA1M3",
  "object": "batch",
  "status": "completed",
  "cancelled_at": null,
  "cancelling_at": null,
  "completed_at": 1757999947,
  "error_file_id": null,
  "errors": null,
  "expired_at": null,
  "expires_at": 1758086207,
  "failed_at": null,
  "finalizing_at": 1757999941,
  "in_progress_at": 1757999869,
  "metadata": {
    "description": "20250916_071647 - nutri chef embedding"
  },
  "output_file_id": "file-BhMQY4",
  "request_counts": {
    "completed": 25,
    "failed": 0,
    "total": 25
  },
  "usage": {
    "input_tokens": 3533,
    "output_tokens": 0,
    "total_tokens": 3533,
    "input_tokens_details": {
      "cached_tokens": 0
    },
    "output_tokens_details": {
      "reasoning_tokens": 0
    }
  }
}

The most important information for us is the status. Above, it states that it is completed, and we can see that it also contains information about the output_file_id - a file with all results.

Before completing it a batch may have various statuses, all of them are listed here. Most likely you will see one of these:

• in_progress - batch is running,
• finalizing - all requests have been executed and the results are being prepared,
• completed - all requests have been executed and the results are available to be downloaded.

💾 Download batch job results

Having the id of an output file, we can download it with this simple script:

from openai import OpenAI
from dotenv import load_dotenv
import os
from pathlib import Path
from datetime import datetime

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
client = OpenAI(api_key=OPENAI_API_KEY)

file_name = "file-BhMQY4"

file_response = client.files.content(file_name)

output_dir = Path("batch-files")

output_file = output_dir / f"{file_name}.jsonl"

with open(output_file, "w", encoding='utf-8') as f:
    f.write(file_response.text)

print(f"Response saved to: {output_file}")

The output:

Response saved to: batch-files\file-BhMQY4.jsonl

After opening the resulting file, we would get something like this:

{
  "id": "batch_req_68c8f34a2aa",
  "custom_id": "61bf11bb_name",
  "response": {
    "status_code": 200,
    "request_id": "8724e1825b5baa0",
    "body": {
      "object": "list",
      "data": [
        {
          "object": "embedding",
          "index": 0,
          "embedding": [
            -0.0075121797,
            0.016524209,
            0.012069914,
            -0.032091618,
           .... other values
          ]
        }
      ],
      "model": "text-embedding-3-small",
      "usage": {
        "prompt_tokens": 13,
        "total_tokens": 13
      }
    }
  },
  "error": null
}

It contains a list of responses for each request, where each one contains a vector representation of each chunk, which will be extracted and inserted into the database in the next step.

📥 Insert vector data into database

The last step is to insert the embeddings into the PostgreSQL database. We could use the already existing table with recipes, but because each recipe was chunked into five pieces, let’s have a new one:

CREATE TABLE recipe_embeddings (
    id                      UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    recipe_id               UUID NOT NULL REFERENCES recipe(id),
    chunk_type              VARCHAR(255),
    token_count             INTEGER,
    embedding               VECTOR(1536)
);

where:

• id - is an id of a row,
• recipe_id - is an id of a recipe,
• chunk_type - tells what kind of chunk it is, values like name, description, ingredients, instructions, tags,
• token_count - tells how many tokens this chunk translates to,
• embedding - is the stored vector.

Having this table set we can execute a following Python script:

import json
import psycopg2
import tiktoken
from typing import List, Dict, Any
from pathlib import Path

embedding_file = "file-BhMQY4.jsonl"

DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "database": "meal_planner",
    "user": "postgres",
    "password": "postgres"
}
EMBEDDING_MODEL = "text-embedding-3-small"

_loaded_request_data = None

def get_token_count(text: Any, model: str) -> int:
    if text is None:
        text = ""
    if not isinstance(text, str):
        text = str(text)
    
    text = text.replace('\x00', '')
    
    enc = tiktoken.encoding_for_model(model)
    try:
        token_count = len(enc.encode(text))
        print(f"Token count: {token_count:,} for text of length {len(text):,}")
        return token_count
    except Exception as e:
        print(f"Error counting tokens for text: {text[:100]}... Error: {str(e)}")
        return 0

def load_request_data():
    global _loaded_request_data
    if _loaded_request_data is None:
        _loaded_request_data = []
        with open('batch-files/recipes_for_embedding.jsonl', 'r', encoding='utf-8') as file:
            for line in file:
                _loaded_request_data.append(json.loads(line))

def find_content(custom_id):
    load_request_data()
    
    for item in _loaded_request_data:
        if item.get('custom_id') == custom_id:
            return item.get('body', {}).get('input')
    
    print(f"Content not found for custom_id: {custom_id}")
    return None

def load_embeddings(file: str) -> List[Dict[str, Any]]:
    all_embeddings = []
    
    file_path = Path("batch-files") / file
    if not file_path.exists():
        print(f"Embedding file not found: {file_path}")
        return all_embeddings
    
    print(f"Loading embeddings from file: {file_path}")
    with open(file_path, "r", encoding='utf-8') as f:
        for line in f:
            try:
                record = json.loads(line.strip())
                custom_id = record.get("custom_id", {})
                recipe_id, chunk_type = custom_id.split('_', 1)

                embedding = record.get("response", []).get("body", []).get("data", [])[0].get("embedding", [])
                content = find_content(custom_id)
                token_count = get_token_count(content, EMBEDDING_MODEL)
                
                all_embeddings.append({
                    "recipe_id": recipe_id,
                    "chunk_type": chunk_type,
                    "embedding": embedding,
                    "token_count": token_count
                })
                    
            except json.JSONDecodeError as e:
                print(f"JSON parsing error in {file_path}: {str(e)}")
            except Exception as e:
                print(f"Error processing line in {file_path}: {str(e)}")
    
    print(f"Loaded total of {len(all_embeddings)} embeddings from all files")
    return all_embeddings


def get_db_connection():
    conn = psycopg2.connect(**DB_CONFIG)
    print("Database connection established successfully")
    return conn


def store_embeddings(conn, embeddings: List[Dict[str, Any]]):
    print("Starting to store embeddings in database")
    with conn.cursor() as cur:
        for idx, emb in enumerate(embeddings, 1):
            cur.execute("""
                INSERT INTO recipe_embeddings (recipe_id, chunk_type, embedding, token_count)
                VALUES (%s, %s, %s, %s)
            """, (emb["recipe_id"], emb["chunk_type"], emb["embedding"], emb["token_count"]))
        conn.commit()
    print(f"Successfully stored all {len(embeddings)} embeddings in database")

def main():
    try:

        all_embeddings = load_embeddings(embedding_file)
        conn = get_db_connection()

        if all_embeddings:
            print(f"Storing total of {len(all_embeddings)} embeddings from all files")
            store_embeddings(conn, all_embeddings)
        else:
            print("No embeddings were generated")
    
    except Exception as e:
            print(f"Error during embedding generation: {str(e)}")
            raise
    finally:
        print("Performing cleanup")
        if 'conn' in locals():
            conn.close()
            print("Database connection closed")


if __name__ == "__main__":
    main()

Here’s what’s really happening under the hood (and don’t worry, it’s not as scary as it sounds!):

1. First, the script grabs both the request file (that’s the one with all the original embedding requests) and the output file (the one OpenAI sends back with the results).
2. Then, for each line in the output file, it finds the matching custom_id in the request file to pull out the original input text. It also extracts the recipe_id, chunk_type, and the actual embedding vector, and counts the tokens in the input (because, well, tokens matter!).
3. Finally, it bundles all this info together and pops it into the database.

That’s it! If you’re like me and sometimes get lost in the details, just follow the steps one by one. And if you spot a better way, let me know-I’m always up for learning something new.

Here is an example output from the script:

Loading embeddings from file: batch-files\file-BhMQY4umjMy228GzgVqvgB.jsonl
Token count: 11 for text of length 28
Token count: 36 for text of length 89
Token count: 210 for text of length 509
Token count: 348 for text of length 855
....
Loaded total of 25 embeddings from all files
Database connection established successfully
Storing total of 25 embeddings from all files
Starting to store embeddings in database
Successfully stored all 25 embeddings in database
Performing cleanup
Database connection closed

Utilizing embeddings in application

We have the embeddings prepared, so we can move on to the actual application. I’ll be writing it in Kotlin with the Spring framework because I feel the most comfortable with it, and since recently it also has its own AI module that I wanted to try out. If you prefer another language or framework, you can still read it since I’ve tried to keep the code as simple as possible so it’s easy to understand and translate to another language/framework.

Search best fitting recipes

The first endpoint that I would like to implement, and which will utilize RAG, is for searching the most matching queries to the user input. It will be the GET /api/recipes/search?limit=<number of recipes>&prompt=<user input> which will return a list of recipes in well-structured JSON format. The limit query parameter refers to how many recipes a client wants to retrieve, and the prompt is a user input string, which could be anything, e.g., find low-carb options for breakfast.

Starting from the controller, which is rather standard:

@RestController
@RequestMapping("/api/recipes")
class RecipeController(
    val recipeSearchFacade: RecipeSearchFacade
) {

    @GetMapping("/search")
    fun findRecipes(
        @RequestParam prompt: String?,
        @RequestParam(defaultValue = "10") limit: Int
    ): ResponseEntity<RecipeSearchResponse> {

        lateinit var matches: List<Recipe>

        val duration = measureTimeMillis {
            matches = recipeSearchFacade
                .findRecipes(prompt, limit)
        }

        val response = RecipeSearchResponse(matches = matches, totalFound = matches.size, searchTimeMs = duration)

        return ResponseEntity.ok(response)
    }
}

data class RecipeSearchResponse(
    val matches: List<Recipe> = emptyList(),
    val totalFound: Int = 0,
    val searchTimeMs: Long = 0,
)

data class Recipe(
    val id: UUID,
    val name: String,
    val description: String? = null,
    val ingredients: List<Ingredients>,
    val instructions: List<Instruction>,
    val sourceUrl: String? = null,
    val servings: String? = null,
    val tags: List<String> = emptyList(),
    val similarityScore: Double? = null,
)

data class Ingredients(val section: String, val ingredients: String)

data class Instruction(val steps: List<String>, val section: String)

The only thing that may raise an eyebrow is the measureTimeMillis block, which I've added to measure the waiting time for a result from the facade, which includes all the steps required to retrieve recipes. And speaking of the steps that the facade implements, here is its code:

@Service
class RecipeSearchFacade(
    private val embeddingEngine: EmbeddingEngine,
    private val repository: RecipeRepository
) {

    companion object {
        private val log = logger {}
    }

    fun findRecipes(prompt: String?, limit: Int): List<Recipe> {
        if (prompt == null || prompt.isBlank()) return emptyList()
        log.info { "Searching for $limit recipes based on a user prompt: '$prompt'..."}

        val promptEmbedding: FloatArray = embeddingEngine.embed(prompt)
        log.info { "User input was embedded. Looking for closest recipes..." }
        return repository.findNearestRecipes(promptEmbedding, limit)
    }
}

The role of a facade is to orchestrate the entire process; therefore, it covers first creating the embedding out of user input and then finding the most suitable recipes.

Let’s check the code of the EmbeddingEngine:

import org.springframework.ai.embedding.EmbeddingModel

interface EmbeddingEngine {
    fun embed(prompt: String): FloatArray
}

@Component
class OpenAIEmbeddingEngine(
    private val embeddingModel: EmbeddingModel
): EmbeddingEngine {
    override fun embed(prompt: String): FloatArray {
        val response = embeddingModel.embedForResponse(listOf(prompt))
        return response.result.output
    }
}

Again, there are not that many lines of code here. With EmbeddingModel, we can call any embedding model to get the vector representation of a user prompt, which in Kotlin is a FloatArray object. It's thanks to the Spring AI framework. The EmbeddingModel is Spring's interface for interacting with various embedding models. The only thing that needs to be done to add it is to insert the following dependency to the build tool (Gradle or Maven):

dependencies {
    implementation("org.springframework.ai:spring-ai-starter-model-openai:1.0.1")
    implementation("org.springframework.ai:spring-ai-autoconfigure-model-openai:1.0.1")
}

I’ve selected the OpenAI model to start, so this is the reason both libraries were added. Apart from that, we also need to select which embedding model we want to use (it needs to be the same as was used for embedding data in PostgreSQL) and provide the API key. The simplest way to achieve this is via autoconfiguration, which requires only providing those values to the application.yaml file:

spring:
  ai:
    openai:
      api-key: ${OPENAI_API_KEY}
      embedding:
        options:
          model: "text-embedding-3-small"

That’s pretty much all that needs to be done to enable text embedding.

With OpenAIEmbeddingEngine, the user input was transformed into a vector, so the only thing to do now is to pass this vector to the SQL query to find the best matching recipes. The logic of it is encapsulated in the findNearestRecipes(promptEmbedding: FloatArray, limit: Int = 10) method of the RecipeRepository class:

@Repository
class RecipeRepository(
    private val jdbcTemplate: NamedParameterJdbcTemplate,
) {

    fun findNearestRecipes(promptEmbedding: FloatArray, limit: Int = 10): List<Recipe> {
        val sql = """
            SELECT
              r.id AS recipe_id,
              r.name,
              r.description,
              re.similarity_score,
              r.ingredients,
              r.instructions,
              r.source_url,
              r.servings,
              r.tags
            FROM (
                SELECT
                  recipe_id,
                  MIN(embedding <=> CAST(:prompt_embedding AS vector)) AS similarity_score
                FROM recipe_embeddings
                GROUP BY recipe_id
                ORDER BY similarity_score ASC
                LIMIT :limit
            ) AS re
            JOIN recipe r ON r.id = re.recipe_id
            ORDER BY re.similarity_score ASC;
        """

        val params = MapSqlParameterSource()
            .addValue("prompt_embedding", promptEmbedding)
            .addValue("limit", limit)

        return jdbcTemplate.query(sql, params, rowMapper())
    }

    fun rowMapper() = RowMapper<Recipe> { rs, _ ->
        Recipe(
            id = UUID.fromString(rs.getString("recipe_id")),
            name = rs.getString("name"),
            description = rs.getString("description"),
            ingredients = parseFromJson<Ingredients>(rs.getString("ingredients")),
            instructions =  parseFromJson<Instruction>(rs.getString("instructions")),
            sourceUrl = rs.getString("source_url"),
            servings = rs.getString("servings"),
            tags =  parseFromJson<String>(rs.getString("tags")),
            similarityScore = rs.getDouble("similarity_score"),
        )
    }

    private inline fun <reified T> parseFromJson(jsonString: String?): List<T> {
        if (jsonString.isNullOrBlank()) return emptyList()

        return try {
            jsonString.toObject<List<T>>()
        } catch (e: Exception) {
            emptyList()
        }
    }

Here I’m using the NamedParameterJdbcTemplate to execute the raw SQL query, which finds the top n (10 by default) recipeIds from the recipe_embeddings table. The list of retrieved recipeIds is then used to get the full recipe data from the recipe table.

The most interesting part is how vectors are compared here:

SELECT
    recipe_id,
    MIN(embedding <=> CAST(:prompt_embedding AS vector)) AS similarity_score
FROM recipe_embeddings
GROUP BY recipe_id
ORDER BY similarity_score ASC
LIMIT :limit

I’m using the <=> operator provided by the pgvector extension. This particular operator finds the shortest value of cosine distance. The results are then sorted from the smallest to the largest value and limited to only the top n results.

pgvector offers other methods of comparing vectors which are:

In order to find the best one for your case, you should try out all of them, experiment, and measure which one is the most suitable for you.

The jsonString.toObject<List<T>>() function code on the String object is an extension method (this is Kotlin's ability to add additional method to a class that is not part of the current codebase):

class ObjectMapperConfig {

    companion object {
        val objectMapper = objectMapper()
    }
}

fun objectMapper() =
    jsonMapper {
        addModule(kotlinModule())
        findAndAddModules()
        serializationInclusion(JsonInclude.Include.NON_NULL)
        defaultDateFormat(StdDateFormat())
        configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
        build()
    }

fun Any.toJson(): String = objectMapper().writeValueAsString(this)

inline fun <reified T> String.toObject(): T = objectMapper().readValue(this)

Everything is coded, so after starting the application and running the GET /api/recipes/search?prompt=<my_prompt> endpoint, where my prompt was a simple find low-carb options for breakfast, I got this result:

{
    "totalFound": 10,
    "searchTimeMs": 873,
    "matches": [
        {
            "id": "00280180-a2ca-4d1d-930e-7a8baf6ff187",
            "name": "Scrambled Eggs with Mushrooms on Toast",
            "description": "A tasty breakfast idea. Scrambled eggs and browned small mushrooms are placed on toasts and sprinkled with chopped parsley or chives.",
            "ingredients": [
                {
                "section": "all",
                "ingredients": [
                    "4 eggs",
                    "4 tablespoons butter",
                    "1 tablespoon vegetable oil",
                    "2 cups small mushrooms",
                    "sea salt and freshly ground black pepper",
                    "4 slices of toasted baguette",
                    "2 tablespoons chopped parsley"
                ]
                }
            ],
            "instructions": [
                {
                "section": "all",
                "steps": [
                    "Prepare two small frying pans. Heat the first, add the oil and 2 tablespoons of butter and melt. Add washed and dried mushrooms and fry for a few minutes until nicely browned on each side. Season with salt and pepper.",
                    "Meanwhile, toast the bread slices in a toaster, spread them with butter and place them on two plates.",
                    "Heat the second pan and melt 2 tablespoons of butter in it, crack in the eggs and season with salt. Fry quickly, stirring only 4–5 times. As soon as the whites set, remove from the pan and place on one prepared toast. Put the mushrooms on the other toast. Sprinkle with pepper and parsley."
                ]
                }
            ],
            "sourceUrl": "https://www.kwestiasmaku.com/dania_dla_dwojga/sniadania/jajecznica_pieczarki/przepis.html",
            "servings": "2 servings",
            "tags": [
                "Mushrooms",
                "Eggs",
                "Eggs for breakfast"
            ],
            "similarityScore": 0.365850391911005
        },
        /// other recipes
    ]
}

Looks amazing 🤩! As you can see, even with a small amount of work and without really using any LLM, we can “intelligently” query our data to find the best matches! Awesome!

Enriching prompt with recipe data

With previous success, let’s build a new thing on top of that — a simple diet advisor. Let’s use the LLM to act as a dietician that focuses on delivering nutritionally balanced recipes based on the user prompt (assuming that the user will be asking for ideas for a specific meal and giving certain boundaries, like a specific diet or favorite ingredients). It won’t generate a full meal plan for the entire week or even a full day-this will be covered in a future blog post.

The first thing to do is to define an endpoint and controller for that:

@RestController
@RequestMapping("/api/planner")
class MealPlannerController(
    private val mealPlanner: MealPlanner
) {

    @GetMapping("/single")
    fun proposeMeal(
        @RequestParam prompt: String,
    ): ResponseEntity<RecipeProposals> {
        val proposals = mealPlanner.proposeMeal(prompt)
        return ResponseEntity.ok(proposals)
    }
}

The MealPlanner is a service that takes responsibility for orchestrating the entire process, which is - finding the most relevant recipes to the user prompt (we already have that) and adding them to the request made to the AI model. Here is the entire code of this class:

import org.springframework.ai.chat.client.ChatClient

data class RawRecipeProposals(val response: String, val nextActions: List<String>, val recipes: List<RecipeIdWithRationale>)

class RecipeIdWithRationale(val rationale: String, val recipeId: UUID)

data class RecipeProposals(val response: String, val nextActions: List<String>, val recipes: List<RecipeWithRationale>)

data class RecipeWithRationale(val rationale: String, val recipe: Recipe?)

@Service
class MealPlanner(
    private val recipeSearch: RecipeSearchFacade,
    private val builder: ChatClient.Builder
) {

    companion object {
        private val log = logger {}
    }

    fun proposeMeal(userPrompt: String): RecipeProposals? {
        log.info { "Searching for best meal proposals based on user prompt... '$userPrompt'"}
        val recipes = recipeSearch.findRecipes(userPrompt, 10)

        log.info {" Found ${recipes.size} recipes with ids: ${recipes.map { it.id }} "}
        log.info { "Calling an AI agent..."}
        val answer = builder.build().prompt()
            .system(
                """
                You are a nutrition assistant that selects the best fitting recipes for meal planning.
                
                TASK: Analyze the provided recipes and select the most suitable ones based on nutritional benefits and user goals. If no specific goal is provided, assume recommendations for a regular healthy adult diet.
                
                RESPONSE FORMAT: You must respond with valid JSON in exactly this structure:
                {
                  "response": "Brief overall explanation of your selection strategy and nutritional focus",
                  "nextActions": ["suggested action 1", "suggested action 2"],
                  "recipes": [
                    {
                      "recipeId": "recipe-uuid-here",
                      "rationale": "Detailed explanation of why this recipe was selected, focusing on specific nutritional benefits"
                    }
                  ]
                }
                
                REQUIREMENTS:
                - Select 3-5 most suitable recipes from the provided list
                - Focus on nutritional balance, variety, and health benefits
                - Include specific nutritional reasons in each rationale
                - Suggest 2-3 relevant next actions for the user
                - Use the exact recipe IDs provided
                - Respond in the same language as the user's request
                - Return only valid JSON, no additional text
                
                RECIPES: ${recipes.toJson()}
                """.trimIndent()
            )
            .user { u -> u.text("USER_QUERY: \"$userPrompt\"") }
            .call()
            .content()
        log.info { "Response from AI Agent:\n $answer" }
        return answer?.let {
            mapToRawRecipeProposals(it)
        }?.let {
            mapToRecipeProposals(it, recipes)
        }
    }

    private fun mapToRawRecipeProposals(answer: String): RawRecipeProposals? {
        return answer.toObject<RawRecipeProposals>()
    }

    private fun mapToRecipeProposals(raw: RawRecipeProposals, fullRecipes: List<Recipe>): RecipeProposals? {
        return RecipeProposals(
            response = raw.response,
            nextActions = raw.nextActions,
            recipes = raw.recipes.map { RecipeWithRationale(rationale = it.rationale, recipe = fullRecipes.find { recipe -> recipe.id == it.recipeId}) }
        )
    }
}

In the first lines of the fun proposeMeal(userPrompt: String) method, we're invoking the already existing recipe search code. The result of it is then transformed into JSON and added to the end of the system prompt. For now, let's not focus too much on the prompt itself; the whole topic of how to write a good prompt will be covered in another article.

In this post, I want to focus on two key points — how the recipes are added to the system prompt and how to enforce the AI to respond in a certain format. In the system prompt, I’m adding those retrieved recipes as JSONs, but in your system, you can try to use different formats like XML or something less structured. Because they are added to the prompt, it is always important to keep in mind that the content of each recipe should be relatively small in order to be below the limit of how many tokens can be added in the request to the AI model.

The second important thing in this prompt is how the output result should be returned because further processing of the AI response depends on that. This is a crucial part because if we want to build a complex system with an AI in the middle of it, we need to enforce returning structured responses from it to make it easier to process by other elements of our system (backend or UI code).

To make it work, we also need to specify which AI model we want to use. It needs to be added to the configuration in the application.yaml file. I've decided to use OpenAI's gpt-5-mini model:

spring:
  ai:
    openai:
      api-key: ${OPENAI_API_KEY}
      chat:
        options:
          model: "gpt-5-mini"
          temperature: 1

After rebuilding the entire project, we can test the GET /api/planner/single endpoint with a prompt: looking for dinner ideas, I'm vegetarian and my favorite vegetable is tomato. And here is an example result:

{
  "response": "I selected vegetarian dishes from the list that maximize the use of tomatoes while providing balanced nutrients: good sources of plant/animal protein (eggs, cheese, beans), healthy fats (olive oil), fiber, and vitamins and antioxidants (lycopene, vitamin C, vitamin A) from vegetables. I focus on a variety of textures and seasons (warm one-pot dish, pasta, baked cheese, cold soup) so you have options for different moods and energy needs.",
  "nextActions": [
    "Choose 1 of the following recipes for today's dinner and I'll provide a shopping list.",
    "If you want more protein, I can suggest simple modifications (e.g., more beans, addition of nuts/seeds).",
    "Do you want a vegan version of any recipe? I'll suggest substitutes for cheese and eggs."
  ],
  "recipes": [
    {
      "rationale": "Halloumi cheese in tomatoes with beans — a great combination of plant protein (beans) and protein/calcium memory from halloumi. Beans increase fiber, iron, and plant protein content, making the dish more filling and stabilizing blood glucose levels. Tomatoes and spices add lycopene and micronutrients. Note: halloumi is quite salty — if you limit sodium, use less cheese or choose less salty mozzarella/packaged cheese.",
      "recipe": {
        "id": "8fbb8ce5-8367-4660-a984-5f5756ad8780",
        "name": "Halloumi cheese in tomatoes with beans",
        "description": "Halloumi cheese can not only be fried or grilled but also baked or cooked in tomato sauce. Cooking keeps the cheese moist and pleasantly soft.",
        "ingredients": [
          {
            "section": "all",
            "ingredients": [
              "600 g fresh tomatoes (e.g., sauce variety) or 400 g canned peeled",
              "1 tablespoon olive oil",
              "2 garlic cloves",
              "1 can (400 g) white beans",
              "about 200 g halloumi cheese",
              "parsley and chives",
              "spices: 1/2 teaspoon smoked paprika, a pinch of chili flakes, 1 teaspoon oregano, salt and pepper"
            ]
          }
        ],
        "instructions": [
          {
            "section": "all",
            "steps": [
              "Peel fresh tomatoes, cut into quarters, remove stems. Dice the flesh. Cut canned tomatoes into smaller pieces, keep the sauce.",
              "In a large pan, gently fry grated garlic in olive oil. Add chopped fresh or canned tomatoes with the sauce and bring to a boil, season with salt, pepper, and other spices.",
              "Cook until the tomatoes reduce (about 10 minutes for fresh tomatoes or 6 minutes for canned tomatoes).",
              "Add canned beans with the liquid and mix. Cook for about 3 minutes.",
              "Place halloumi cheese on top and cook for another 5 minutes. Sprinkle with chopped parsley and chives."
            ]
          }
        ],
        "sourceUrl": "https://www.kwestiasmaku.com/przepis/pieczony-ser-halloumi-w-pomidorach-z-fasola",
        "servings": "2 servings",
        "tags": [
          "Gluten-free",
          "Warm dinners",
          "halloumi",
          "Greek cuisine",
          "Fit recipes",
          "Dinners",
          "Lunches",
          "Tomatoes",
          "Vegetarian",
          "Beans",
          "Hits of Kwestia Smaku"
        ],
        "similarityScore": null
      }
    },
    //other proposals
  ]
}

Looks amazing 🤩! As you can see even with small amount of work and with out really using any LLM we can “intelligently” query our data to find best matches! Awesome!

It returns not only the list of matched recipes but also the rationale for each recipe. There is also a text summary for the entire prompt with actions that the user may want to do next to get better results.

Going deeper

That’s pretty much it. With only a few lines of code, we have built a RAG system! It is very simple, but it may already make an impact. So what’s next? How could we make it even better? Oh, there are a lot of things that may be done better.

For instance, before providing the user prompt to the recipe semantic search service, we could first ask another AI agent to analyze the user input and produce a better prompt that would be embedded and then used to search for proper recipes. With this technique, we could get even more precise results, focusing on finding the most fitting but also nutritious meals possible from the cookbook.

This is just one of many possible improvements to make the RAG system more precise, faster, or cheaper. Before wrapping up, I want to briefly describe two additional enhancements you can consider.

Indexing vectors

The knowledge base for this project is relatively small. Finding the most similar vectors across thousands of vectors is a relatively quick operation. But what if we had billions or even trillions of vectors? Definitely, the retrieval process could get significantly longer.

One way of coping with that is to create indexes with pgvector that would cluster similar vectors into a smaller number of groups of vectors. With indexes, we do not search for vectors with the shortest distance but for groups of vectors. This way, we lose a little bit of the precision of finding the exact vector, but in return, we gain faster retrieval.

The pgvector extension supports two algorithms for creating such indexes — HNSW (Hierarchical Navigable Small World) & IVFFlat (Inverted File Flat). I won’t be going deep into either of them, but you can check these great articles on the first and the second algorithm respectively.

Different ways for document embedding process

In the hands-on section of this article, I’ve described how I wrote the Python script that takes care of creating embeddings. I’ve decided to go with a Python script and use the batch API of OpenAI because, for my case, it is a one-time job. I don’t modify my cookbook too often, so the entire procedure could be done only once and never repeated. However, this is not the case in every system. Data in some systems changes quite often, and if new data should be embedded, then other approaches should be taken.

One of them could be to run a similar script periodically on a monthly, weekly, or daily basis. The script would either wipe down all existing vectors and replace them with new ones or only update and add those that have changed or were added. This, of course, requires further development of a script and good monitoring of it since it becomes a crucial part of the data pipeline for the entire system.

If periodic processing is not enough and we would like to have those vectors stored at the moment of storing the corresponding data, then another approach should be taken. Instead of the script, the logic of creating the embedding should be baked into the application logic itself. When the data is updated/inserted into the database, the application should also create a vector and store it. This approach certainly has advantages, like the data could be searched semantically right away, but the caveat is that since we’re introducing the external dependency (embedding model), it may slow down the modification operation or even prevent them from happening. There are, of course, workarounds for those problems, but with the cost of higher complexity of the entire system and slowing down the process of embedding. As always, each solution has its own tradeoffs.

The last solution could be to use the pgai extension for PostgreSQL. It enables you to create and synchronize vector embeddings directly in the PostgreSQL instance. You don’t need to add any logic to the app or create complicated scripts to periodically update the entire database. With a simple configuration, all of that could be achieved inside PostgreSQL. This way, we move the logic from the app to the database, which in some cases could be a good solution, but in others it may be bad because it makes the entire data flow less explicit.

Summary

This is it for today! I hope you learned something new, like how to transform your data into vector embeddings, how to use them for semantic search, and how to write your first AI agent that leverages your own knowledge base.

The entire codebase for this project can be found on my GitHub: wkrzywiec/nutri-chef-ai.

Until the next time!

References

AI Engineering

https://www.datacamp.com/tutorial/introduction-to-text-embeddings-with-the-open-ai-api

• A Beginner's Guide to Vector Embeddings
• Vector Similarity Search with PostgreSQL's pgvector
• PostgreSQL as a Vector Database: A Pgvector Tutorial
• Chunking Strategies to Improve LLM RAG Pipeline Performance | Weaviate

https://github.com/openai/openai-cookbook/blob/main/examples/Question_answering_using_embeddings.ipynb

Originally published at https://wkrzywiec.is-a.dev on October 22, 2025.

The key element in event sourcing are (surprise, surprise) events. These events are created after each business action is performed on a domain object (aggregate) and need to be persisted somewhere. They are also used to rebuild the state of domain objects, but first, they need to be fetched from some kind of storage. For these (and other) reasons, the central piece of an event sourcing system is the event store which will be built through out this post.

If you are not familiar with event sourcing, please check my previous article on that topic. — Is Event Sourcing hard? Part 1: Let’s build a domain object from events.

An event store is crucial in event sourcing applications. But what requirements should it meet?

• Event store should allow storing events.
• Events stored in the event store can’t be modified — Events are business actions that occurred in the past. We cannot change history, and we cannot change events stored in the event store. Therefore, everything stored there remains as it was.
• Events must be added to the end of a stream — Every event persisted must be ordered, at least within a stream. A stream is a collection of events that refer to a single domain object. The event store should forbid saving any event in any position other than the end of a stream.
• All events in a stream are ordered — Not only must storing events preserve their order, but loading them must also maintain the order of their occurrence. This is crucial for replaying them to rebuild the state of a domain object.

These are the requirements that will be covered by the implementation of the event store in this article. Fully-fledged event stores provide other capabilities, such as:

• Events can be fetched up to a certain point in time — Sometimes, there is a need to see how the domain object looked in the past. Therefore, only a subset of events can be loaded, starting from the beginning and finishing at the event that occurred before the given point in time. That’s real time traveling in practice!
• Events are globally ordered — It’s beneficial to order events not only within a stream but also across streams, especially when interaction between domain objects is needed or building projections from multiple streams.
• Notify subscribers about changes in a stream — An event store may not only play the role of data storage but also act as a pub-sub solution. Whenever a new event is appended to a stream, subscribers (e.g., a service responsible for updating projections) are notified about it to perform their tasks.
• Store bi-temporal events — An event store may not only store information about the past but can also be used to store facts about the future! Bi-temporal event sourcing is a technique of adding a new property validAt to events, representing when they are valid and not when they were stored. This allows storing events from the future, which may be beneficial if we're sure that something will happen at a certain point in time. For example, a discount can be applied to a product in an e-commerce shop before Black Friday, but it will be valid only on that day.

These are only examples of capabilities. There are certainly more, and you can check them in one of the off-the-shelf solutions, which are either fully operational databases or frameworks built on top of popular databases (e.g., PostgreSQL):

That’s it for the core concepts of an event store. There are, of course, more nuances, but for this blog post, I’ll limit it to the basics.

Let’s get our hands dirty and implement these concepts in a Java application using PostgreSQL as the storage technology.

Readers discretion is adivsed — the following implementation is only one of many possible ones. Depending on the project, requirements, and team preferences, it may look different. There is no one implementation to rule them all. I want you to keep that in mind; this is my preferred solution for my pet project. But of course, you’re more than welcome to share your opinion so we can all learn from each other.

Creating a Database Table

The first step is to define a table where all events will be persisted. Here is the SQL statement to create a simple events table:

CREATE TABLE IF NOT EXISTS events
 (
    id                  UUID NOT NULL,
    stream_id           UUID NOT NULL,
    type                VARCHAR(256) NOT NULL,
    data                JSONB NOT NULL,
    added_at            TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
);

Where:

• id - is the identification of an event,
• stream_id - is the ID of the domain object that the event pertains to,
• type - is the definition of the event type (e.g., OrderDelivered),
• data - all information that is part of an event is stored here. The JSONB column type indicates that it is stored in a JSON format,
• added_at - is the timestamp of when the event was added/occurred.

This statement can be part of a data migration tool, such as Liquibase or Flyway.

And that’s pretty much it (at least for now). We have a very simple table where we can store events. Therefore, let’s move on to implementing the part of the code that is responsible for saving events.

Storing Events

The first step is to define the interface of the event store so that we can have multiple implementations of it:

public interface EventStore {

    void store(EventEntity event);

    default void store(List<EventEntity> events) {
        events.forEach(this::store);
    }
}

Nothing fancy here. The class that will implement this interface will need to cover the store(EventEntity event) method. In the case of a list of events, the interface already provides a default implementation.

The EventEntity object represents the record from the database table that was just defined:

@Data
public class EventEntity {

    protected UUID id;
    protected UUID streamId;
    protected String type;
    protected DomainEvent data;
    protected Instant addedAt;
}

The DomainEvent in my case is an interface. Each event holds its own set of data but must provide information about the ID of the stream into which the event will be stored:

public interface DomainEvent {
    UUID streamId();
}

And here is an example of a concrete class, which is a Java record:

record DeliveryCanceled(
        UUID orderId,
        String reason
) implements DomainEvent {

    @Overrides
    default UUID streamId() {
        return orderId();
    }
}

Since in the context of a delivery the streamId is the same as the orderId, the method defined in the interface returns the orderId.

Returning to the EventStore interface, here is an implementation of the store(EventEntity event) method:

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;

import java.sql.Timestamp;
import java.util.List;

@Slf4j
public class PostgresEventStore implements EventStore {

    private final NamedParameterJdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;

    public PostgresEventStore(NamedParameterJdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
    }

    @Override
    public void store(EventEntity event) {
        log.info("Saving event in a store. Event: {}", event);

        var bodyAsJsonString = mapEventData(event.data());

        SqlParameterSource parameters = new MapSqlParameterSource()
                .addValue("event_id", event.id())
                .addValue("stream_id", event.streamId())
                .addValue("type", event.type())
                .addValue("data", bodyAsJsonString)
                .addValue("added_at", Timestamp.from(event.addedAt()));

        jdbcTemplate.update("""
                INSERT INTO events(id, stream_id, type, data, added_at)
                SELECT :event_id, :stream_id, :type, :data::jsonb, :added_at
                """,
                parameters)
        );
        log.info("Event was stored.");
    }

    private String mapEventData(DomainEvent domainEvent) throws RuntimeException {
        try {
            return objectMapper.writeValueAsString(domainEvent);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to map DomainEvent to JSON before storing event.", e);
        }
    }
}

Since my application is written with Spring Boot, the NamedParameterJdbcTemplate was used to integrate with PostgreSQL. Apart from that, the only other role of the method is to transform the Java object of the DomainEvent into its JSON String representation, for which the Jackson ObjectMapper is used.

If you would like to test if this code is really working, you can check the integration test that I’ve written, which is available here: PostgresEventStoreIT.groovy .

Loading Events

Persisting events is covered, so let’s now implement the code for fetching them. First, we need to add a method definition to the interface:

public interface EventStore {

    List<EventEntity> loadEvents(UUID streamId);
}

And concrete implementation of it:

@Slf4j
public class PostgresEventStore implements EventStore {

    private final NamedParameterJdbcTemplate jdbcTemplate;
    private final EventPostgresRowMapper eventPostgresRowMapper;
    private final ObjectMapper objectMapper;

    public PostgresEventStore(NamedParameterJdbcTemplate jdbcTemplate, ObjectMapper objectMapper, EventClassTypeProvider eventClassTypeProvider) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
        this.eventPostgresRowMapper = new EventPostgresRowMapper(objectMapper, eventClassTypeProvider);
    }

    @Override
    public List<EventEntity> loadEvents(UUID streamId) {

        return jdbcTemplate.query("""
                SELECT id, stream_id, type, data, added_at
                FROM events
                ORDER BY added_at ASC;
                """,
                eventPostgresRowMapper,
                streamId);
    }
}

Nothing fancy here, except for the EventPostgresRowMapper and EventClassTypeProvider. These components are responsible for deserializing information stored in the database into EventEntity and its data part ( DomainEvent), respectively.

@RequiredArgsConstructor
@Slf4j
class EventPostgresRowMapper implements RowMapper<EventEntity> {

    private final ObjectMapper objectMapper;
    private final EventClassTypeProvider caster;

    @Override
    public EventEntity mapRow(ResultSet rs, int rowNum) throws SQLException {
        String eventType = rs.getString("type");
        DomainEvent event = extractData(rs, eventType);
        return new EventEntity(
                rs.getString("id"),
                rs.getString("stream_id"),
                eventType,
                event,
                extractInstant(rs, "added_at")
        );
    }

    private static Instant extractInstant(ResultSet rs, String columnName) throws SQLException {
        if (rs.getTimestamp(columnName) == null) {
            return null;
        }
        return rs.getTimestamp(columnName).toInstant();
    }

    private DomainEvent extractData(ResultSet rs, String eventType) throws SQLException {
        String bodyAsString = rs.getString("data");
        JsonNode bodyAsJsonNode = mapToJsonNode(bodyAsString);
        return mapEventBody(bodyAsJsonNode, eventType);
    }

    private JsonNode mapToJsonNode(String eventAsString) {
        try {
            return objectMapper.readTree(eventAsString);
        } catch (JsonProcessingException e) {
            log.error("Failed to parse event to json: {}", eventAsString);
            throw new RuntimeException("Parsing error", e);
        }
    }

    private DomainEvent mapEventBody(JsonNode eventBody, String eventType) {
        Class<? extends DomainEvent> classType = caster.getClassType(eventType);
        return mapEventBody(eventBody, classType);
    }

    private <T extends DomainEvent> T mapEventBody(JsonNode eventBody, Class<T> valueType) {
        try {
            return objectMapper.treeToValue(eventBody, valueType);
        } catch (JsonProcessingException e) {
            log.error("Failed to parse event json: {} to '{}' class", eventBody, valueType.getCanonicalName());
            throw new RuntimeException("Parsing error", e);
        }
    }
}

The implementation of the EventPostgresRowMapper is quite generic and can be reused. It extends Spring's RowMapper, where we define how the ResultSet from the database should be used to build an EventEntity.

The only part that must be adjusted to specific DomainEvents is the EventClassTypeProvider, which is an interface that helps the ObjectMapper cast the result from the data column into the appropriate Java class:

public interface EventClassTypeProvider {

    Class<? extends DomainEvent> getClassType(String type);
}

And an example of its implementation:

@Slf4j
class DeliveryEventClassTypeProvider implements EventClassTypeProvider {

    @Override
    public Class<? extends DomainEvent> getClassType(String type) {
        return switch (type) {
            case "DeliveryCanceled" -> DeliveryEvent.DeliveryCanceled.class;
            default -> {
                log.error("There is not logic for mapping {} event from a store", type);
                throw new IllegalArgumentException();
            }
        };
    }
}

As before, tests for this functionality are available in PostgresEventStoreIT.groovy class.

Optimistic Locking for Strong Consistency

The implementation we have so far is pretty basic. It may work for a small side project used by only one person, but it may not fit the needs of a larger solution (and I’m not claiming it’s production-ready yet!) where multiple users may try to store different events to the same stream at the same moment.

The foundation of the event store is that all events are stored in the order of their occurrence. But what happens if, at the same time, two threads want to append to the same stream? Which one should be accepted? Maybe both? If it were a bank transaction, do we want to accept two events that would reduce the account balance below zero? How do we ensure that events that should not happen are not stored in the event store?

One way of handling this would be pessimistic locking. It’s a database mechanism that prevents a row (or an entire table) from being read or modified if another transaction is making changes to it. If there is any chance that a thread reading data performs it at the moment when an update is occurring, the database will force the reading thread to wait until the update thread finishes.

This could be a good candidate for handling conflicts in the event store. However, besides the inherent problem of possible deadlock occurrence, it introduces additional limitations. Locking the entire table would introduce a bottleneck to the entire application because, in the case of adding a new event, reading them (or adding new ones to another stream) won’t be possible due to the lock. It would make the event store really slow, especially in the case of lots of write operations.

A better approach would be to use optimistic locking. This is not a built-in database mechanism, however. It’s a strategy for handling concurrent operations. In the “regular” approach to storing business information, optimistic locking is realized by introducing an additional integer column — version. An application first needs to load data from the database, including the version, and then it performs tasks based on it. As a result, it updates the same row in the database. Within an update statement, the current version of the row must be provided. If it is the same as in the database, the row is updated, and the version column is increased by one. If one of the threads tries to update a row but with an old version, the database should forbid it.

With the event store, we don’t follow this pattern since we don’t modify rows, but we can use optimistic locking to ensure that only correct events are stored and they are ordered.

The first step is to add a BIGINT version column to the events table:

CREATE TABLE IF NOT EXISTS events
 (
    stream_id           UUID NOT NULL,
    version             BIGINT NOT NULL,
    ... other columns

    UNIQUE (stream_id, version)
);

Please note that besides adding a new column, I’ve added a unique constraint on the pair of stream_id and version columns. This way, we will ensure that there are no rows with the same pair of values.

However, this is not the only change that needs to be made. The concept of versions must be introduced to the code itself. The decision on how deeply this concept will penetrate the code is an important one to make.

The least invasive method would be to extend the storing method of the EventStore interface with the latestVersion parameter. This version still needs to be obtained from somewhere. REST endpoints must include it, and in a message-based system, incoming messages must include it. But the good news is that the version does not leak into the domain logic.

I, however, have decided to go in the opposite direction. I have introduced versions into the domain logic as well, mainly because I wanted to encapsulate the versioning logic within the domain layer to keep the application layer as thin as possible. Was it a good decision? Maybe. Maybe not. I’ll see in the future as the project evolves.

From the previous article, you may remember that I generate DomainEvent within each domain method, e.g.:

public class Delivery {

    private int version;
    private DeliveryStatus status;
    List<DomainEvent> changes = new ArrayList<>();

    public void cancel(String reason) {
        if (status != DeliveryStatus.CREATED) {
            throw new DeliveryException(format("Failed to cancel a %s delivery. It's not possible do it for a delivery with '%s' status", orderId, status));
        }
        this.status = DeliveryStatus.CANCELED;

        increaseVersion();
        changes.add(new DeliveryEvent.DeliveryCanceled(orderId, version, reason));
    }

    private void increaseVersion() {
        this.version = this.version + 1;
    }
}

The introduction of versioning requires incrementing the version whenever a domain object is modified (hence the increaseVersion() method). It also necessitates introducing a version to each DomainEvent:

public interface DomainEvent {
    UUID streamId();
    int version();
}

And that’s pretty much what is peculiar in my project. Other parts should be the same or quite similar. So let’s jump into the new implementation of the EventStore.

But before that, let’s list all the requirements that it needs to cover because optimistic locking introduces these:

• Only events with a version succeeding the latest version within a stream can be stored.
• Events with an already stored version can’t be persisted.
• There should not be gaps between versions.
• Loaded events are ordered by version.

With all of this in mind, let’s see the implementation:

public class PostgresEventStore implements EventStore {

    @Override
    public void store(EventEntity event) {
        log.info("Saving event in a store. Event: {}", event);

        var bodyAsJsonString = mapEventData(event.data());

        SqlParameterSource parameters = new MapSqlParameterSource()
                .addValue("event_id", event.id())
                .addValue("stream_id", event.streamId())
                .addValue("version", event.version())
                .addValue("type", event.type())
                .addValue("data", bodyAsJsonString)
                .addValue("added_at", Timestamp.from(event.addedAt()));

        int insertedRowsCount = 0;
        try {
            insertedRowsCount = jdbcTemplate.update("""
                WITH latest_version AS (
                    SELECT MAX(version) AS value
                    FROM events
                    WHERE stream_id = :stream_id
                )
                INSERT INTO events(id, stream_id, version, type, data, added_at)
                SELECT :event_id, :stream_id, :version, :type, :data::jsonb, :added_at
                FROM latest_version
                WHERE :version = value + 1 OR :version = 0
                """,
                    parameters
            );
        } catch (DuplicateKeyException exception) {
            log.error("Failed to persist event in event store due to duplicate key violation", exception);
            throw new InvalidEventVersionException();
        }

        if (insertedRowsCount == 0) {
            log.error("Event was not stored due to invalid event version");
            throw new InvalidEventVersionException();
        }
        log.info("Event was stored");
    }
}

Phew, that’s a lot of code! Let’s unwrap it.

First of all, we need to know which is the latest version in a stream. Hence, the Common Table Expressions (CTE), WITH latest_version, was added to fetch it. Then it is used in the WHERE clause to verify that the version provided in the event is exactly one larger than the latest version. There is also another condition that checks if an event is the first one in a stream; for such events, the previous condition is not needed.

The jdbcTemplate.update(...) method is wrapped with a try ... catch ... block because the unique key constraint for stream_id and version was introduced. If a violation occurs, a custom InvalidEventVersionException is thrown.

The same exception is thrown if nothing is persisted in the database. This may happen if any of the mentioned conditions are not met.

The above implementation covers all requirements, and if you would like to verify it, there are tests available in the PostgresEventStoreIT.groovy class.

Please note that these are not the only changes that are needed. The ones that I have skipped include adding the version field to the EventEntity, adding deserialization logic for this field in EventPostgresRowMapper, and adjusting the loading of events in the PostgresEventStore class to order events by version. These changes are very simple, so I have kept them out of this article, but they can be reviewed in the repository wkrzywiec/farm-to-table.

Microservice world — one or per-service event store?

I really like the shape of the event store that we have so far, so let’s add another twist to it: how to fit an event store in a microservice world?

The rule of thumb in the microservice world is that shared databases must be avoided. Services should have their own databases and communicate with each other using an established API. Sooner or later, a shared database will lead to a situation where one of the services needs to introduce a non-backward-compatible change (e.g., rename/delete a column). Since other services are using the same database, they also need to change something on their side. This makes independent service deployment impossible. Deployment of all services must be done simultaneously with data migration.

For this main reason, most microservices have their own databases. So in the case of an event store, should we follow the same pattern? As always, it depends.

Let’s play devil’s advocate and try to figure out if a shared event store is a good idea. First of all, the problem mentioned above is rather non-existent because an event store schema is relatively stable. Moreover, event stores are append-only logs, so once we save data there, it should not be modified, which also limits the need for data migration. Besides addressing the major pain point of a shared database, a common event store may be very helpful in integrating services. By introducing a subscription mechanism, services may start to listen for certain events that are stored and act upon them. So there would be no need to have a fancy event bus (like Kafka or RabbitMQ). Also, having data in a single place makes it much easier to query, which may be very helpful in troubleshooting or preparing various reports.

Of course, having a single database has its downsides too. It is a single point of failure — if it does not work, most parts of your system are not working either. Also, with a large number of microservices, it must be prepared to handle a large number of concurrent connections, which might be quite a challenge even for a medium-sized distributed system. And finally, I mentioned that the database schema is stable and does not change, but that may not be true. It can evolve over time, especially when new concepts must be introduced (like global ordering or bi-temporal events) but weren’t planned ahead.

There might be, of course, other factors than the ones mentioned above. So whenever you need to decide which solution to use, don’t limit yourself to one way or another. Try to weigh all the pros and cons and decide which approach is better for you.

In my pet project, I have decided to use the shared event store, chiefly because it is a small project. A single event store removes the unnecessary task of managing multiple databases and finally gives me an opportunity to try to implement the subscription mechanism.

As a consequence of this decision a new column to the events table was added. The channel VARCHAR column stores information which service is writing to the particualr stream. In my case if it's an ordering service a value is ordering and if it's a delivery service a value if delivery. Pretty straight-forward.

As a consequence of this decision, a new column was added to the events table. The channel VARCHAR column stores information about which service is writing to the particular stream. In my case, if it's an ordering service, the value is "ordering", and if it's a delivery service, the value is "delivery". Pretty straightforward.

The implementation of this is simple (it is only the addition of a new column), so I’ll skip it here. But if you would like to check the final solution, go check the code in the wkrzywiec/farm-to-table repository.

And Why Not Kafka?

Before we wrap up, let’s dive into one more very polarizing issue: Is Kafka a good candidate for an event store?

At the first glance it is and there are lots of people that confirm that. But let’s check the requirements that were mentioned earlier, which an event store needs to fulfill, to figure it out on our own:

• ✅ Event store should allow storing events — By default, Kafka stores messages for 2 weeks, but it is possible to extend the retention time infinitely.
• ✅ Events stored in the event store can’t be modified — Every message stored in Kafka can’t be modified.
• ✅ All events in a stream are ordered — Kafka has topic partitions and ensures that all messages within a given partition (in our case, it could be a streamId) are ordered.
• ✅ / ❌ Events must be added to the end of a stream — In a highly concurrent system where multiple producers write to the same topic partition, it is not possible to control the order of messages within it or to disallow incorrect ones. There is a proposal KAFKA-2260 to be able to provide an offset during the write, but as of today, it is still open.
• ✅ / ❌ Events can be fetched up to a certain point in time — There is no built-in functionality, but there is a way to achieve it.
• ❌ Events are globally ordered — Kafka guarantees the order of messages only within a single partition. Events stored within a single or multiple topics would not be ordered.
• ✅ Notify subscribers about changes in a stream — Kafka allows subscribing to its topics and notifies consumers about adding new messages.
• ✅ Store bi-temporal events

Most of the points are green, so should we go ahead with Kafka? Not so fast.

Depending on the situation, a lack of one of these capabilities may be a showstopper. The biggest, in my eyes, is the fact that Kafka does not support and does not have a way to introduce optimistic locking. This means that in the case of concurrent writes, there will be no mechanism to protect one of these writes. For instance, let’s say that we have a delivery that is canceled and updated in 2 separate threads. With Kafka, we could end up with a situation where the update is written to the stream after the delivery is canceled (which, from a business point of view, should never happen).

Also, the lack of global ordering may be a blocker for those systems that need to merge events from multiple streams.

Therefore, in my opinion, for small systems, it may be a good candidate to consider, but there are cases where it’s better to keep it away. But still, Kafka is a great tool designed to move data from one place to another, and for that (not storage), it is the best solution.

I hope that after reading this article, you know more about event stores and how they can be implemented in a Java application using a PostgreSQL database. I encourage you to try it yourself, perhaps with a slightly different approach, to get the full experience of learning all its bits and pieces.

If you would like to check the entire code of my project, visit the GitHub repository: wkrzywiec/farm-to-table.

And if you would like to learn more about event stores and event sourcing, check the links below.

References

• Building an Event Storage
• Let's build event store in one hour! - Event-Driven.io
• Implementing event sourcing using a relational database | SoftwareMill
• Fixing the past and dealing with the future using bi-temporal EventSourcing
• Event Streaming is not Event Sourcing! - Event-Driven.io

Originally published at https://wkrzywiec.is-a.dev on July 15, 2024.

What if I told you that there is a different way of storing business entities other than as entries in relational database tables, JSONs in document key-value stores, or graphs in graph databases? Of course, people can invent anything, but the key question is: will it be usable?

And the answer is yes-there is another way that is not just a different approach to the same problem but also adds value to traditional methods of storing entities. This method is called event sourcing and it is gaining momentum nowadays.

This is my first article in the “Is event sourcing hard?” series, where I’ll take a practical approach and try to implement key concepts of event sourcing in Java. In each article, I will tackle different aspects of it. For example, in this one, I’m focusing on explaining what event sourcing is and showing how it could look in code. In the next ones, I’ll focus on adding additional elements connected with event sourcing, like an event store or an outbox. Moreover, I’ll explore more advanced concepts such as how to version events, create snapshot events, and handle bi-temporal events. But that’s not everything I have planned for this series!

My article is, of course, not the first on this topic. In fact, if you google it, you will find a plethora of great resources far more advanced and in-depth than this one. For instance, a great blog by Oskar Dudycz — event-driven.io is an excellent source of information on event sourcing.

With my series, I’m not trying to cover this topic exhaustively. It’s just my take on it, how I understand it, and a bit of playing around with its concepts.

What is event souring?

In essence, the concept is very simple. Instead of storing information about our models as database entries, we store facts about them. Rather than storing properties of an entity, we store all events that pertain to it.

To visualize this, let’s consider an example. Suppose we want to model a delivery service that provides food to homes. Here are some business events that can occur:

Now, the idea is to store these events in an event store. When a new action needs to be performed on a specific delivery (e.g., it needs to be canceled), the application first needs to load all events and then process them to rebuild the current state of the delivery:

There are multiple ways in which this could be achieved in the code. Here’s an example of the Delivery class, which includes a static factory method from(List<Event> events). This method accepts a list of events and produces the domain object:

@Getter  
@EqualsAndHashCode  
@ToString  
public class Delivery {  
    private String orderId;  
    private String customerId;  
    private String farmId;  
    private String deliveryManId;  
    private DeliveryStatus status;  
    private String address;  
    private List<Item> items;  
    private BigDecimal deliveryCharge = new BigDecimal(0);  
    private BigDecimal tip = new BigDecimal(0);  
    private BigDecimal total = new BigDecimal(0);  
    private Map<String, String> metadata = new HashMap<>();

    private Delivery() {};

    public static Delivery from(List<Event> events) {
        Delivery delivery = null;
        
        for (Event event: events) {
            switch (event.body()) {
                case DeliveryCreated created -> {
                    delivery = new Delivery();

                    delivery.orderId = created.orderId();
                    delivery.customerId = created.customerId();
                    delivery.farmId = created.farmId();
                    delivery.status = DeliveryStatus.CREATED;
                    delivery.address = created.address();
                    delivery.items = mapItems(created.items());
                    delivery.deliveryCharge = created.deliveryCharge();
                    delivery.total = created.total();

                    Map<String, String> metadata = new HashMap<>();
                    metadata.put("creationTimestamp", event.header().createdAt().toString());
                    delivery.metadata = metadata;
                }

                case DeliveryManAssigned deliveryManAssigned -> {
                    delivery.deliveryManId = deliveryManAssigned.deliveryManId();
                }

                case FoodWasPickedUp foodWasPickedUp -> {
                    var metadata = delivery.getMetadata();
                    metadata.put("foodPickedUpTimestamp", event.header().createdAt().toString());

                    delivery.metadata = metadata;
                    delivery.status = DeliveryStatus.FOOD_PICKED;
                }

                case FoodDelivered foodDelivered -> {
                    var metadata = delivery.getMetadata();
                    metadata.put("foodDeliveredTimestamp", event.header().createdAt().toString());

                    delivery.metadata = metadata;
                    delivery.status = DeliveryStatus.FOOD_DELIVERED;
                }

                case TipAddedToDelivery tipAddedToDelivery -> {
                    delivery.tip = tipAddedToDelivery.tip();
                    delivery.total = tipAddedToDelivery.total();
                }

                default -> throw new IllegalStateException("Failed to replay events to build delivery object. Unhandled events: " + event.body().getClass());
            }
        }
        return delivery;
    }

    private static List<Item> mapItems(List<io.wkrzywiec.fooddelivery.delivery.domain.incoming.Item> items) {
        return items.stream().map(dto -> Item.builder()
                .name(dto.name())
                .amount(dto.amount())
                .pricePerItem(dto.pricePerItem())
                .build()).toList();
    }
}

The idea is quite simple. The factory method iterates through the entire list of events (it’s important that events are sorted by their occurrence) and creates a new object from the information stored in an event and/or the previous state of the delivery.

To understand this better, let’s take a closer look. The first step of the method is to define a variable for the resulting Delivery, which initially is set to null. Then, the iteration through events begins, starting with the DeliveryCreated event that holds some data:

public record DeliveryCreated (
  String orderId, String customerId,
  String farmId, String address, List<Item> items,
  BigDecimal deliveryCharge, BigDecimal total ) { }

Based on this event, a private constructor is used to create an empty Delivery object and then set all the initial fields:

public static Delivery from(List<Event> events) {
    Delivery delivery = null
    
    for (Event event: events) {
        switch (event.body()) {

            case DeliveryCreated created -> {
                delivery = new Delivery();

                delivery.orderId = created.orderId();
                delivery.customerId = created.customerId();
                delivery.farmId = created.farmId();
                delivery.status = DeliveryStatus.CREATED;
                delivery.address = created.address();
                delivery.items = mapItems(created.items());
                delivery.deliveryCharge = created.deliveryCharge();
                delivery.total = created.total();

                Map<String, String> metadata = new HashMap<>();
                metadata.put("creationTimestamp", event.header().createdAt().toString());
                delivery.metadata = metadata;
            }
        }
    }
    return delivery;
}

All subsequent events will mutate the Delivery object, but in a very limited way. For instance, the FoodWasPickedUp event contains only two pieces of information: orderId and what happened from a business perspective, which is represented by the class name:

public record FoodWasPickedUp(String orderId) implements DomainEventBody {}

Going back to the event replaying loop — if an event is of the FoodWasPickedUp type, the delivery status is changed to FOOD_PICKED and the metadata is updated:

    public static Delivery from(List<Event> events) {
        Delivery delivery = // existing object recreated from previous events
        
        for (Event event: events) {
            switch (event.body()) {

                case FoodWasPickedUp foodWasPickedUp -> {
                    var metadata = delivery.getMetadata();
                    metadata.put("foodPickedUpTimestamp", event.header().createdAt().toString());

                    delivery.metadata = metadata;
                    delivery.status = DeliveryStatus.FOOD_PICKED;
                }
            }
        }
        return delivery;
    }

The same applies to all other events, which either change the status of a delivery or one of its properties. For instance, adding a tip to a delivery results in changing the tip value and the overall cost of the delivery:

public static Delivery from(List<Event> events) {
    Delivery delivery = // existing object recreated from previous events
    
    for (Event event: events) {
        switch (event.body()) {

            case TipAddedToDelivery tipAddedToDelivery -> {
                delivery.tip = tipAddedToDelivery.tip();
                delivery.total = tipAddedToDelivery.total();
            }
        }
    }
    return delivery;
}

And that’s it. Essentially, this is how the current state of an entity can be rebuilt from events.

Are there any other ways to achieve this? Certainly. For instance, the entire method could be extracted from a domain object and moved to a separate factory class:

public class DeliveryFactory {

    public static Delivery from(List<Message> events) {
        Delivery delivery = null;
        for (Message event: events) {
            switch (event.body()) {
                case DeliveryCreated created -> {
                    delivery = new Delivery();

                    delivery.setOrderId(created.orderId());
                    delivery.setCustomerId(created.customerId());
                    delivery.setFarmId(created.farmId());
                    delivery.setStatus(DeliveryStatus.CREATED);
                    delivery.setAddress(created.address());
                    delivery.setItems(mapItems(created.items()));
                    delivery.setDeliveryCharge(created.deliveryCharge());
                    delivery.setTotal(created.total());

                    Map<String, String> metadata = new HashMap<>();
                    metadata.put("creationTimestamp", event.header().createdAt().toString());
                    delivery.setMetadata(metadata);
                }

                case DeliveryManAssigned deliveryManAssigned -> {
                    delivery.setDeliveryManId(deliveryManAssigned.deliveryManId());
                }

                // all other cases

                default -> throw new IllegalStateException("Failed to replay events to build delivery object. Unhandled events: " + event.body().getClass());
            }
        }
        return delivery;
    }
}

The problem with this approach is that it requires creating setters for each property of an object, which opens up the object to be modified in other parts of the code. If we want our domain object to adhere to Domain Driven Design principles, rather than being a simple Data Transfer Object, this can’t happen. Modification of any property of an object must go through special business methods in Delivery that protect business rules. Methods such as:

public class Delivery {  

    public void assignDeliveryMan(String deliveryManId) {
        if (this.deliveryManId != null) {
            throw new DeliveryException(format("Failed to assign delivery man to an '%s' order. There is already a delivery man assigned with an orderId %s", orderId, this.deliveryManId));
        }

        if (List.of(DeliveryStatus.CANCELED, DeliveryStatus.FOOD_PICKED, DeliveryStatus.FOOD_DELIVERED).contains(status)) {
            throw new DeliveryException(format("Failed to assign a delivery man to an '%s' order. It's not possible do it for a delivery with '%s' status", orderId, status));
        }

        this.deliveryManId = deliveryManId;
    }
}

Alright, so now that you know how to implement event sourcing and have an idea of what the implementation could look like, it might seem like the logical next step to select the approach you like the most, code it, and push it to production, right? I mean, with so many software engineering gurus advocating for it, what possibly could go wrong?

When not to use it?

While event sourcing may sound great, there are certain aspects of it that need to be carefully considered.

For example, a large number of events for a single business entity can lead to performance issues. Replaying thousands of events to obtain the current state of an entity may take a considerable amount of time, even if each event processing method is relatively fast. To address this problem, there are a couple of approaches, such as creating snapshot events (with the full state of an entity) and replaying events starting from them. Another approach involves creating a projection of the current state of an entity, where the current state is stored in a database alongside events. However, it’s important to note that projections should be disposable because events should remain the source of truth.

Additionally, one reason not to use event sourcing is its limited adoption. Not every developer has experience with it, and in situations where the business is pushing for the addition of new capabilities to a system, it may be safer to rely on familiar tools and approaches.

Furthermore, like every tool, event sourcing is not a one-size-fits-all solution. While it provides valuable information for analytics and tracking system activity, it’s not always necessary. For some domains, a traditional approach may suffice.

Lastly, implementing event sourcing requires specific tools to function properly. While the database layer may seem simple in theory, it needs to meet certain criteria, such as supporting concurrent writes and maintaining an append-only storage model. This introduces various challenges, such as handling different versions of the same events, compensating for incorrect events, and managing the removal of personal information (the right to be forgotten). These complexities may deter some from adopting event sourcing.

And when to use it?

So far, I’ve covered what event sourcing is and discussed a few reasons why you might hesitate to use it. But let’s turn things around because there are plenty of compelling reasons to embrace it.

One major advantage of event sourcing is its ability to store the time of a business event. This data, often overlooked in traditional approaches, can be invaluable for advanced analytics. Events not only drive the processing part of a system but also serve as a goldmine for analytics. For instance, they can answer important business questions like how long a customer waited for a delivery or which step of an order process took the most time. Such insights can lead to better user experiences and informed decision-making.

Moreover, events can be a game-changer in highly regulated industries such as healthcare and finance. They allow the construction of a comprehensive log of actions, enabling auditors to identify malicious activities within the system. However, it’s essential to exercise caution with this capability, as events should strike a balance between granularity and providing sufficient context for auditors.

Keeping a log of events also benefits operational teams by facilitating error detection and resolution. When an issue arises, such as a broken order, a quick review of the events log can often pinpoint the actions that led to the problem. Furthermore, teams can easily revert malicious events to restore an entity to its previous state.

From a development perspective, event sourcing offers flexibility and agility. Storing only facts about entities eliminates the need to adhere strictly to a specific business model. If the initial domain model doesn’t align with business needs and requires changes, there’s no need for extensive data migration. Simply adjust how events are replayed to rebuild the entity. Additionally, the same set of events can serve as the foundation for building different projects, such as distinct views of an order from the perspectives of customers and sellers.

Another boon for development teams is the aid event sourcing provides in debugging. A continuous stream of events allows developers to navigate back and forth, enabling what is often referred to as “time travel” debugging. This capability provides invaluable insights for identifying and fixing bugs efficiently.

Example of Event sourcing systems

Event sourcing proves valuable across a wide range of use cases. For instance:

• Account ledger — This is a classic and natural application of event sourcing. An account ledger maintains a log of all incoming and outgoing transactions, forming the basis for calculating the account balance. Some banks even leverage event data to provide insights into spending habits, categorizing expenses into groups such as rent, food, or entertainment. All of it are different projections and are based on the same data — transaction events.
• Hospital inventory — Hospitals rely on meticulous inventory management to ensure patients receive correct medication dosages at the right times. Additionally, maintaining control over drug supplies is crucial for regulatory compliance and legal purposes. event sourcing facilitates this by recording events of medical consumption, detailing which patient took which medication, when, and by whom it was dispensed.
• Ordering system — Event sourcing finds extensive application in e-commerce ordering systems. By logging events related to order processing and fulfillment, businesses can not only reconstruct the order’s state but also derive insights for optimizing the ordering and delivery processes. Essentially, any system operating as a state machine stands to benefit from event sourcing.

Event sourcing in application

Now that we understand event sourcing, let’s delve into its implementation in application code. The basic flow for each use case can be straightforwardly encapsulated within a Service class responsible for handling it. The essence of each method's flow can be summarized into a few key steps:

1. Accept incoming arguments, typically in the form of a command.
2. Retrieve events for an entity from an event store.
3. Replay events to rebuild the entity’s state.
4. Invoke a business method on the entity.
5. Create the resulting event.
6. Store the resulting event.

This general approach can be translated into code as follows:

class Service {

    private final EventStore eventStore;

    public void handle(Command command) {

        List<Event> storedEvents = retrieveEvents(command.entityId)
        Entity entity = Entity.from(storedEvents);

        entity.handle(command.newState);

        Event resultingEvent = new NewEntityStateEvent(entity.id);
        eventStore.store(resultingEvent);
    }

    private List<Event> retrieveEvents(String streamId) {
        var storedEvents = eventStore.getEventsFor(streamId);
        if (storedEvents.size() == 0) {
            throw new EventStoreException("There are no events for stream: ", streamId);
        }
        return storedEvents;
    }
}

Let’s consider a more concrete example of a DeliveryService responsible for handling a PickUpFood command:

class DeliveryService {

    private final EventStore eventStore;

    public void handle(PickUpFood pickUpFood) {

        List<Event> storedEvents = retrieveEvents(pickUpFood.orderId())
        var delivery = Delivery.from(storedEvents);

        delivery.pickUpFood();

        Event resultingEvent = new FoodWasPickedUp(delivery.orderId());
        eventStore.store(resultingEvent);
    }

    private List<Event> retrieveEvents(String orderId) {
        var storedEvents = eventStore.getEventsFor(orderId);
        if (storedEvents.size() == 0) {
            throw new EventStoreException("There are no events for order: ", orderId);
        }
        return storedEvents;
    }
}

If we aim to ensure that every invocation of a business method on a domain object results in the generation of an event, we can shift the responsibility of event generation to the domain object itself:

class Delivery {

    private String orderId;

    public Event pickUpFood() {
        if (status != DeliveryStatus.FOOD_READY) {
            throw new DeliveryException(format("Failed to set food as picked up for an '%s' order. It's not possible do it for a delivery with '%s' status", orderId, status));
        }
        this.status = DeliveryStatus.FOOD_PICKED;
        return new FoodWasPickedUp(orderId);
    }
}

As a result, the code in the DeliveryService can be simplified:

class DeliveryService {

    public void handle(PickUpFood pickUpFood) {

        List<Event> storedEvents = retrieveEvents(pickUpFood.orderId())
        var delivery = Delivery.from(storedEvents);

        Event resultingEvent = delivery.pickUpFood();

        eventStore.store(resultingEvent);
    }
}

In my opinion, this approach is far superior. By moving event creation to the domain object, which encapsulates business logic, we ensure that every action taken results in the generation of an event-a factual statement of what just occurred. If event creation were to remain in the service layer, there’s a risk of forgetting to create the corresponding event. Additionally, there are situations where event creation requires information from private fields within the domain object, which may not be accessible to the service.

Another approach, similar to the previous one, involves temporarily storing resulting events within the domain object itself. Again, the events would be created within the domain object, but instead of being returned after the method is invoked, they would be added to a list of events maintained by the domain object:

class Delivery {

    private String orderId;
    List<Event> changes = new ArrayList<Event>(); 

    public void pickUpFood() {
        if (status != DeliveryStatus.FOOD_READY) {
            throw new DeliveryException(format("Failed to set food as picked up for an '%s' order. It's not possible do it for a delivery with '%s' status", orderId, status));
        }
        this.status = DeliveryStatus.FOOD_PICKED;
        changes.add(new FoodWasPickedUp(orderId));
    }

    public List<Event> uncommittedChanges() {
        return changes;
    }
}

In that case, the flow of the handling method in the DeliveryService might resemble the following:

class DeliveryService {

    public void handle(PickUpFood pickUpFood) {

        List<Event> storedEvents = retrieveEvents(pickUpFood.orderId())
        var delivery = Delivery.from(storedEvents);

        delivery.pickUpFood();

        eventStore.store(delivery.uncommittedChanges());
    }
}

This flow can prove beneficial, especially when multiple business methods of the domain object are invoked within a single service method. By deferring the collection of events until the end of the method, we streamline the process and commit them in one go.

Now, let’s introduce the last component: the EventStore. Since this article revolves around event sourcing, here's a simple definition of the EventStore interface:

public interface EventStore {

    List<Event> getEventsFor(String streamId);
    
    void store(Event event);
    
    default void store(List<Event> events) {
        events.forEach(this::store);
    }
}

And a simple in-memory implementation:

public class InMemoryEventStore implements EventStore {

    Map<String, List<Event>> store = new ConcurrentHashMap<>();

    @Override
    public void store(Event event) {
        var stream = store.getOrDefault(event.orderId(), new ArrayList<>());
        stream.add(event);
        store.put(event.orderId(), stream);
    }

    @Override
    public List<Event> getEventsFor(String streamId) {
        return store.getOrDefault(streamId, List.of());
    }
}

In real-world scenarios, the implementation of an event store can vary widely. Events may be persisted in relational or non-relational databases, each offering its own set of advantages and trade-offs. In one of my upcoming posts, I’ll delve into the implementation of an event store using PostgreSQL.

Summary

Event sourcing is a powerful technique that not only preserves information about domain objects but also records the time of each action that led to a given state. This additional dimension of information can be incredibly valuable for advanced analytics, improved error handling, and facilitating adjustments to the business model.

I hope this article has provided you with some insights into event sourcing. For further information, check out the References section. Additionally, if you’re interested in the code mentioned in this post, you can find it in my repository on GitHub.

GitHub - wkrzywiec/farm-to-table

References

• How to get the current entity state from events? - Event-Driven.io
• Beginner's Guide to Event Sourcing | Event Store
• Event Sourcing
• Never Lose Data Again - Event Sourcing to the Rescue! - Event-Driven.io
• EventSourcing.JVM/workshops/introduction-to-event-sourcing at main · oskardudycz/EventSourcing.JVM
• GitHub - eugene-khyst/postgresql-event-sourcing: A reference implementation of an event-sourced system that uses PostgreSQL as an event store built with Spring Boot. Fork the repository and use it as a template for your projects. Or clone the repository and run end-to-end tests to see how everything works together.

Originally published at https://wkrzywiec.is-a.dev on June 10, 2024.

Not enough memory

Learning a programming language can be a challenging journey. First, you need to grasp the basic syntax to accomplish standard tasks. The next step involves learning one or two popular frameworks within a given ecosystem to enhance your visibility in the job market. However, once you secure that dream job, you quickly realize that having a solid foundation in a programming language and popular frameworks is not sufficient. Elements like best practices, design patterns, and system design play crucial roles in becoming a successful expert in software development.

Finally, a day arrives when you have mastered all this knowledge (or at least you think you have), and you receive a call from the operations team stating that your application is experiencing overwhelming success but struggles to handle the large influx of user requests. None of the previous steps prepared you for the skyrocketing CPU usage or memory issues. Panic sets in as you desperately try to recall why these components are essential and how they relate to your state-of-the-art application.

Look into inside

Does it sound familiar?

How could you prevent this from happening? The first step is to come to the realization that you don’t know everything about your piece of software. Yes, you have coded it, but do you know how it behaves in the wild production environment? If you received a similar call, you probably don’t know. Therefore, the next step for you would be to initiate app monitoring to observe trends and receive alerts when key metrics approach dangerous thresholds. Additionally, understanding how the app responds under various types of pressure is crucial.

Numerous tools can be employed for this purpose. Some are built into the JDK, such as Java Flight Recorder with Java Mission Control, while others can be downloaded, like JProfiler, and some are cloud platforms, such as Datadog. It’s easy to feel overwhelmed initially, as each tool provides different types of information, often at a very low level.

So, where do you begin with monitoring app performance? Which tool should you choose first?

Grafana dashboard

I would recommend starting with Prometheus and Grafana. Both tools are open source and non-invasive, meaning there’s no need to add any special tool/agent that may affect overall performance. They only require adding metrics libraries to a project. The setup is relatively straightforward, as I’ve outlined in a previous article — How to set up monitoring tools for JVM.

The mentioned post also covers how to import one of the most popular JVM dashboards — JVM (Micrometer) — which will be the subject of this post. It will walk you through key metrics and explain when and why they’re important.

The dashboard consists of sections that will be described in the following parts, which I’ve grouped and changed the order a little bit.

JVM Memory

If you ever wonder what’s the first thing that you must keep eye on when monitoring Java application in most of the case it would be the memory usage. This is a vital information about the health of an app, since JVM needs it to run it.

Therefore it is not a suprise thet this is a first information that is provided by the dashboard — the percentage of how much heap and non-heap memory is fillied up (along with data about how long the app is running).

In most applications, these two metrics are the ones we should keep an eye on. They have a significant impact on its performance and can quickly explain why the app is running slow or even crashing.

Why these two metrics are that important? First of all we need to know that Java is object-oriented language. It means that every program is a set of objects which interact with each other. They have methods that are invoked by other objects and they also hold data. And in the real applications number of objects may be very large. And that’s why it is worth to monitor their number and size with the first indicator. The latter is accountable for measuring memory use of non-object related things, which in some cases are also worth to monitor.

Moving on, let’s examine the charts above, which illustrate how memory usage has changed over time. The charts depict the actual memory usage ( used) at a given moment, the maximum available ( max), and the memory guaranteed to be available for the JVM ( committed).

What can we read from the Heap memory chart? Certainly, the used memory shouldn't reach the max value. A dangerous situation arises when the increasing amount of memory in the Heap area indicates that many objects are created and never destroyed (collected by the Garbage Collector, which will be mentioned later). If this happens, the app will start to act slower and eventually crash, throwing the java.lang.OutOfMemoryError.

In addition to Heap data, the JVM has a second data area called Non-Heap (or Off-Heap or Native Memory). It stores various data, which can be divided into the following spaces:

• Metaspace (aka Method Area): It holds class-related information about class code, their fields, and methods after being loaded by the classloader (more about it in the article). It also contains information about constants (in the Constant Pool).
• Code Cache: This area stores optimized native code produced by the Just-In-Time compiler.
• Program Counters: These point to the address of the currently executed instruction (each thread has its own Program Counter).
• Stacks: The Last-In-First-Out (LIFO) stack of frames for each executed method. It contains primitive variables and references to the objects on the heap.

While it is typically the heap that may be too small, we need to understand that memory problems with non-heap data could also crash an app. Too many classes loaded into Metaspace may cause a java.lang.OutOfMemoryError, and too many frames in the Stack may result in a java.lang.StackOverflowError being thrown.

JVM Memory Pools (Heap)

Let’s now delve into Heap usage. But first, we need to understand how the Heap is structured because it is not a monolith. The Heap is divided into various spaces, the number of which depends on the type of Garbage Collector used. Above, we can see three charts for Eden Space, Survivor Space, and Old Gen, which are used by the G1 Garbage Collector.

But why is the Heap split into spaces? It comes from something called the weak generational hypothesis. This hypothesis presumes that the lifespan of objects varies. Most of them are very short-lived, used only once, and then the memory they occupy can be freed. Others, usually a smaller fraction, are the ones that are constantly used and cannot be destroyed. Due to this fact, the Garbage Collector (GC) splits the heap into generations (spaces), where short-lived objects reside in young generations ( Eden Space and Survivor Space), and long-lived objects in an old generation ( Old Gen).

Having the heap split into generations allows the GC to avoid performing one long garbage collection on the entire heap. Instead, the GC performs it only on a subset at a time. It also allows for different collection frequencies in each generation; the Eden Space is cleaned more often than the Old Gen because objects located there are needed only for a short period.

The above screenshot confirms this. We can see that the plot for used memory in Eden Space is formed in a characteristic jigsaw pattern. It quickly fills up with new objects (expressed as a rising plot on the chart), and once it reaches a certain threshold, it is emptied (causing the plot to go down).

On the contrary, the Survivor Space and Old Gen are much more stable. There are some changes in them, but they are very subtle.

These three charts give us vital information about the content of the heap. We can read how often new objects are created and removed. If the cleanup for them is done very often and the old objects are kept at a relatively same level, it is tempting to increase the ratio of young to old generation size. The larger the young generation is, the less often cleanup will be done. But on the other hand, it will be bigger, so the cleanup may take a little bit longer. The same goes for old generations; the smaller it gets, the more frequently it is cleaned. So we must keep a balance of it.

To adjust it, use -XX:NewRatio=N (which is a ratio of young and old generations, e.g., N=2 means that the young generation is twice as big as the old one) or -XX:NewSize=N (which is the initial size of the young generation - all remaining will be assigned to the old generation).

What if the app is consuming too much memory?

The heap is one of the most crucial memory areas for JVM and typically requires a substantial amount of space. If it approaches the maximum threshold, the simplest solution is to increase heaps’ maximum size. A larger heap ensures that all necessary objects can fit into it. This can be accomplished by specifying -Xmx followed by the size in megabytes allocated for the heap. For example, -Xmx1024m will allocate 1024 MB for the heap. It's easy and straightforward.

While adjusting the maximum size of the heap may provide a temporary solution, in many cases, it merely masks the symptoms rather than addressing the root cause of the problem. An increasing number of objects could indicate a memory leak within the application, requiring identification and resolution in the code.

Another possibility is that an excessive amount of data is being loaded into memory, either due to an inefficient data structure or an unexpectedly large volume of it. In the former case, optimizing how data is stored in the JVM is the solution, while in the latter, considerations such as increasing the heap size or chunking the large amount of data into smaller pieces.

But how to spot if we have inefficient data structure? There are couple of signs and one of them is using objects with lots of fields (e.g. entities) from which only a small subset is used. Limiting the number of fields per task can be beneficial here. Also we need to remember that even if fields won’t be initialized they also consume memory!

Additionally, strive to avoid using objects instead of primitives. Many instances exist where objects contain fields like Boolean instead of plain boolean. While this may seem like a minor change, it can significantly impacts memory usage. Similarly, treating String objects as containers for numbers or booleans variables consumes more memory than necessary.

Another optimization to reduce memory footprint involves refraining from initializing String variables with identical values repeatedly.

In certain situations, it can be highly beneficial to avoid creating immutable objects. This is particularly true when new objects need to be created as copies of others. Of course, I’m not discouraging the use of immutable objects altogether, as they prove invaluable in many scenarios. However, for objects created and used within a small method once or twice, it is sometimes more advantageous to make them mutable.

These suggestions only scratch the surface of addressing memory-related issues. Numerous resources are available to guide you in further optimizing memory usage.

Garbage Collection

The previous section already mentioned the Garbage Collector, which we will now examine more closely. It is a key component of the JVM because, in contrast to languages like C++, we, as developers, don’t need to worry about freeing memory once an object is no longer needed. The role of the GC is to decide which objects should be destroyed and which should be preserved.

All new objects are initially allocated in the Eden Space. However, during the next garbage collection, they are promoted to one of the Survivor Spaces if they are still in use. Subsequently, during subsequent garbage collections, they are again checked to determine whether they should be promoted to the next subspace of the Survivor Space or even to the Old Gen.

The first plot in the above screenshot, Collections, tells us how often a certain type of garbage collection occurs in a second. We can see that only one minor garbage collection (cleaning only the young generation of the heap) is happening, but it also shows a major garbage collection (cleaning objects from the old generation). By reading this chart, we can figure out how often each type of collection is occurring, and if it’s too frequent, it may suggest that one of the spaces is filling up too quickly with a large number of objects.

However, even if the number of collection occurrences is relatively high, it’s not always a cause for concern. If each one of them takes a very short time and does not significantly affect the overall application performance, we have nothing to worry about. To ensure that this is the case, we can look into the second plot, Pause Durations, which provides information about the average time of each garbage collection.

The last plot, Allocated/Promoted, shows us how much memory was used to allocate objects in the young generation and how much of it was promoted to older generations. This information is really helpful when we want to understand the memory load that GC needs to deal with in each cycle.

After identifying an issue in one of these graphs, we can take two approaches: tune JVM memory limits and GC behavior or select a different GC. Depending on the Java version and distribution (e.g., Oracle’s HotSpot, Amazon Corretto, Azul Zulu, or Eclipse Adoptium), the list of available GCs is different, but most of them can be categorized into:

• Generational vs. non-generational — Some GCs may not split a heap into generations but treat it as a single memory space. However, usually it is not the case (though the approach varies among distributions).
• Stop-the-world vs. concurrent — GCs in the first group pauses the entire application for cleanup, while the latter performs cleanup while the application is still running (or, to be precise, they briefly stop the application).
• Single-thread vs. multi-thread — The first group operates using only one thread, while the latter utilizes at least two.
• Incremental vs. monolithic — This is specific to stop-the-world GCs and indicates whether garbage collection is done only for a subset of the generation or until it’s complete.

Here is a list of available GCs in the Amazon Corretto distribution of OpenJDK, one of the most popular distributions according to the 2023 State of the Java Ecosystem for 21 version of Java:

The above table lists all available GCs along with their basic properties. Even though some of them may seem better than others, it is crucial not to overlook trade-offs. For instance, the short garbage collection pauses of Shenandoah and ZGC come with the cost of higher CPU usage. Therefore, before selecting one, make sure you understand the price you would need to pay.

If you are looking for more information about each GC, check the References section, where I’ve included links to the official documentation of each one of them.

JVM Memory Pools (Non-Heap)

Apart from objects, JVM needs memory to store other things. This is represented by the next graphs from JVM Memory Pools (Non-Heap) in the discussed dashboard.

• Metaspace (a.k.a. Method Area) — This is the place where metadata and definitions of every class are stored (in simpler terms, every .class file in the runtime representation). Additionally, Metaspace holds method counters used by the JIT compiler and a constant pool (a data structure to store information about references and constants associated with a class, such as static final fields). By default, there are no limits on how big it can get, but it can be tuned using the -XX:MaxMetaspaceSize flag. If not provided, it may occur that the JVM will consume too much memory and slow down the machine on which the application is running.
• Compressed Class Space — This is a subset of Metaspace with classes metadata.

The remaining three graphs show segments of the code cache. It is used to store code optimized by the JIT (Just-In-Time) compiler. This compiler plays a vital role in JVM by identifying and optimizing places in the code that are executed very often. These hot spots (hence the name HotSpot for Oracle’s JDK distribution) are identified and recompiled from bytecode to machine (native) code. Along with other optimizations, such as method inlining or dead code elimination, it makes those parts of the code ultra-fast to execute, making Java applications robust.

• CodeHeap ‘profiled nmethods’ — This segment holds lightly optimized methods.
• CodeHeap ‘non-profiled nmethods’ — This segment contains fully optimized methods.
• CodeHeap ‘non-nmethods’ — The smallest segment that contains compiler buffers and the bytecode interpreter.

Code cache is split into three segments:

This provides all the insights that the discussed dashboard gives us about the non-heap area. While these are valuable pieces of information, it’s crucial to be aware that it’s not everything. All mentioned data areas, like the heap and parts of the non-heap, are shared among all application threads. However, some areas are assigned to specific threads, such as the stack or program counter, and these are also vital parts of the JVM.

Classloading

Besides information about the amount of memory occupied by loaded classes, the dashboard also provides details on how many classes were loaded into memory and when these loadings occurred. This information can be gleaned from the first two charts:

Most of the class loading occurs during the application startup, especially for solutions utilizing Inversion of Control. This is when most objects need to be initialized, preceded by the loading of their definitions into the JVM process. However, not all classes are loaded immediately. Some class definitions are loaded lazily during runtime.

JVM Misc

Next section available in the dashboard govers various things worth monitoring.

The name of the first chart, CPU Usage, suggests that it shows how the CPU is utilized. It presents three series: system (total CPU usage of the host), process (CPU usage for all JVM processes), and process-15m, which is an average of the latter gauge from the last 15 minutes.

The next graph, Load, provides more insight into the number of processes (load) running and queued for the CPU on average in 1 minute. If this indicator is below the number of available CPUs (also available on this chart), it means that there are no waiting processes due to occupied CPU, which is good news. Otherwise, it would indicate that the CPU can’t keep up with executing all processes.

The next two graphs provide information about JVM threads. The first one, Threads, displays the number of daemon threads, which are the opposite of user threads. Essentially, these two types of threads in the JVM are almost the same. The first difference is that when the program is stopped, first the user threads are halted, and then the daemon threads are exited. Only after that is the JVM considered as non-running. The second difference is that daemon threads serve the user thread. Typically, they are assigned to handle garbage collection, but they can also be utilized in our code by marking the Thread object with the setDaemon(true) methods.

The second graph, Thread States, presents the distribution of threads divided by their state.

The next row of charts starts with GC Pressure, indicating the percentage of CPU used for garbage collection. It’s essential to monitor this metric along with other garbage collection metrics mentioned earlier.

Another noteworthy metric to keep an eye on is the number of logs produced by the app, grouped by level, known as Log Events. This information can be valuable for spotting problems if the number of warn or error level logs is higher than usual, which may not be apparent in other metrics.

The last chart, File Descriptors, displays the number of files opened by the process.

I/O Overview

The last section provides data not directly connected with JVM, but it’s crucial to keep an eye on, especially if it provides a REST API (which is the case for most Java applications nowadays).

The first chart shows how many HTTP requests per second were made. The next one provides the same information, but this time narrowed down to all 5xx responses. A significant increase in 5xx responses may indicate unwanted issues. Also, in the case of high user traffic (as indicated in the first chart), it’s worth monitoring how much memory is consumed by the app, how often garbage collection is performed, or other indicators mentioned earlier. All of these factors may be correlated and affect how fast users receive a response from your service.

The next indicator measures how much time, on average and at maximum, HTTP requests took (excluding 5xx responses). A sudden increase in this metric may indicate that garbage collection pauses are taking longer than usual, which is worth checking in that case.

The last chart provides information about Tomcat’s (or Jetty’s) busy and active threads. It shows how many of them are currently in use, which is crucial for understanding the throughput of an application. The chart also displays information about the configured maximum number of Tomcat’s threads. If the number of currently used threads is very close to this limit, it should serve as a warning that the application may not be able to handle all incoming requests.

Conclusion

And there you have it! 🚀 We’ve journeyed through the entire JVM (Micrometer) Grafana dashboard and soaked in a ton of insights about the JVM and its crucial indicators. 🧠✨ I hope that now the JVM feels more like a friendly companion than a mysterious black box. If you’re hungry for more knowledge, check out the references mentioned below or dive into tools like Java Flight Recorder or profilers. 🛠️ Who knows, maybe they’ll make an appearance on this blog someday!

Until then, why not set up the Grafana dashboard for your project and start unveiling the secrets of how your software dances in the production environment? Happy monitoring! 🎉👩‍💻👨‍💻

References

• Java Performance: The Definitive Guide
• The Java Garbage Collection Mini-Book
• Getting Started with the G1 Garbage Collector
• HotSpot Virtual Machine Garbage Collection Tuning Guide
• HotSpot Virtual Machine Garbage Collection Tuning Guide
• Main
• Main

https://inside.java/2023/11/28/gen-zgc-explainer

• Main
• What is Metaspace?

Originally published at https://wkrzywiec.is-a.dev on January 13, 2024.

Preparing test data is boring and tedious!

Let’s be honest, how familiar does this code look to you?

class DeliveryTest {

  @Test
  void delivery_is_canceled_if_was_just_created() {
    //given
    Food pizza = new Food("Pizza Margherita", 1);
    Food coke = new Food("Coke", 1);
    Delivery delivery = new Delivery(UUID.randomUUID(), List.of(pizza, coke), "Main Street 1, Naples")    
    
    //when
    delivery.changeStatus(CANCELED)    
    
    //then
    assertEquals(CANCELED, delivery.status());
  }
}

Even for a seemingly simple test case, we must create the Delivery class with all its necessary fields. Now, envision if this class were significantly more complex, perhaps requiring an address represented by a separate class (possibly with GPS coordinates); it would swiftly become challenging to manage.

Furthermore, it would complicate the readability and comprehension of the code. Why should we input an address when the sole purpose of the test is to validate the cancellation of a delivery? Our tests should clearly convey their intent and the preconditions required to fulfill them.

One way to tackle these problems is to move all setup steps into a private method so that it is not only encapsulated in an elegant method but also can be reused across other test cases.

class DeliveryTest {

  @Test
  void delivery_is_canceled_if_was_just_created() {
    //given
    Delivery delivery = aNewDelivery();  
    
    //when
    delivery.changeStatus(CANCELED);  
    
    //then
    assertEquals(CANCELED, delivery.status());
  }

  private Delivery aNewDelivery() {
    Food pizza = new Food("Pizza Margherita", 1);
    Food coke = new Food("Coke", 1);
    return new Delivery(UUID.randomUUID(), List.of(pizza, coke), "Main Street 1, Naples");
  }
}

Such approach solves the problem of tedious work for test setup but only partially. What if each test requires a slightly different setup? Arguments can be added to this method to make it more customizable, but in a very large set of tests, the number of arguments may quickly become very big and hard to maintain.

private Delivery aNewDelivery(String address, List<Food> foods) {
  return new Delivery(UUID.randomUUID(), foods, address);
}

Is there a different approach to it? Is there a way to avoid all this setup and at the same time give a flexibility for adjustments of our test data?

So let’s use the magic tool…

Let’s introduce a Test Data Builder object which will be responsible for generating test data.

class DeliveryTest {

  @Test
  void delivery_is_canceled_if_was_just_created() {
    //given
    Delivery delivery = DeliveryTestData.aNewDelivery().entity();  
    
    //when
    delivery.changeStatus(CANCELED);  
    
    //then
    assertEquals(CANCELED, delivery.status());
  }
}

It is somehow easier to read if we compare with a first approach, but it looks a little bit off with this entity() at the end. Without it would look pretty close to the second approach. Where is a gain here then?

Let’s have another test example to see the benefit of using Test Data Builder:

class DeliveryTest {

  @Test
  void delivery_can_not_be_completed_if_was_canceled() {
    //given
    Delivery delivery = aDelivery()
                            .withStatus(CANCELED)
                            .entity();  
    
    //when & then
    assertThrows(
        IllegalStateException.class, 
        () -> delivery.changeStatus(COMPLETED));
  }
}

Here, the Delivery object is instantiated with all default values except for one - its status. The Test Data Builder offers a an easy to use method to override these values. By using it, we can concentrate on the essence of this test - it becomes evident that only the status affects the test's outcome. Other properties, having no such impact, are not explicitly provided here. This results in a cleaner and more comprehensible test.

Furthermore, the way this object is set up closely resembles natural language. If we omit the brackets in the previous example, we obtain a clear statement: a delivery with status canceled. Comparing this to the initial approach with numerous constructor arguments, it becomes evident which is easier to read.

The last benefit and the essential reason why to use the discussed pattern is that it allows to construct objects for various classes. Typically, projects contain a subset of entities, data transfer objects (DTOs), domain classes, messaging classes, and more. While they serve different purposes, they often share common data fields/schemes. The logic for initializing each of them can be encapsulated in a method that takes all required data (whether default or overriden) stored in the Test Data Builder and invokes the class’s constructor to yield a new instance. This is the purpose of the previously mentioned entity() method in the examples - it prepares an entity class for use in the test.

…which is not a magic at all…

Let’s have our hands dirty and and dive into creating the Test Data Builder for delivery objects, as mentioned in the previous section.

The first step is to create a class with fields designed to hold the default values necessary for constructing the Delivery object:

import java.util.UUID;

import static io.delivery.domain.DeliveryStatus.CREATED;
import static io.delivery.domain.FoodTestData.anFood;

class DeliveryTestData {

  private UUID id = UUID.randomUUID();
  private DeliveryStatus status = CREATED;
  private String address = "Main Street 1, Naples";
  private List<FoodTestData> foods = List.of(aFood());
}

It’s up to you how to name this class. Personally, I tend to use the suffix TestData, but other options like Builder work equally well. What's crucial is maintaining consistent naming conventions throughout the entire project.

It’s worth noting that the DeliveryTestData class contains a list of FoodTestData. This represents another builder for classes associated with food, and it is structured as follows:

class FoodTestData {

  private String name = "Pizza Margherita";
  private int amount = 1;
}

The next step is to prohibit the creation of DeliveryTestData using the default constructor and, instead, enable its creation using the aDelivery() static factory method:

class DeliveryTestData {

  private UUID id = UUID.randomUUID();
  private DeliveryStatus status = CREATED;
  private String address = "Main Street 1, Naples";
  private List<FoodTestData> foods = List.of(aFood());  
  
  private DeliveryTestData() {};  
  
  public static DeliveryTestData aDelivery() {
    return new DeliveryTestData();
  }
}

The same thing needs to be done in the second builder class — FoodTestData, but let's skip it here and all other further steps, to keep this post short.

Since all fields are private we need to allow to modify them. Instead of creating setters let’s create with... methods for each one of them:

class DeliveryTestData {

  private UUID id = UUID.randomUUID();
  private DeliveryStatus status = CREATED;
  private String address = "Main Street 1, Naples";
  private List<FoodTestData> foods = List.of(aFood());  
  
  private DeliveryTestData() {};  
  
  public static DeliveryTestData aDelivery() {
    return new DeliveryTestData();
  }

  public DeliveryTestData withId(UUID id) {
    this.id = id;
    return this;
  }

  public DeliveryTestData withStatus(DeliveryStatus status) {
    this.status = status;
    return this;
  }

  public DeliveryTestData withAddress(String address) {
    this.address = address;
    return this;
  }

  public DeliveryTestData withFoods(FoodTestData... foods) {
    this.foods = Arrays.asList(foods);
    return this;
  }

}

Notice that each method returns the instance of modified class (“classic” setter method is always void). This little trick enables chaing with() methods, for example for the test we would like to override two default properties:

class SomeTest {

  void test() {
      
    var delivery = aDelivery()
                     .withStatus(COMPLETED)
                     .withAddress("Main Road 2");
  }
}

Next, crucial, part is to prepare methods that will be responsible for creating target objects. In my case I want to allow to create an entity and DTO (for controller) classes, but you can create more if needed.

class DeliveryTestData {

  private UUID id = UUID.randomUUID();
  private DeliveryStatus status = CREATED;
  private String address = "Main Street 1, Naples";
  private List<FoodTestData> foods = List.of(aFood());  
  
  private DeliveryTestData() {};  
  
  public static DeliveryTestData aDelivery() {
    return new DeliveryTestData();
  }

  public DeliveryEntity entity() {
    DeliveryEntity entity = new DeliveryEntity();
    entity.setId(id);
    entity.setStatus(status);
    entity.setAddress(address);
    entity.setFoods(
      foods.stream()
        .map(FoodTestData::entity)
        .toList()
    );

    return entity;
  }

  public DeliveryDTO dto() {
    return new DeliveryEntity(
        id,
        status,
        address,
        foods.stream()
            .map(FoodTestData::dto)
            .toList()
    );
  }

  //all with...(...) methods
}

The final step is to enable the reading of values from private fields so that they can be used in test assertions. This can be achieved either by creating them manually or generating them using an IDE, as demonstrated here with the Lombok annotation:

import lombok.Getter;

@Getter
class DeliveryTestData {

  //content of the class    
}

…and can be used in Integration tests!

Test Data Builders are excellent for unit tests, but what about integration tests? In the setup phase of an integration test, we often need to insert data into the database. Can we also use these types of classes for that purpose? The answer is — absolutely!

Let’s assume we want to create an integration test for a custom repository method in a Spring Data repository. For instance, we have a method to retrieve the count of deliveries for a specific type of food:

import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;


interface DeliveryRepository extends JpaRepository<Delivery, UUID> {


  @Query(nativeQuery = true, value = """
  SELECT count(*) 
  FROM delivery d 
  LEFT JOIN delivery_food df ON d.id = df.delivery_id
  WHERE df.food_name = ':foodName'
  """)
  long totalDeliveriesCountForFood(@Param("foodName") String foodName);
}

A test can take this form, assuming that within the IntegrationTest class, there is code responsible for launching the database using Testcontainers (exemplary code can be found here). This allows us to skip the setup here:

import org.springframework.beans.factory.annotation.Autowired;

class DeliveryRepositoryIntegrationTest extends IntegrationTest {

  @Autowired
  private DeliveryRepository deliveryRepository;
  private TestRepository testRepository;  
  
  @Before
  void init() {
      testRepository = new TestRepository(deliveryRepository);
  } 

  void two_delivieres_are_returned_if_two_were_made() {
    //given
    String foodName = "Pierogi";  
    DeliveryTestData delivery1 = aDelivery()
      .withFood(
        aFood()
          .withName(foodName)
      );  
    
    DeliveryTestData delivery2 = aDelivery()
      .withFood(
        aFood()
          .withName(foodName)
      );  
    
    testRepository.saveAll(delivery1, delivery2);  
    
    //when
    long result = deliveryRepository.totalDeliveriesCountForFood(foodName);  
    
    //then
    assertEquals(2, result);
  }
}

The test above is very similar to the unit tests written earlier. All the setup phases are consolidated into one place, and it’s straightforward: two deliveries with the same food are created and then saved into a database. There is no need to create any SQL inserts to make this test work. So, what’s the key to this structure of an integration test? The TestRepository class.

As the name suggests, its purpose is to be used during tests to persist the Test Data Builder object in a database. There are several ways to implement it. Here’s one way it could be implemented for a Postgres database using Spring Data’s repository:

import org.springframework.transaction.annotation;

public class TestRepository {

  private final DeliveryRepository deliveryRepository;  
  
  public TestRepository(DeliveryRepository deliveryRepository) {
    this.deliveryRepository = deliveryRepository;
  }

  @Transactional
  public List<Delivery> saveAll(DeliveryTestData... deliveries) {
    return Arrays.stream(deliveries)
      .map(this::save)
      .toList();
  }

  @Transactional
  public Delivery save(DeliveryTestData testData) {
    Delivery entity = testData.entity();
    return deliveryRepository.save(entity);
  }
}

Here, you can see that the TestRepository is a straightforward wrapper for Spring Data's repository. It takes on this role because the object graph (comprising Delivery and a list of Food) is relatively simple. However, when dealing with more complex data structures or intricate ORM mappings (such as bidirectional many-to-one associations), it becomes beneficial to centralize entity preparation and persistence logic within the TestRepository class.

Of course, you’re not limited to the above solution. Instead of using Hibernate, you can implement it using JDBC (or Spring’s JdbcTemplate), jOOQ, or any other tool of your choice. So, should we use it everywhere?

Like any other tool, Test Data Builder should be employed where it fits best. Does it fit every project? Certainly not!

When should you avoid using it?

First and foremost, improved readability comes at the cost of higher maintenance. To use it, you need to create it first. If a data model is extensive (meaning many properties or elements in the object graph), it will take some time to set up Test Data Builder. Moreover, if the data model changes frequently, it will require more effort to keep up with these changes (though the same challenge applies to the classic approach).

Using it for a small project (e.g., serverless functions) might be overkill as well. In such cases, where a project has a small number of use cases, it may be quicker and easier to prepare specific data for each test case. The same holds true if a particular object graph is only used a couple of times. There’s no sense in creating a Test Data Builder and then using it just a few times. It’s better to set up those tests without it.

Lastly, not all projects follow the pattern of fetching data, modifying it, persisting it, and notifying an invoker of this action’s result. Some projects are less typical and may require a special approach.

Aside from these three cases, I encourage you to give Test Data Builder a try in your applications. With a little effort, you can reap the benefits described earlier.

It’s time to wrap up this post with a quick summary of why Test Data Builders are worth using:

• They provide a convenient and efficient way to prepare test data.
• They enhance test readability by introducing notation close to natural language.
• They conceal irrelevant parts of test data preparation, allowing you to focus only on the key aspects important for a particular test.
• They offer a single location for initializing various data transfer objects used in tests.
• Thanks to them, unit and integration tests can closely resemble each other.

Of course, it’s important to remember that for certain projects, this pattern may not be the best choice. Especially for small or rapidly changing projects, the maintenance effort can be significant.

If you’re looking for more examples of Test Data Builders, feel free to check out my project on GitHub — wkrzywiec/farm-to-table.

GitHub - wkrzywiec/farm-to-table

And if you need more sources

How to Create a Test Data Builder | Code With Arho

Originally published at https://wkrzywiec.is-a.dev on October 8, 2023.

This article is part of the “Java Series”, which covers useful Java functions from standard and popular libraries. More posts on that can be found on Medium or on my home page here.

Problem statement

Many times when we work with Java code we end up with a following Plain Old Java Object (POJO):

public record Parent(List<Child> childs) {}

They can represent database entities or data transfer objects (DTOs). In general, they’re used to structure data. Let’s say that we got the list of Parent objects, but we want to operate on a list of all Child that are part of the Parent. How we could extract Child objects from all Parent objects and combine them into a single list? The naive approach would be to use a loop:

List<Child> children = new ArrayList<>();

for (Parent parent: parents) {
    List<Child> bs = parent.childs();
    children.addAll(bs);
}

But it doesn’t look nice and clean. Instead, we could use the Java stream:

List<Child> children = new ArrayList<>();

parents.stream()
    .map(parent -> parent.Childs())
    .forEach(list -> children.addAll(list));

But it has a drawback too. Let’s say that once we get a list of all Child objects we would like to modify them, aggregate, or do calculations on them. Ideally it would be good to perform those action in the same stream. Unfortunately, this is not the case, the forEach() method in the above example is ending the stream processing, which makes it impossible to process Child records in the same stream.

Solution

Luckily Java creators foresaw this problem and introduced a flatMap() function that is part of a java.util.stream.Stream class.

The idea is pretty straightforward. It does two things with every element of a stream:

• maps — transforms one element from a stream into a new stream, so as a result we would get a stream of streams,
• flattens — results of a previous operation are merged into one stream.

To visualize it consider the following situation:

Let’s say that we have a stream of Author objects, that has a method called books() which returns a list of Book objects. And now let's say that we would like to have access to all books written by all authors to make further operations on them. If we would use the map() function within which we would call the books() method we would get a stream of lists of Book objects. This is not what we want to have.

What we would like to have is a stream of Book objects, not a stream of their lists. How we can overcome it? Using flatmap() instead:

As previously we need to invoke the books() method of the Author class to get a list of Books. The only difference is that input needs to be converted into multiple values represented by a Java stream. This is the requirement of the flatMap() method. Whatever operations we do within it needs to return a Stream<T> object.

The same situation can be reflected with a code:

List<List<Book>> listsOfListOfBooks = authors.stream()
    .map(Author::books)
    .toList();

List<Book> listOfBooks = authors.stream()
    .flatMap(author -> author.books().stream())
    .toList();

To translate a list of Book objects the standard Collection.stream() method was used.

When to use it?

Unwrap & operate

The most common case when the flatMap() method might be handy is when one of the stream operations produces a collection of objects and we would like to make further actions on each one of them.

To visualize it let’s go back to the previous example with Book and Author records. Here are their definitions:

public record Book(String title) {}
public record Author(String name, List<Book> books) {}

Now let’s say that we want to create a method that takes a list of Author objects as input and produces the list of all book titles that these authors wrote:

List<String> getAllBookTitles(List<Author> authors) {
return authors.stream()
    .flatMap(author -> author.books().stream())
    .map(Book::title)
    .toList();
}

After converting a list of Author objects into a stream the flatMap() operation is used first. The books() method is invoked to get their list and then it's changed into the stream. The flatMap() is then merging all resulting streams into one so next the title() can be called to get a String representation of a book title. Finally, the results of each element in a stream are collected into the list.

The above method can be written even better. We can split invoking books() and stream() methods into two operations - map() and flatMap() respectively - to get a nice looking code:

List<String> getAllBookTitles(List<Author> authors) {
return authors.stream()
    .map(Author::books)
    .flatMap(List::stream)
    .map(Book::title)
    .toList();
}

Merge lists

Another case when flatMap() can be very useful is when we would like to combine two or more lists (or any other java.util.Collection).

List<String> mergeLists(List<String> left, List<String> right) {
return Stream.of(left, right)
    .flatMap(List::stream)
    .toList();
}

A big plus for this approach is that after flatMap() we don't need to close the stream immediately. Instead, we can apply other operations on every object, like filtering, mapping, aggregating, etc. Which is cleaner and more efficient.

Get value from nested Optional

Apart from Java streams flatMap() method can be invoked on an Optional object. It's used to unwrap an Optional that is nested inside another Optional.

Let’s say that we’ve got the following record:

public record Address(String street, String buildingNo, Optional<String> apartmentNo) {}

Now suppose we would have an Optional<Address> and would like to extract a value of an apartmentNo. The code without flatMap() would look like this:

String extractApartmentNo(Optional<Address> address) {
    if (address.isEmpty()) {
        return "";
    }

    return address.get().apartmentNo().orElse("");
}

The first step is to unwrap the value from the Optional, which might be empty. Only after checking it, we can proceed with unwrapping (and handling empty values) an apartment address.

This approach is ok, but can be done better with flatMap():

String extractApartmentNo(Optional<Address> address) {
return address
        .flatMap(Address::apartmentNo)
        .orElse("");
}

This approach a way nicer than the previous one. Both Optionals — parent and child — are validated whether they hold a null value in a single expression.

Introducing streams into Java made Data processing easier. It brings us a lot of handy operations. flatMap() is one of them which gives us the possibility to merge multiple streams into one or flatten nested streams into one. It's a very common pattern and is used many times in real projects.

Code and tests from this post can be found in my repository:

GitHub - wkrzywiec/java-series: Repo with various Java examples

References

• Part 2: Processing Data with Java SE 8 Streams
• Stream (Java Platform SE 8 )
• Tired of Null Pointer Exceptions? Consider Using Java SE 8's Optional!

Originally published at https://wkrzywiec.is-a.dev on December 12, 2021.

When your application reaches a state when it’s production ready it’s a good time to celebrate. Soon users will start using it, which is great, but it creates a new type of problem. How to monitor how an application behaves under the load of users traffic? How to validate if everything goes smoothly and users are not experiencing any kind of slowness or errors? The answer to these questions would be to use a performance monitoring tools.

With this blog post I’ll show how to set up two opensource monitoring tools available for JVM — Java Flight Recorder with JDK Mission Control and Prometheus with Grafana. It’s a first article out of two. In this one I’ll focus on installing both tools and in the second I’ll explain how read the indicators that both of these tools are providing.

Project preparation

The entire code base that will be mentioned thru the article is available on GitHub — monitoring-sandbox. It contains a source code of a Java application and all configuration files for both monitoring tools.

The only thing that is required is to have Docker installed. Using simple Docker Compose command (docker-compose up -d) we can start all necessary pieces (monitored application with JFR enabled, Prometheus and Grafana) so you don't need to follow all my steps. They're just for explanation what's going on.

Setting up Java Flight Recorder with JDK Mission Control

Let’s now move to more practical part of the post and have a look on JDK Flight Recorder (JFR) and JDK Mission Control (JMC).

As a name might suggest for a first one it has something in common with black box flight recorder used in airplanes. Similarly to the ones used in aircrafts their primary function is to record and save everything that is happening inside a plane, so in case of an accident specialists could rewind all the steps back to find out what was the cause to prevent from happening it again. In Java JFR plays similar role. It records events that occur inside JVM and application. It collects them and store e.g. in a file. Many of events are already built-in, which enables us to see what’s happening inside our application, e.g. we can see how Garbage Collector is working, how Threads are used etc., but it’s also possible to define custom events inside an application. It could be used for monitoring business processes and performance or whatever we would like.

As already mentioned an output from JFR will be a recording, usually a file. Now how to open it? It could be achieved by using jfr CLI which is bundled with JDK since version 11 (for open source version of Java distributions, for commercial ones it was from Java 7). It allows to list all events by the type and print them in the consol. But it's not convenient. Luckily there is a tool which was already mentioned - JDK Mission Control (JMC).

JMC is a desktop tool that can read JFR recordings and visualize it in a very nice way. Similarly to JFR it was bundled with JDK, but now it’s distributed separately and can be downloaded from the official website.

Having this introduction let’s play around with JFR. For demonstration I’ve prepared a small Java project that is available on GitHub — monitoring-sandbox. For convenience we will be running an app using Docker. Therefore here is a Dockerfile located in a /app folder:

It’s a very simple configuration. The application is running on a recently released OpenJdk 17 from Oracle. This is very important information because depending on the distribution of JDK we can have different results as internals of each distribution might be a little different.

Another thing that is worth to mention is a line with RUN mkdir -p /recordings. With it we're creating a folder inside Docker container in which we will next store JFR recordings.

We could run an app using docker run command, but let's have another approach and use Docker Compose, which will also be used in the next part when we'll be setting up Grafana. Here is the docker-compose.yaml from the project:

A service is called jvm and its Docker image is build using a Dockerfile from /app directory. With volumes we're mapping a project's recording directory with a directory with the same name inside a Docker container. It's necessary to add it, because JFR recordings will be saved there and in order to use them in JDK Mission Control we need extract them first from a running container. Finally there is ports mapping provided.

Having all of that we can move on to the main course — starting an application along with Java Flight Recorder.

There are several ways to start recording. One of them is to enable it using the -XX:StartFlightRecording argument of the java command during application startup. This is why there is an entrypoint provided in the docker-compose.yaml file. Below there is a command that is used to bootstrap an application:

> java -XX:StartFlightRecording:filename=/recordings/recording.jfr,dumponexit=true -jar app.jar

To tune JFR recording we can add some flags. In above example two of them were introduced:

• filename - specifies location and name of a recording, i.e. /recordings/recording.jfr inside of a container,
• dumponexit - if in an application crashes or just stop we can specify if a recording will be dumped, or in other words if JFR should save a recording after stopping a program. In our case a value is true which means it will be persisted.

More of available options can be found here and here.

Let’s start an application and see what will happen. To do that just run following command:

> docker-compose up -d jvm

Now inside /recordings folder there should be a recording.jfr file. If we want now to use JDK Mission Control to inspect recording file we would receive an error. The reason is that first we need to end recording, only then a file can be read. Therefore we need to stop a program:

> docker-compose down

After that we are able to use the recording.jfr file in JDK Mission Control.

But is it possible to get access to recording other way than “crashing” an application? Sure it is!

First of all we need to get inside the running container. We can achieve it by opening an interactive shell for a container with a name jvm-app:

> docker exec -it jvm-app sh

sh-4.4#

As a result we’re in the running container. If for example we run the ls command we gets an information what's inside a container (starting from now on all commands will be executed inside of the container and it's manifested by the sh-4.4#):

> docker exec -it jvm-app sh

sh-4.4# ls
app.jar  bin  boot  dev  etc  home  lib  lib64  media  mnt  opt  proc  recordings  root  run  sbin  srv  sys  tmp  usr  var

To control JFR we use a jcmd CLI which is bundled with JDK. To check what other tools are provided with JDK run the command:

sh-4.4# ls $JAVA_HOME/bin

jar        java   javadoc  jcmd      jdb        jdeps  jhsdb   jinfo  jmap  jpackage  jrunscript  jstack  jstatd   rmiregistry
jarsigner  javac  javap    jconsole  jdeprscan  jfr    jimage  jlink  jmod  jps       jshell      jstat   keytool  serialver

But for now let’s focus on jcmd. By typing only the jcmd we will get a list of Java processes that are happening right now inside of the container:

sh-4.4# jcmd

1 app.jar
183 jdk.jcmd/sun.tools.jcmd.JCmd

Above we can see that there is an app.jar process with a pid (process id) which equals 1. To check if there is an active recording for app.jar type following command:

sh-4.4# jcmd 1 JFR.check
1:
Recording 1: name=1 maxsize=250.0MB (running)

Recording is active, great! Now let’s dump what has been collected so far into a separate file located in the recordings folder:

sh-4.4# jcmd 1 JFR.dump filename=/recordings/short-recording.jfr
1:
Dumped recording, 3.2 MB written to:

/recordings/short-recording.jfr

In above command first we specify apid, which is 1 as we already checked, and then a JFR.dump command that tells to write a recording to a file located in specific directory (filename=/recordings/short-recording.jfr).

Let’s now check a recording using jfr CLI which, as well as jcmd, is provided in JDK.

sh-4.4# jfr summary /recordings/short-recording.jfr

 Version: 2.1
 Chunks: 1
 Start: 2021-09-26 17:26:22 (UTC)
 Duration: 1081 s

 Event Type                                   Count  Size (bytes)
==================================================================
 jdk.NativeMethodSample                       53054        635109
 jdk.JavaMonitorWait                          51660       1393022
 jdk.ThreadPark                                2489         96828
 jdk.GCPhaseParallel                           2195         53536
 jdk.CheckPoint                                1184        833801
 jdk.CPULoad                                   1079         21547
 jdk.JavaThreadStatistics                      1079         12915
 jdk.ClassLoadingStatistics                    1079         11836
 jdk.CompilerStatistics                        1079         31258
 jdk.ExceptionStatistics                       1079         15073
 jdk.ModuleExport                               946         10256
 jdk.ObjectAllocationSample                     780         11772
 jdk.BooleanFlag                                533         15762
 jdk.ActiveSetting                              355         10945
 jdk.ThreadCPULoad                              314          5311

We can print all events that are stored in recording, but be aware that if your recording was very long you’ll get a huge amount of events:

sh-4.4# jfr print /recordings/short-recording.jfr

jdk.ActiveSetting {
  startTime = 17:26:22.702
  id = 1185
  name = "threshold"
  value = "0 ns"
  eventThread = "main" (javaThreadId = 1)
}

jdk.ActiveSetting {
  startTime = 17:26:22.702
  duration = 0.732 ms
  id = 132
  name = "period"
  value = "beginChunk"
  eventThread = "main" (javaThreadId = 1)
}

jdk.ActiveSetting {
  startTime = 17:26:22.702
  duration = 0.732 ms
  id = 132
  name = "enabled"
  value = "true"
  eventThread = "main" (javaThreadId = 1)
}

It might be quite overwhelming, therefore we can filter them out using --categories option:

sh-4.4# jfr print --categories 'GC' /recordings/short-recording.jfr

jdk.GCPhasePause {
  startTime = 17:32:37.583
  duration = 3.55 ms
  gcId = 10
  name = "GC Pause"
  eventThread = "VM Thread" (osThreadId = 14)
}

jdk.GarbageCollection {
  startTime = 17:32:37.583
  duration = 3.55 ms
  gcId = 10
  name = "G1New"
  cause = "Metadata GC Threshold"
  sumOfPauses = 3.55 ms
  longestPause = 3.55 ms
}

jdk.YoungGarbageCollection {
  startTime = 17:32:37.583
  duration = 3.55 ms
  gcId = 10
  tenuringThreshold = 15
}

jdk.G1GarbageCollection {
  startTime = 17:32:37.583
  duration = 3.55 ms
  gcId = 10
  type = "Concurrent Start"
}

Even if we can filter those events and as they are quite easy to read still they’re pretty hard to analyze. We need to have a bigger picture. So this is why we need to use a different tool — JDK Mission Control, which will make for us some visualizations.

First exit from a running container, so just type exit. Now let's open JMC assuming that we have already installed it from the official website. The procedure will depends on the OS system and it's described in instructions.

Once it has started we need to load recording file. Therefore select File > Open file… and find a recording file, which can be either a short recording or the entire one that’s created when application shuts down (remember both of them are located in recordings folder). After that you should get similar screen:

On the left there are couple categories into which we can dive in. After clicking on one of them we can see more details, e.g. by clicking on Memory we can get an insight of application memory allocated and how long Garbage collection took:

We can see even more details about Garbage Collection by clicking on the Garbage Collection item:

Another good example of information that we can get from JMC is total CPU usage (Processes item):

But there a way more information that we can read from it, which I would encourage you to explore on your own (or check my next article on that topic 😉). For now this was everything that I would like to show and let’s move on to the next section.

Setting up JVM dashboards in Grafana

Previously discussed JFR is an amazing tool to get very in depth information about working JVM. But the tricky part is how to set it up so the information will be collected and stored in continuous way, without manually dumping data every time we would like to make an analysis. The solution to such inconvenience would be to make a use of something that’s called JFR Event Streaming, but probably it was introduced in Java 14 it’s still in very early stages and not many tools were built on top of that. Possible reason is that vast majority of an industry still sticks with the latest LTS version of Java, which is 11. With a recent release of Java 17, which is an LTS version, it will change soon as more and more companies will be migrating to it.

Until then there are other tools which might come with help. There are dozens of them some open-sourced, some commercial. Today I’ll show you how to use a very popular ones — Prometheus and Grafana.

Here are all the pieces of a project that will be built:

• Micrometer — allows to expose metrics from JVM application,
• Prometheus — scrapes and stores metrics from applications,
• Grafana — visualize stored metrics.

Small remark, before we start. If you don’t want to setup everything on your own, you can jump straight forward to the next part. All code is available in my GitHub repository — https://github.com/wkrzywiec/monitoring-sandbox.

First step is to expose metrics by the application itself. They will be than periodically collected and stored by Prometheus. Exposing metrics is very easy in Spring Boot application. The only thing that is needed is to add micrometer-registry-prometheus dependency (e.g. to Maven's pom.xml file):

Another thing that should be added, but it’s not a must, is to include application name into exposed metrics which can be done by adding management.metrics.tags.application property to application.properties or application.yaml file:

After running an application and entering its /actuator/prometheus endpoint something like this should appear:

# HELP jvm_gc_memory_allocated_bytes_total Incremented for an increase in the size of the (young) heap memory pool after one GC to before the next
# TYPE jvm_gc_memory_allocated_bytes_total counter
jvm_gc_memory_allocated_bytes_total{application="jvm-app",} 6.291456E7
# HELP jvm_memory_max_bytes The maximum amount of memory in bytes that can be used for memory management
# TYPE jvm_memory_max_bytes gauge
jvm_memory_max_bytes{application="jvm-app",area="nonheap",id="CodeHeap 'non-nmethods'",} 5840896.0
jvm_memory_max_bytes{application="jvm-app",area="nonheap",id="Metaspace",} -1.0
jvm_memory_max_bytes{application="jvm-app",area="heap",id="G1 Old Gen",} 3.31350016E9
jvm_memory_max_bytes{application="jvm-app",area="nonheap",id="CodeHeap 'profiled nmethods'",} 1.22908672E8
jvm_memory_max_bytes{application="jvm-app",area="heap",id="G1 Survivor Space",} -1.0
jvm_memory_max_bytes{application="jvm-app",area="nonheap",id="Compressed Class Space",} 1.073741824E9
jvm_memory_max_bytes{application="jvm-app",area="heap",id="G1 Eden Space",} -1.0
jvm_memory_max_bytes{application="jvm-app",area="nonheap",id="CodeHeap 'non-profiled nmethods'",} 1.22908672E8

Great! Metrics are now exposed by the application, so we can move on to building monitoring infrastructure with Docker.

Because both Prometheus and Grafana are standalone application we need first to run them. For convenience I’m using Docker Compose, therefore in the docker-compose.yaml file I’ve defined 3 services:

First service, jvm, is a Spring Boot application which will be monitored and it's the same appliacation that was investigated before.

Second service, prometheus, will be used to collect and store metrics. It's based on an official Docker image prom/prometheus, it's exposing 9090 port and map it to a local 2000 port. Because Prometheus persists metrics data over time, we need to create a volume so it will be saved between container startups on local disk.

But that’s not enough. We need to tell Prometheus from where and how to collect data. Luckily, there is an easy way. We can provide a prometheus.yml configuration file and insert it into specific folder in a running container.

For one application only one scrape job is defined. It contains information about the address of scraped application and how often Prometheus needs to call it.

The last service, grafana, is used to visualize downloaded data. Like Prometheus it's based on an official Docker image and is exposing 3000 port. Beside that there are couple things added more for configuration purposes. With environment variables we're providing address to Prometheus service, data source of metrics. Next, in volumes section we're mapping couple of local folders into the Grafana's container.

First file is config.ini, with simple configs (more about it can be found in the official docs):

Next is a provisioning folder with two subfolders. First one, datasources, contains datasources.yml file, which helps to add Prometheus datasource:

In the second folder, dashboards, there is also one YAML file dashboards.yml, in which we configure from where dashboards definitions will be loaded:

On the same level as provisioning folder create another one called dashboards. For now it will be empty, but soon we will add here a JVM dashboard definition so it will be persisted.

All configurations files I’ve included in a common infra folder, which structure is as following (or you can just check it in GitHub repository):

.
├───grafana
│   ├───dashboards
│   └───provisioning
│       ├───dashboards
│       └───datasources
└───prometheus

Having all of that, we can now run the entire stack using Docker Compose.

> docker compose up -d

[+] Building 33.6s (14/14) FINISHED
...                                                                                         0.0s 

Use 'docker scan' to run Snyk tests against images to find vulnerabilities and learn how to fix them
[+] Running 4/4
 - Network "monitoring-sandbox_default"  Created                                                                                                     0.0s 
 - Container prometheus                  Started                                                                                                     1.2s
 - Container jvm-app                     Started                                                                                                     1.2s 
 - Container grafana                     Started                                                                                                     3.3s

Great, all three containers are now running. Let’s go now to Grafana’s UI, therefore in a web browser enter http://localhost:3000 address so the login page will show up. Then provide a username and password which is admin in both cases. After successful login a main page should be visible (you will be asked to change the password, but you can go ahead and skip it).

In a meantime Prometheus is already collecting metrics from the jvm-app, but we’re not yet able to visualize it in Grafana. To fix it we need to have a dashboard. We could create our own one, but there is already a great one available to be imported, which was prepared by the community.

To add it, from left panel select Create (plus sign) and than Import. A new page will appear where we need to provide an ID of dashboard which we’re gonna import. I want to import a very popular one — JVM (Micrometer) with an id = 4701.

After providing it in an input field and clicking Load button a summary page will be visible. We can decide what will be the name of imported dashboard, its location, identifier and the most important, don’t forget to select it, a data source from where data will be pulled for visualizations. In our case it should be available only one source — prometheus.

After clicking Import button a dashboard will be created.

The last thing that we need to do here is to export newly created dashboard to JSON file and place it to the empty dashboards folder, mentioned couple lines before. To do that click on the share icon (left, near the star icon) so the pop up window will appear. Next, go to Export tab and click Save to file so a file will be downloaded.

Next copy-paste it to the empty dashboards folder, stop containers (using docker compose down command) and restart them. After that a dashboard will be persisted.

That would be it! The dashboard is set, you can check it, see what indicators it’s presenting, like memory usage, CPU usage or others. Definitely there are here way less information than we could read from the JDK Mission Control, but they’re one of the most important one.

Conclusion

With this article I’ve showed how to set up two popular monitoring tools that can be used in your production system. They can offer lots of in-depth information what’s happening with your application and JVM. It can be especially helpful when something mysterious is starting to happen. Information about memory allocation, Garbage collection, CPU usage etc. can than help you find a root cause. But how to make sense from all these metrics? This is a story for another article, which will be published soon.

Until then, see you soon!

References

• Continuous Monitoring with JDK Flight Recorder (JFR)
• JDK Flight Recorder - a gem hidden in OpenJDK
• GitHub - petrbouda/docker-java-flight-recorder

It doesn’t matter if you’ve just started your journey with Kubernetes or you’re already an expert in it; writing object definitions is not an easy task. You can very easily make a mistake that could be very costly, if deployed to a production environment. And if you’re just starting to learn Kubernetes, you might need help understanding which metadata and specifications should be provided, but, by design, are not mandatory.

The ideal way to navigate these challenges would be to have an experienced colleague perform a review of your code changes. But sometimes you don’t have such a person around, or it’s very hard to get feedback from them. And even if you do have an amazing person to help you, it still won’t prevent you from making mistakes from time to time.

Instead, I would like to show you another approach: create a simple, automated quality check of your K8s object definitions using Datree.io, Google Kubernetes Engine and Github Actions.

We’ll build a GitHub Actions workflow which will be triggered after each change made on the master branch. The workflow will have two stages:

• Datree analysis of missing configurations, and quality check,
• Deployment and testing an example application in a real cluster (GKE).

So roll up your sleeves and let’s automate it!

Workflow structure

First of all you need a GitHub repository to hold our YAML files. I’m using my old project — k8s-helm-helmfile. This repository has three folders, each containing a different approach to deploy applications into Kubernetes clusters. You can read more about those approaches in my previous blog posts about vanilla K8s, Helm and helmfile deployments.

To keep this blog post short, I’ll show how to create a workflow that will use Helm to deploy an application, but you can easily do this with other approaches, such as those mentioned previously.

The first step is to create a workflow definition file. In the root folder of a repository create a new directory ./github/workflows inside of which there will be a master.yaml file:

It contains following specifications:

• name - name of our workflow
• on - a condition based on which a workflow will be triggered. A workflow will only be started when changes are committed on the master branch.

With that, you can move on to the best part — defining jobs.

Datree analysis

In this part you will use a free tool called Datree, which analyzes K8s definitions and will stop the workflow if it finds any problems. It’s very important to have a safety net like this, so you can feel confident that even if you make a mistake, or aren’t aware of best practices, an assistant will keep you on track.

Before defining a GitHub workflow, let’s install Datree locally. To do that, go to their official website which will guide you how to install the Datree CLI. I’m using Linux (or to be precise, WSL), so only the following command is necessary:

> curl https://get.datree.io | /bin/bash

After couple of seconds Datree will be installed.

To test it out, go to the folder where Kubernetes YAML files are located and run Datree test command (in my case I’m using a vanilla K8s file from k8s-helm-helmfile project):

> datree test adminer-deployment.yaml

where the adminer-deployment.yaml file is a Kubernetes object definition.

Here is the output that I got:

As you can see, Datree has prepared a short summary of how many rules this YAML file violates. It provides a very useful explanation with a hint on how to fix them.

Starting from here, you can work on each issue to make these tests pass. It’s a great way to learn and practise. But what if you deliberately chose not to comply with some of the rules? Luckily, Datree gives a possibility to prepare a custom policy, a set of rules against which YAML files will be checked.

To set up a policy, you need to go to your dashboard. Your personal link to it is located at the end of each scan, in the summary table, in the last row called See all rules in policy (I've marked it in a previous screenshot). It will take you to a login page.

Then you need to login. For convenience, I would suggest using the GitHub account. After successful authorization you will reach a Policies page.

Here you can inspect all the rules that you can turn on and off. By default, only some of them are enabled. To activate or deactivate them, use the toggle button visible next to the rule name.

If a name of a rule is too enigmatic, you can check its details by clicking on its name. It will then show more information and a sample output when this rule is violated.

Another cool feature is that you can define your own tips on how to fix a problem. The defaults should be enough, but if you prefer to extend it, view it in your own language, or add a link to a blog post, Stack Overflow discussion or any other online material, you can put it here by simply clicking an Edit button.

Finally, you can create your own set of rules, known as policies, to run for different applications, environments and stages. Simply click the Create Policy button at the top of the page.

Let’s now check the second page, available on the left panel, called History. As the name might suggest, here you can see a convenient summary of all previous test runs.

Before writing this post, I’d already played around with Datree, which is why I already have several test runs listed in the history panel, but in your case you should have only one.

That would be it for a quick tour around Datree dashboard!

Now let’s build a GitHub Actions workflow.

First, provide your Datree token as an environment variable in a workflow. To achieve this, click on your avatar in the top right corner of the Datree dashboard, then click Settings. It’ll guide you to the page where you can find your token.

Once you’ve copied the token, go to your GitHub project/repository’s Settings page. Then select Secrets and click the New repository secret button.

In the Name field put DATREE_TOKEN and in the Value field a token copied from Datree.

Now you can move on to the workflow’s definition file and define the first job:

It’s called datree, it runs on the latest Ubuntu inside a dtzar/helm-kubectl Docker container. I've selected this setup because I would like to run tests against Helm release (and not vanilla K8s as it was done previously). The reason to choose this Docker image is because it contains necessary dependencies (K8s and Helm), so I can skip their installation and make my workflow faster.

Let’s define the next three steps:

With the first one, actions/checkout@v2, you get code from the repository. In step two, you install the Datree Helm plugin to run Datree tests using Helm. In the third step, you run actual tests using a Helm CLI. The DATREE_TOKEN environment variable is added to it so the result will be linked with a Datree account.

Finally, in an actual run script, I provide the location of a Helm template and the location of testing values.yaml file.

After that you can commit changes and push it to GitHub. It should trigger a workflow, which will be available in the Actions tab.

In my case, several tests are failing, demonstrated by the failing status of a workflow. To find out more, click on the failing job. It will take you to the console output, where you can investigate what went wrong.

As well as checking policy check result(s) in the workflow’s console, you can go back to the History page in the Datree dashboard and analyze errors there.

If you have a similar screen to mine, stop here, correct mistakes and push changes to GitHub. If you feel that some rules are obsolete for you, turn them off in the Datree dashboard, but do not turn off too many!

After fixing all the issues and re-running the workflow, the previously marked red exes [X] will become green check marks [V], indicating that the workflow has passed the validation and policy check.

And some details about a job:

The successful result will be also visible in the Datree dashboard:

Awesome! We can now move on to the next part.

Testing on GKE

After making sure that templates are okay from Datree ‘s point of view, move to deploying them to a real Kubernetes cluster (which is not production) and then check if everything is working there, e.g. if a website is available over the internet, etc.

First step would be to have a cluster. I’ve picked Google Kubernetes Engine ( GKE) because of its free tier, but if you have your own cluster (on AWS or any other cloud provider) just use it.

Right now I’ll follow with steps that are necessary to set up a job for GKE, so if you already have a GKE or any other Kubernetes cluster up and running, skip this part.

Before adding a new job to a workflow, you need to set up couple of things in GKE (these instructions are based on official GKE Quickstart guide):

Until now, all configuration was carried out in a web browser, but now you need to move to either Google Cloud Shell, or install Cloud SDK and follow the instructions in a terminal:

Set up basic configuration in gcloud tool, like default project, region and zone. In my case, the project name is k8s-helm-helmfile and the region is europe-central2, but for you it may be different.

> gcloud config set project k8s-helm-helmfile

> gcloud config set compute/region europe-central2

> gcloud config set compute/zone europe-central2-a

Create an IAM Service Account, which is an account which will be used in GitHub Action workflow. I’ve called mine helm-github-actions-service:

> gcloud iam service-accounts create helm-github-actions-service

Get an email from the newly created Service Account. You will need it for the next step:

> gcloud iam service-accounts list

Assign roles to a Service Account, where <EMAIL> tag is taken from previous step:

> gcloud projects add-iam-policy-binding k8s-helm-helmfile \
  --member=serviceAccount:<EMAIL> \
  --role=roles/container.admin \
  --role=roles/storage.admin \
  --role=roles/container.clusterAdmin \
  --role=roles/iam.serviceAccountUser \
  --role=roles/container.developer

Export the Service Account Key:

> gcloud iam service-accounts keys create key.json --iam-account=<EMAIL>

> export GKE_SA_KEY=$(cat key.json | base64)

Add it as a GitHub Secret to a project (the same way as for DATREE_TOKEN) with GKE_SA_KEY as key. To see the value of the exported key you can use the command:

> printenv GKE_SA_KEY

Everything is set up, now go back to workflow definition where you first created a Kubernetes cluster on GKE, deploy a sample Helm release, and test it.

First create a new job called gke :

Similarly to the previous example, you have name and runs-on configurations. There is also a needs configuration, which means that in order to run this job, the datree job needs to first be completed successfully. This prevents spinning up clusters and deploying a sample application if something goes wrong during a Datree check. The last part of job configuration are environment variables ( env), which will be used in workflow steps. They're my GCP project id, K8s cluster name, and my GKE region.

Moving on to the next steps:

The first one is for getting code from the project. The second is to configure (login and set up project) Google Cloud CLI which will be used in following steps.

The above steps create a new GKE cluster and configure kubectl so it’s connected with the newly created cluster. With that, you can move on to the step where an adminer Helm release will be installed:

Stop here and analyze what’s going on. First of all, you’re using the deliverybot/helm GitHub Actions, which provides a convenient way to use Helm. By adding a few parameters, you can deploy an application onto a Kubernetes cluster. The entire list of available parameters can be found on the official website.

In the above example, I’ve used the following steps alone:

• release - a release name,
• namespace - specifies the K8s namespace where the app will be installed,
• chart - gives information about the location of a Helm chart,
• helm - indicates which version of Helm will be used,
• value-files - file used to override the default values from a Helm chart, in my case it's an Adminer's values.yaml file (the Helm chart I use for testing, which deploys popular database client - Adminer),
• values - this parameter works pretty the same as previous one - it's used to override the default values from a Helm chart, but instead of doing it with a file, we can directly specify values that need to be overridden; here, I'm overriding only the Service kind, as by default it's a ClusterIP, but I don't want to change it in the adminer.yaml file.

After successful installation, our workflow can proceed with tests. I’ve decided to run a very simple test, which only checks if a page is opening, but you can of course build a more sophisticated test chain.

The first line in a testing script is for finding out under which IP address the adminer application is exposed. The second is for debugging, and the last for actual testing.

Once the tests are done, you need to destroy a cluster:

The if: ${{ always() }} part is very important here. It makes sure that even if any of the previous steps fail, this one will always run. Otherwise you could end up with a bill from Google at the end of a month.

After committing changes and pushing them to GitHub, a workflow will look like this:

First a Datree step is executed, then the installation is done on GKE. To check the details of the second step, click on its name.

Conclusion

And that’s it for today! I hope that this blog post encourages you to build something like this on your own, give Datree (or any other static code analysis tool) a try, and set up a cluster for automated tests, so you will feel more confident about the changes you made in your code base. It can all be set up and operational in a flash.

Originally published at https://dev.to on September 2, 2021.

This is a fourth and the last part of my series on OAuth 2.0. If you’re not familiar with I would recommend to stop here and go check the first one — Introduction to OAuth 2.0.

The aim of this post is to show you a basic set up an Angular application so that it will be integrated with Keycloak and it will be able to consume protected HTTP resource that requires an access token. A full description of how to set up Keycloak instance and how to create a simple protected resource with Spring Boot I’ve described in previous two articles. But don’t worry if you don’t want to do that, I’ll explain it later how.

The goal

For a purpose of this article I’ve prepared a very simple application which could be used to find out information about movies and display them in a table. Application supports two actions — either fetch all the data or select a single movie by its id.

All data are served by Java application, which requires to provide a valid OAuth 2.0 access token. If anything goes wrong, for example user will not be allowed to consume certain resource a toast message at the bottom of the screen will show up.

Apart from frontend app there are two other components in this project. Backend application and Keycloak. First one was already mentioned, in second there are defined users and clients, which are necessary when issuing for an access token.

Both parts are crucial to make our Angular application work. Both are required to be running. I assume that you might not be interested in going thru my previous tutorials to implement everything by yourself and that’s why I’ve prepared Docker compose file which will allow to start all necessary applications with a single command with my default configuration. Therefore clone my keycloak-security-example repository and checkout to a frontend-start-blog branch. It will be our starting point.

Once you’ve got that we need to start both services, so go to the terminal and in the root directory of a project run following command:

> docker-compose up -d backend

In a first run it could take a little while, but after that it should be good. To check if the applications are running you can use command:

>  docker ps
IMAGE                               STATUS                NAMES
keycloak-security-example_backend   Up About a minute     backend
jboss/keycloak:11.0.2               Up About a minute     keycloak
postgres:13.0-alpine                Up About a minute     postgres

All three containers are Up so we're good to proceed with configuring a hosts file on your machine. Its location depends on the OS:

• Linux (Ubuntu)
• Windows 10
• Mac

So please find it, open in an admin mode or relevant and add following line:

127.0.0.1	keycloak

This step is necessary because the address of Keycloak server is different depending on the point of view. From the point of view of an app that is running in the web browser it’ll be http://localhost:8080, but from the perspective of backend application that is running inside the Docker network it will be http://keycloak:8080. To overcome it above config is necessary.

If you like you can add similar entries for frontend and backend (IP address remains the same), because the side effect of it will be that instead of using for example http://localhost:4200 you can use more meaningful http://frontend:4200. But it's totally optional and depends on your preferences.

Register frontend application in Keycloak

If you’re running my preconfigured Keycloak instance you can skip this part or just quickly go thru it.

Based on OAuth 2.0 protocol we need to register our application in Keycloak, because only allowed services can issue an access token.

Therefore, open the Keycloak page — http://localhost:8080, select Administration Console and provide following credentials:

username: admin
password: admin

After login, in the top right corner there will be displayed a name of a realm into which you’re logged in. Make sure that Test is selected.

Now go to the Clients page (it’s located in the left menu) and click Create button located on a right.

In a new view we need to provide only a client name, which will be frontend.

After hitting the Save button a detailed page for newly created client will show up. To finish configurating frontend client we need first to make sure that Access Type is set to public. It indicates that our client is public and will not require secret to get access token. This config is valid for client-side frontends which are not able to securely store secrets for their clients (because the entire code of an application is running in users browser).

Apart from that we need to set up a Root URL to http://localhost:4200 which is a root URL of our application. Also we need to provide a Valid Redirect URIs which is very important from security point of view, because here we’re defining into which URLs Keycloak can redirect after successful authentication. We should provide here only our application’s URLs, to prevent hackers from doing messy stuff. Therefore here I’ve got 3 entries: http://localhost:80/*, http://localhost:4200/*, http://localhost/*.

The last thing that needs to be configured is Web Origins parameter and here I’m providing + which will handle the CORS problem.

After that click Save button at the bottom of the screen to finish Keycloak configuration (all other steps, like user and roles definition was already covered in previous blog posts).

Main view of a page

Now we can deep dive into Angular project. It has only one component (apart from AppComponent) - ContentComponent.

To keep you sane and to keep this post short I’ll skip showing HTML and CSS files, as the aim of the project is not to look nice.

From the above code you can see that in order to connect to the backend application it uses the MovieBackendService which encapsulates HTTP requests:

Again, fairly simple.

To make it work I’ve also set up a proxy configuration, both for development and production mode. For a first one I’ve created a proxy.conf.dev.json file:

The target field indicates under which address the backend service is available. To apply it I’ve also modified the start script in package.json file:

For production proxy config I’ve created the default.conf file, which is a configuration of Nginx server:

server {

    listen 80;
    server_name frontend;
    root /usr/share/nginx/html;
    index index.html index.html;

    location /movies {
        proxy_pass http://backend:9000/movies;

    }

    location / {
        try_files $uri $uri/ /index.html;
    }
}

The http://backend:9000/movies is an address of the backend service (if a backend service is running in the same Docker network, or in the same Kubernetes cluster with the same Service name).

Having all of that we can now proceed with adding and configuring Keycloak libraries.

Basic configuration of Keycloak libraries

First step would be to add keycloak-angular dependencies to the project, therefore in a terminal run following command:

> npm install keycloak-angular keycloak-js

Having it we now need to config it with our application. What is important here, before running our own code first we need to authenticate a user, show Keycloak’s login page and so on. To achieve it we will use an Angular’s special InjectionToken called APP_INITIALIZER. A function, which will be assigned to this token will be executed before loading the rest of an application.

To create a custom function, that we will than assign to APP_INITIALIZER just run this command:

> ng g class init/keycloak-init --type=factory --skip-tests
CREATE src/app/init/keycloak-init.factory.ts (22 bytes)

A new file was created. Open it and paste following content:

Above function takes as an argument a KeycloakService which is an object from keycloak-angular library so that we can use it in a body of a function to config it by providing URL to the server, realm name and application's client id which was created a step before.

For now all values are hardcoded here, which of course is not good, but we’ll tackle this problem later on.

Next we need to register this function in the app.module.ts file, as a provider:

Take a closer look in the imports section. There is KeycloakAngularModule added there.

Next step would be to create a Guard that will protect routes in an application. The library that we’re using is already providing a preconfigured, abstract KeycloakAuthGuard class from which we will extend ours. But first let's create one:

> ng g guard guard/auth --skip-tests

? Which interfaces would you like to implement? CanActivate
CREATE src/app/guard/auth.guard.ts (458 bytes)

The content of this Guard would be:

In the isAccessAllowed method we're adding the piece that will redirect a user to a login page if she/he is not authenticated yet.

To finish this part, the only thing to do is to add this Guard to the app-routing.module.ts.

Having all of that we can test it! Make sure that both backend and Keycloak services needs to be running ( docker-compose up -d backend ) and run the command:

> npm start

> frontend@0.0.0 start .\keycloak-security-example\frontend
> ng serve --proxy-config src/assets/proxy.conf.dev.json

- Generating browser application bundles...[HPM] Proxy created: /movies  ->  http://localhost:9000
[HPM] Subscribed to http-proxy events:  [ 'error', 'close' ]
√ Browser application bundle generation complete.

Initial Chunk Files   | Names         |      Size
vendor.js             | vendor        |   5.11 MB
styles.css, styles.js | styles        | 562.38 kB
polyfills.js          | polyfills     | 493.08 kB
main.js               | main          |  40.76 kB
runtime.js            | runtime       |   6.15 kB

                      | Initial Total |   6.18 MB

Build at: 2021-02-20T15:55:55.362Z - Hash: 360dd85cc05c08ec0698 - Time: 14478ms

** Angular Live Development Server is listening on localhost:4200, open your browser on http://localhost:4200/ **

√ Compiled successfully.

Now if you open the http://localhost:4200/ page you should be redirected to the login page:

And provide credentials:

username: han/luke
password: password

After that you should be able to enter the main page! If you now try to play around with an app, by clicking All movies button or selecting movie by id, in one of the cases you'll get this kind of an error:

This toast message inform the user that she/he doesn’t have sufficient roles. But how application knows that?

First of all after successful login OAuth2.0 tokens were fetched from the Keycloak server and stored in your web browser. Now if there is any request made to the backend service our great Keycloak library is adding the Authorization HTTP header with an access token. And such is required to be consumed by the backend application. It validates a content of a token and returns OK response if everything is fine, or returns an error if token is corrupted or user is not allowed to consume specific endpoint (because she/he doesn't have sufficient roles).

Productionize Keycloak configuration

So far so good, we’ve got a working piece of code. But there is one problem with this approach. If we would need to change the Keycloak’s URL, realm name or client (development, test and production environment can be different) we would be forced to change it in the code for each version, as all these values are hardcoded.

To overcome this problem we need to externalize configuration of an app. My approach would be to pass those values from the environment variable during startup of an application. Because we’re using Docker it’s quite easy to achieve.

First step is to create separate configuration files for two cases — development and production. Therefore in the assets folder create a new config folder inside of which place these two following files:

• config.dev.json

• config.prod.json

In the latter config file there are no hardcode values, just a placeholder for environment variables, which will be replaced during the startup of an application.

Next step would be to make use of these two configuration. Therefore got to environments directory and add a configFile value which will point out to a different config file.

• environment.ts

• environment.prod.ts

Also create a new file, environment.dev.ts which will be the same as the default one:

Having that we could now move on to the code and make use of it, but after the dev environment is not fully setup. We need to add this new environment to an Angular configuration.

Therefore open the angular.json file and locate the projects.frontend.architect.build.configurations section and add dev configuration which will be replacing the default environment.ts file with environment.dev.ts:

In the same file, scroll down a little bit to the serve section and in configurations add new dev entry with browserTarget:

Finally we can go to the package.json file in which we can specify that when we want to run the start script we would like to enable the dev profile, by adding the -c dev parameter.

After this change once we run the npm start the Angular app will be build with dev configuration.

If we run an application right now nothing will change, because we only registered a new application profile, but haven’t make a use of it in the code. Therefore let’s now focus on the code itself.

Right now we’re initializing KeycloakService before the entire application. What we need to do now is to load configuration before KeycloakService. To do that first let's create a ConfigInitService which will load a configuration file and assign it's values to a variable.

> ng g service init/config-init --skip-tests
CREATE src/app/init/config-init.service.ts (140 bytes)

And the content of this service class in which we’re using the environment’s value that points out to the correct configuration file (dev or prod).

Next, we need to go back to the keycloak-init.factory.ts and adjust it so that getConfig() method of above service is invoked before KeycloakService initialization.

This time a code is a little bit complicated. confiService is returning the config object from which we can get necessary values.

Last thing to do in the app code is to register a ConfigInitService in app.module.ts and pass it to the APP_INITIALIZER.

If we run the npm start command the app will start up and works as previously. Awesome, we're in the same situation as we were previously 😏.

Let’s move forward and change the Docker image definition. If you look at the Dockerfile for frontend app it looks pretty straightforward:

It’s a two stage Dockerfile, in first step a project is build to HTML, CSS and JSes and in the second step all these files are copied to an NGINX server together with simple server configuration in default.conf file.

What we need to change in above file is to somehow replace placeholders from config.prod.json with relevant environment variables that will injected into Docker container during startup. Apart from that it would be to do similar thing with default.conf file in which we’ve got a proxy_pass defined which routes traffic to the backend service. Instead of hardcoding the backend URL we can allow to dynamically assigning it during container startup.

Therefore in default.conf file we need to replace the http://backend:900 value to ${BACKEND_BASE_PATH}.

server {
    listen 80;
    server_name frontend;
    root /usr/share/nginx/html;
    index index.html index.html;

    location /movies {
        proxy_pass ${BACKEND_BASE_PATH}/movies;
    }

    location / {
        try_files $uri $uri/ /index.html;
    }
}

But we still have a problem, how to replace those placeholders with values from environment variables? Nothing simpler, we can use envsubt bash command.

First, let’s create a bash script located in the root folder of frontend project and call it replace_placeholders.sh:

In a first line we’re replacing the placeholders from config.prod.json and as a result it generates a config.json file with values from environment variables (which holds Keycloak url, realm and client info). With the second line we’re replacing only one placeholder with a base url to the backend application in default.conf — NGINX server configuration.

Having it setup we can change Dockerfile definition, so that it will copy a script to a container and run it during the startup.

Finally to run the application we need to declare those environment variables. Depending on how you would like to run the application there are many ways to achieve it. For example they can be declared in OS, passing them as parameters using Docker commands, using Docker Compose file or be declared in Kubernetes Deployment or ConfigMap.

For me the most convenient is to use Docker Compose (docker-compose.yaml file located in the project root)

version: "3.8"

services:
  
  frontend:
    build: ./frontend
    container_name: frontend
    ports:
      - 80:80
    environment: 
      - KEYCLOAK_URL=http://keycloak:8080
      - KEYCLOAK_REALM=test
      - KEYCLOAK_CLIENT_ID=frontend
      - BACKEND_BASE_PATH=http://backend:9000
    depends_on:
      - keycloak
      - backend

  # definitions of other services

Now run the following command and a Docker container will be created and then started.

> docker-compose up -d frontend

To check if everything is working we can list Docker containers:

> docker ps

CONTAINER ID    STATUS          NAMES
1840d7564aeb   Up 46 seconds    frontend
cba18013881c   Up 47 seconds    backend
01f15608d210   Up 47 seconds    keycloak
ac67959019f9   Up 48 seconds    postgres

Than go to a web browser and type http://localhost so it will redirect you to the login page and once you login you can play around with an app.

That's it for today and for entire series about OAuth 2.0 framework. I hope you've enjoyed and learnt some new things. Please let me know in the comments what do you think.

The entire code for this project can be found in my GitHub repository:

wkrzywiec/keycloak-security-example

Here is the list of all other posts in this series:

• Introduction to OAuth 2.0
• Create and configure Keycloak OAuth 2.0 authorization server
• Implementing OAuth 2.0 access token validation with Spring Security

References

• mauriciovigolo/keycloak-angular
• Angular Authentication: Using Route Guards