CodeLab

Level: Third-semester Java programming
Theme: Threads, task queues, worker threads and concurrent HTTP requests

You are building a small batch-processing application. The application receives many independent tasks, but it must not create a new thread for every task.

You will first build a manual fixed-size thread pool. Your pool will contain a shared task queue and a limited number of long-running worker threads. You will then use the pool to process several HTTP requests concurrently.

In the final part, each task will send a different prompt to the Gemini API and display the response.

You must implement the pool management yourself. Do not use ExecutorService, Executors, CompletableFuture, parallel streams or virtual threads in the required solution.

Imagine that your application has 100 tasks. This implementation creates 100 threads:

for (Runnable task : tasks) {
    new Thread(task).start();
}

That approach gives the application no control over how many tasks run simultaneously. Instead, your application should place tasks in a queue and create only a small number of workers.

100 submitted tasks -> shared queue -> 5 worker threads

When one worker finishes a task, it takes the next task from the queue.

By the end of the Codelab, you should be able to:

• distinguish between a task and a thread;
• explain why creating one thread per task may not scale;
• safely share a BlockingQueue between threads;
• implement long-running worker threads;
• limit concurrency with a fixed number of workers;
• stop workers using poison-pill tasks;
• wait for worker threads using join();
• keep one failed task from terminating a worker;
• use the pool for concurrent HTTP requests; and
• explain how your implementation relates to ExecutorService.

Your manual thread pool will have three important elements:

Each worker should follow this general algorithm:

repeat
    wait for a task
    if the task is a stop signal
        stop the worker
    otherwise
        execute the task

Before implementing the pool, create 20 simulated tasks and execute them sequentially.

public static Runnable createSimulatedTask(int taskNumber) {
    return () -> {
        String threadName = Thread.currentThread().getName();

        System.out.printf(
                "START task %d on %s%n",
                taskNumber,
                threadName
        );

        try {
            Thread.sleep(500);
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            return;
        }

        System.out.printf(
                "END   task %d on %s%n",
                taskNumber,
                threadName
        );
    };
}

Execute the tasks on the main thread:

long start = System.currentTimeMillis();

for (int i = 1; i <= 20; i++) {
    createSimulatedTask(i).run();
}

long duration = System.currentTimeMillis() - start;
System.out.println("Sequential duration: " + duration + " ms");

1. Which thread executes all the tasks?
2. What duration do you expect before running the program?
3. Why does calling run() not create a new thread?

Record the result:

• The program creates 20 tasks.
• The main thread executes every task.
• The total duration is measured correctly.

Create this class:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ManualThreadPool {

    private final BlockingQueue<Runnable> taskQueue;
    private final List<Thread> workers;

    public ManualThreadPool(int workerCount) {
        // TODO: Validate workerCount.
        // TODO: Create the queue and worker collection.
        // TODO: Create and start workerCount workers.
    }

    public void submit(Runnable task) {
        // TODO: Put the task into the queue.
    }

    private void processTasks() {
        // TODO: Repeatedly take tasks from the queue.
        // TODO: Execute each task.
    }

    public void shutdown() {
        // Implement this in Part 4.
    }

    public void awaitTermination() {
        // Implement this in Part 4.
    }
}

The pool must reject invalid sizes:

if (workerCount <= 0) {
    throw new IllegalArgumentException(
            "Worker count must be greater than zero"
    );
}

The constructor should create and start exactly workerCount threads. Every worker executes processTasks.

Name the workers so the output is understandable:

Thread worker = new Thread(
        this::processTasks,
        "manual-pool-worker-" + workerNumber
);

Remember to add each thread to workers before starting it.

The submit method must add a task to taskQueue. Use put() so you practice handling interruption.

try {
    // Add task
} catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
    throw new IllegalStateException("Task submission interrupted", ex);
}

Reject null tasks using Objects.requireNonNull.

Each worker must repeatedly:

1. call taskQueue.take();
2. wait if no task is available;
3. execute the returned task; and
4. return to the queue for another task.

At this stage, use an infinite loop. Controlled shutdown will be added next.

ManualThreadPool pool = new ManualThreadPool(4);

for (int i = 1; i <= 20; i++) {
    pool.submit(createSimulatedTask(i));
}

