Chapter 22 : UseCase19 - Retrying Failed Flows

In this chapter we learn how to recover from failure automatically.

Business Scenario

Assumption:

• Country analytics calls a remote pricing service.

• That service sometimes fails.

• We want to retry a few times before giving up.

This is exactly what retry and retryWhen are built for.

CODE IMPLEMENTATION

package org.kotlinflowlearner.stockflow.usecases.uc19

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import org.kotlinflowlearner.stockflow.model.Stock
import java.io.IOException

/**
 * UC19 demonstrates how retry and retryWhen work.
 *
 * This use case simulates a failing upstream operation.
 * The flow automatically retries before finally succeeding or failing.
 */
class UC19FlowRetry {


    /**
     * Executes the use case.
     *
     * @param stocks List of stocks loaded from CSV.
     * @return Flow of processed stock names with retry logic.
     */
    fun execute(stocks: List<Stock>): Flow<String> {

        return stocks.asFlow().map{
            stock -> simulateUnstableService(stock)
            "Successfully processed: ${stock.name}"
        }.retry(3).catch {
                exception -> emit("Final failure: ${exception.message}")
        }
    }

    /**
     * Simulates an unstable service that randomly fails.
     */
    private suspend fun simulateUnstableService(stock: Stock) {
            delay(200)

            // Simulated failure condition
            if(stock.rank % 5 == 0){
                println("Service failed for ${stock.name}")
                throw IOException("Temporary failure for ${stock.name}")
        }
        println("Service succeeded for ${stock.name}")
    }
}

package org.kotlinflowlearner.stockflow.usecases.uc19

import kotlinx.coroutines.runBlocking
import org.kotlinflowlearner.stockflow.csv.CsvStockLoader
import java.nio.file.Path

fun main(){
    runBlocking {
        val useCase = UC19FlowRetry()

        val resource = requireNotNull(
            object {}.javaClass.classLoader.getResource("stocks.csv")
        )

        val path = Path.of(resource.toURI())

        val stocks = CsvStockLoader.load(path)

        useCase.execute(stocks).collect{
                result -> println("Collected : $result")
        }
    }
}

OUTPUT

Service succeeded for Aurora Systems
Collected : Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected : Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected : Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected : Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Service succeeded for Aurora Systems
Collected : Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected : Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected : Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected : Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Service succeeded for Aurora Systems
Collected : Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected : Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected : Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected : Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Service succeeded for Aurora Systems
Collected : Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected : Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected : Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected : Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Collected : Final failure: Temporary failure for Vertex Financial

What Happens Internally

When an exception is thrown:

1. Flow is cancelled.

2. retry(3) intercepts it.

3. Flow restarts from the beginning.

4. It tries again up to 3 times.

5. If still failing, catch handles it.

   Note: Retry restarts the entire upstream chain.

In the above result log you will find the below failure message four times. Initial error message is produced when the simulated failure is triggered and because of the retry happening three times, we see the same error message again.

Service failed for Vertex Financial

retry(n) versus retryWhen{}

retry(n) retries for a fixed number “n” times. retryWhen() is intelligent.

We have added the addition function executeWhen and revised the class code.

package org.kotlinflowlearner.stockflow.usecases.uc19

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import org.kotlinflowlearner.stockflow.model.Stock
import java.io.IOException

/**
 * UC19 demonstrates how retry and retryWhen work.
 *
 * This use case simulates a failing upstream operation.
 * The flow automatically retries before finally succeeding or failing.
 */
class UC19FlowRetry {


    /**
     * Executes the use case.
     *
     * @param stocks List of stocks loaded from CSV.
     * @return Flow of processed stock names with retry logic.
     */
    fun execute(stocks: List<Stock>): Flow<String> {

        return stocks.asFlow().map{
            stock -> simulateUnstableService(stock)
            "Successfully processed: ${stock.name}"
        }.retry(3).catch {
                exception -> emit("Final failure: ${exception.message}")
        }
    }

    fun executeWhen(stocks: List<Stock>): Flow<String> {

        return stocks.asFlow().map{
                stock -> simulateUnstableService(stock)
            "Successfully processed: ${stock.name}"
        }.retryWhen { cause, attempt ->

            if (cause is IOException && attempt < 3) {
                delay(500)
                true
            } else {
                false
            }
        } .catch { e ->
            emit("Final failure handled: ${e.message}")
        }
    }

    /**
     * Simulates an unstable service that randomly fails.
     */
    private suspend fun simulateUnstableService(stock: Stock) {
            delay(200)

            // Simulated failure condition
            if(stock.rank % 5 == 0){
                println("Service failed for ${stock.name}")
                throw IOException("Temporary failure for ${stock.name}")
        }
        println("Service succeeded for ${stock.name}")
    }
}

package org.kotlinflowlearner.stockflow.usecases.uc19

import kotlinx.coroutines.runBlocking
import org.kotlinflowlearner.stockflow.csv.CsvStockLoader
import java.nio.file.Path

fun main(){
    runBlocking {
        val useCase = UC19FlowRetry()

        val resource = requireNotNull(
            object {}.javaClass.classLoader.getResource("stocks.csv")
        )

        val path = Path.of(resource.toURI())

        val stocks = CsvStockLoader.load(path)

        useCase.execute(stocks).collect{
                result -> println("Collected : $result")
        }

        useCase.executeWhen(stocks).collect{
                result -> println("Collected retryWhen: $result")
        }
    }
}

OUTPUT

Service succeeded for Aurora Systems
Collected retryWhen: Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected retryWhen: Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected retryWhen: Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected retryWhen: Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Service succeeded for Aurora Systems
Collected retryWhen: Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected retryWhen: Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected retryWhen: Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected retryWhen: Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Service succeeded for Aurora Systems
Collected retryWhen: Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected retryWhen: Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected retryWhen: Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected retryWhen: Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Service succeeded for Aurora Systems
Collected retryWhen: Successfully processed: Aurora Systems
Service succeeded for BluePeak Energy
Collected retryWhen: Successfully processed: BluePeak Energy
Service succeeded for Helios Dynamics
Collected retryWhen: Successfully processed: Helios Dynamics
Service succeeded for Nimbus Health Group
Collected retryWhen: Successfully processed: Nimbus Health Group
Service failed for Vertex Financial
Collected retryWhen: Final failure handled: Temporary failure for Vertex Financial

In the executeWhen() function, we do the same simulation for failure but use the retryWhen{} block.

What Is Happening Step-by-Step

If the function simulateUnstableService(stock) throws IOException, then:

1. retryWhen catches it.

2. If attempt < 3, it retries.

3. After 3 retries, attempt becomes 3.

4. The condition fails.

5. It returns false.

6. The exception is re-thrown downstream.

7. Catch operator is invoked which handles the error gracefully and emit a recovery message.

MENTAL MODEL

• Attempt 0 → retry

• Attempt 1 → retry

• Attempt 2 → retry

• Attempt 3 → returns false

When it returns false, the exception is re-thrown.
If you don’t have catch{} after it, the coroutine crashes. catch {} handles the error gracefully.

retryWhen does not resume from the failure point. It restarts upstream.

SUMMARY

Retry is essential when:

• Calling APIs

• Reading from network

• Connecting to databases

• Reading files

• Microservice communication

Retry does NOT retry downstream operators. It only retries upstream of where it is placed.