Run the program and inspect the output. You will need to stop this version manually because shutdown has not been implemented yet.

1. Are new threads created when tasks are submitted?
2. Can a worker execute more than one task?
3. Can two workers take the same queue entry?
4. What happens when all workers are busy?
5. What happens when the queue is empty?

• The constructor creates the requested number of workers.
• Submitted tasks are placed in the queue.
• Workers execute queued tasks.
• The program never uses more workers than configured.

The workers currently wait forever. You need a way to tell each worker to stop after all previously submitted tasks have finished.

Create one special singleton task:

private static final Runnable STOP_TASK = () -> { };

This object represents a stop signal. It is not ordinary work.

Immediately after a worker takes a task, check whether it is the singleton stop task:

if (task == STOP_TASK) {
    break;
}

This is a reference identity comparison, not a type check. It asks whether both variables refer to exactly the same object.

The shutdown() method must add one STOP_TASK for every worker.

for (int i = 0; i < workers.size(); i++) {
    // Add STOP_TASK to the queue.
}

Why is one stop task insufficient?

Because one queue entry can be taken by only one worker. If the pool has four workers, four stop tasks are needed.

Implement awaitTermination() by joining every worker:

for (Thread worker : workers) {
    try {
        worker.join();
    } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        throw new IllegalStateException(
                "Waiting for workers was interrupted",
                ex
        );
    }
}

Use the pool like this:

ManualThreadPool pool = new ManualThreadPool(4);

for (int i = 1; i <= 20; i++) {
    pool.submit(createSimulatedTask(i));
}

pool.shutdown();
pool.awaitTermination();

System.out.println("All tasks and workers have finished");

Call shutdown() only after all required tasks have been submitted. The poison pills are placed behind the existing tasks. A worker therefore processes earlier tasks before reaching a stop signal.

1. Why must the pool add one poison pill per worker?
2. Why does task == STOP_TASK work?
3. Is task == STOP_TASK a type check?
4. What could happen if an ordinary task is submitted after shutdown?
5. What does awaitTermination() make the main thread do?

• shutdown() adds one stop signal per worker.
• Workers finish previously queued tasks before stopping.
• awaitTermination() waits for every worker.
• The Java process exits normally.

Add a field:

private boolean shutdown;

Update submit() so it rejects tasks after shutdown:

if (shutdown) {
    throw new IllegalStateException("Thread pool has been shut down");
}

Set the flag in shutdown().

Could one thread call submit() while another thread calls shutdown()? If so, checking and changing a plain boolean is not enough to make the compound operation safe.

For the required solution, call submit() and shutdown() only from the main thread.

Try this task:

pool.submit(() -> {
    throw new IllegalStateException("Deliberate failure");
});

If task.run() is not protected, the exception terminates that worker. Change processTasks() so a failed task is reported but the worker continues:

try {
    task.run();
} catch (RuntimeException ex) {
    System.err.printf(
            "%s failed: %s%n",
            Thread.currentThread().getName(),
            ex.getMessage()
    );
}

Do not catch the exception around the whole worker loop. Think carefully about which operation should be allowed to fail without killing the worker.

Create an AtomicInteger to count active tasks and another to record the maximum:

// in main() or a test method
AtomicInteger activeTasks = new AtomicInteger();
AtomicInteger maximumActiveTasks = new AtomicInteger();

At the beginning of a test task (runnable), increment the active count and update the maximum:

int active = activeTasks.incrementAndGet();
maximumActiveTasks.accumulateAndGet(active, Math::max); // Math::max is a method reference that returns the larger of two ints. In this case, it compares the current maximum with the new active count and returns the larger of the two.

At the end:

activeTasks.decrementAndGet();

Run 20 tasks through a pool of four workers. The maximum should not exceed four.

Use try/finally so the active count is reduced even if a task fails.

try {
    Thread.sleep(500);
} catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
} finally {
    int remaining = activeTasks.decrementAndGet();
    System.out.printf( "Task %d finished. Active tasks: %d%n", taskNumber, remaining);
}

• A failed task does not permanently remove a worker.
• Later tasks still execute after a failure.
• A pool of four workers never reports more than four active tasks.
• Submission after shutdown is rejected.

Run the 20 simulated tasks with different worker counts.

1. Why does increasing the worker count reduce the duration of these simulated waiting tasks?
2. Would doubling the number of workers always halve the duration?
3. What overhead does concurrency introduce?
4. How might the result differ for CPU-intensive tasks?
5. Who should control the maximum number of simultaneous external requests?

Before involving Gemini, prove that the pool can execute HTTP tasks.

Use one shared HttpClient:

HttpClient client = HttpClient.newBuilder()
        .connectTimeout(Duration.ofSeconds(5))
        .build();

Create a task for each instructor-approved URL:

public static Runnable createHttpTask(
        HttpClient client,
        String url
) {
    return () -> {
        long start = System.currentTimeMillis();

        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(url))
                .timeout(Duration.ofSeconds(10))
                .GET()
                .build();

        try {
            HttpResponse<String> response = client.send(
                    request,
                    HttpResponse.BodyHandlers.ofString()
            );

            long duration = System.currentTimeMillis() - start;

            System.out.printf(
                    "%d  %d chars  %d ms  %s  %s%n",
                    response.statusCode(),
                    response.body().length(),
                    duration,
                    Thread.currentThread().getName(),
                    url
            );
        } catch (IOException ex) {
            throw new IllegalStateException(
                    "Request failed: " + url,
                    ex
            );
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(
                    "Request interrupted: " + url,
                    ex
            );
        }
    };
}

Submit at least eight HTTP tasks to a pool of three workers. Measure the total duration.

1. Why is one shared HttpClient preferable to one client per task?
2. Which part of an HTTP request normally involves waiting?
3. Why is a fixed concurrency limit considerate toward external servers?
4. Does an HTTP 404 necessarily mean that the Java task threw an exception?

In this part, each Runnable sends one prompt to Gemini. The thread pool controls how many requests are active at the same time.

Your instructor will provide the required model name and access configuration. The program expects these environment variables:

GEMINI_API_KEY
GEMINI_MODEL

Do not place the API key in source code or commit it to Git.

Read the variables:

String apiKey = System.getenv("GEMINI_API_KEY");
String model = System.getenv("GEMINI_MODEL");

if (apiKey == null || apiKey.isBlank()) {
    throw new IllegalStateException("GEMINI_API_KEY is missing");
}

if (model == null || model.isBlank()) {
    throw new IllegalStateException("GEMINI_MODEL is missing");
}

The current official request uses POST, an x-goog-api-key header and a JSON request body containing contents and parts.

import java.util.List;

public record GeminiRequest(List<RequestContent> contents) {
}

public record RequestContent(List<RequestPart> parts) {
}

public record RequestPart(String text) {
}

Create the request object:

GeminiRequest requestBody = new GeminiRequest(
        List.of(
                new RequestContent(
                        List.of(new RequestPart(prompt))
                )
        )
);

Convert it to JSON using Jackson:

String json = objectMapper.writeValueAsString(requestBody);

The response contains candidates, content and parts:

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import java.util.List;

@JsonIgnoreProperties(ignoreUnknown = true)
public record GeminiResponse(List<Candidate> candidates) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record Candidate(ResponseContent content) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record ResponseContent(List<ResponsePart> parts) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record ResponsePart(String text) {
}

The annotations allow Jackson to ignore response fields that the exercise does not need.

Create a class with this responsibility:

public class GeminiClient {

    private final HttpClient httpClient;
    private final ObjectMapper objectMapper;
    private final String apiKey;
    private final String model;

    public GeminiClient(
            HttpClient httpClient,
            ObjectMapper objectMapper,
            String apiKey,
            String model
    ) {
        this.httpClient = httpClient;
        this.objectMapper = objectMapper;
        this.apiKey = apiKey;
        this.model = model;
    }

    public String generateText(String prompt) {
        // TODO: Construct GeminiRequest.
        // TODO: Convert it to JSON.
        // TODO: Build and send the POST request.
        // TODO: Check the HTTP status code.
        // TODO: Convert the response JSON to GeminiResponse.
        // TODO: Return the first text part.
        return null;
    }
}

Build the endpoint from the supplied model name:

String endpoint =
        "https://generativelanguage.googleapis.com/v1beta/models/"
                + model
                + ":generateContent";

Build the HTTP request:

HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create(endpoint))
        .timeout(Duration.ofSeconds(30))
        .header("Content-Type", "application/json")
        .header("x-goog-api-key", apiKey)
        .POST(HttpRequest.BodyPublishers.ofString(json))
        .build();

Treat non-success status codes clearly:

if (response.statusCode() < 200
        || response.statusCode() >= 300) {
    throw new IllegalStateException(
            "Gemini returned HTTP " + response.statusCode()
    );
}

Do not include the API key in exception messages.

public static Runnable createGeminiTask(
        GeminiClient geminiClient,
        int promptNumber,
        String prompt
) {
    return () -> {
        long start = System.currentTimeMillis();

        String answer = geminiClient.generateText(prompt);

        long duration = System.currentTimeMillis() - start;

        System.out.printf(
                "%nPROMPT %d completed by %s in %d ms%n%s%n",
                promptNumber,
                Thread.currentThread().getName(),
                duration,
                answer
        );
    };
}

Create six to eight short prompts. For example:

List<String> prompts = List.of(
        "Explain encapsulation in Java in two sentences.",
        "Give one example of polymorphism in Java.",
        "Explain the difference between a process and a thread.",
        "Why is counter++ unsafe when shared by threads?",
        "Explain Thread.join() in two sentences.",
        "Give one reason not to create 10,000 platform threads."
);

Use a small pool:

ManualThreadPool pool = new ManualThreadPool(3);

for (int i = 0; i < prompts.size(); i++) {
    pool.submit(createGeminiTask(
            geminiClient,
            i + 1,
            prompts.get(i)
    ));
}

pool.shutdown();
pool.awaitTermination();

• At least six different prompts are submitted.
• No more than three Gemini tasks run simultaneously.
• Every task displays its worker name and duration.
• One failed request does not stop the remaining tasks.
• The API key is neither hard-coded nor printed.

Run the same prompts in two ways.

Call generateText() in an ordinary loop on the main thread.

Submit all prompt tasks to a manual thread pool with three workers.

Record the results:

Answer the following:

1. Which version completed faster?
2. Why does concurrency help with HTTP calls?
3. Why is the concurrent duration not simply the sequential duration divided by three?
4. Which external factors affect the measurements?
5. Could sending more requests concurrently trigger throttling or rate limits?
6. Why should pool size be configurable?

API response time and rate limits can vary. A faster result is likely but is not guaranteed. The important result is that your application limits and manages concurrent work correctly.

Your project should contain responsibilities similar to these:

src/main/java/
├── Main.java
├── ManualThreadPool.java
├── GeminiClient.java
├── GeminiRequest.java
├── RequestContent.java
├── RequestPart.java
├── GeminiResponse.java
├── Candidate.java
├── ResponseContent.java
└── ResponsePart.java

You may organize the records differently, including nesting closely related records, as long as the design remains readable.

Your required solution must satisfy all of the following:

• It uses a BlockingQueue<Runnable>.
• It creates a fixed number of worker threads.
• Workers are created once and reused for multiple tasks.
• submit() places tasks in the queue.
• An empty queue causes workers to wait rather than busy-wait.
• The pool uses one poison pill per worker.
• shutdown() stops accepting new work.
• Previously submitted work completes before workers stop.
• awaitTermination() joins all worker threads.
• A task throwing a runtime exception does not terminate its worker.
• The measured concurrency never exceeds the worker count.
• The pool runs multiple HTTP tasks.
• The final program sends multiple Gemini prompts.
• The Gemini API key is read from an environment variable.
• The API key is never committed or printed.
• The program reports HTTP and task failures clearly.
• The program does not use ExecutorService in the required solution.

1. What is the difference between a Runnable and a worker thread?
2. Why does the queue need to be thread-safe?
3. What does BlockingQueue.take() do when the queue is empty?
4. Why does the pool reuse its worker threads?
5. Why is one poison pill required for each worker?
6. Why is task == STOP_TASK an identity comparison rather than a type check?
7. Why must task exceptions be caught inside the worker loop?
8. What could happen if tasks are submitted after poison pills?
9. Why are HTTP and LLM requests suitable for concurrent execution?
10. What weaknesses or missing features can you identify in your pool?
11. Which responsibilities would you expect Java’s ExecutorService to handle?
12. Would you use this manual implementation in production? Explain your answer.

• Gemini API: Text generation
• Gemini API: GenerateContent reference