Lab 03 — Patch Archaeology: Retrospective Review

Phase: 06 — Review Culture and Patch Workflow
Difficulty: ⭐⭐⭐⭐☆ | Estimated Time: 2–2.5 hours
Primary Artifact: kafka-pr-17873.patch — a synthetic Kafka-style patch
Deliverable: kafka-pr-17873-review.md — your retrospective review with 4+ findings

Scenario

This patch was merged 6 months ago. A bug was filed last week:

KAFKA-21044 — RequestQueue.drainRequests() occasionally throws NullPointerException in production under high load. Stack trace points to line 89 of RequestQueue.java.

You're doing a post-mortem. Your job: read the patch that introduced drainRequests() and write the review that should have caught the bug before it merged.

What is Patch Archaeology?

When a bug reaches production:

1. Find the commit that introduced it: git log --all -S 'drainRequests' -- RequestQueue.java
2. Read the diff as if you were the reviewer at the time of the PR
3. Write the review you would have written — with the benefit of hindsight, but using only the information that was available when the PR was open (no knowledge of the production failure)

This is a standard Apache post-mortem skill. The goal is not to blame the author; it is to identify what review standards would have caught the issue, and whether to add a rule to the review checklist.

Setup

cd phase-06-review-culture/lab-03-patch-archaeology
cat kafka-pr-17873.patch

Read the patch file. The format is standard git diff:

• Lines starting with --- / +++ are the old/new file
• Lines starting with @@ are hunk headers showing line numbers
• Lines starting with - were removed (red in GitHub)
• Lines starting with + were added (green in GitHub)
• Lines with no prefix are context (unchanged)

Steps

Step 1 — Read the diff without the answer key

Read kafka-pr-17873.patch completely. For each concern you notice, write it down with:

• The line number (from the @@ header + offset)
• A one-sentence problem statement
• The severity label

There are 4 issues. Find as many as you can before expanding the answer.

Step 2 — Write the retrospective review

Create a file kafka-pr-17873-review.md in this directory with the following structure:

# Retrospective Review — KAFKA-PR-17873

**Reviewer**: [your name]
**Review date**: [today]
**Linked bug**: KAFKA-21044

## Summary

[2–3 sentence overview: what the patch does, and what slipped through]

## Findings

### Finding 1 — [Title]
**Severity**: BLOCKER
**Location**: `RequestQueue.java`, line [N]

[Problem statement]

**Evidence** (from the diff):
```java
[relevant lines]

Why it wasn't caught: [hypothesis — what review standard would have found this?]

Fix:

[suggested fix]

[Repeat for each finding]

Process Notes

[What review checklist item, if any, should be added to prevent this class of issue in future patches? 1–3 sentences.]

### Step 3 — Verify your review matches the actual bug

The production bug (KAFKA-21044) is the NPE on `requests.get(nodeId)` returning null when the node has no pending requests (Issue #2 in the answer key). Your review should have found this as a BLOCKER with the label "missing null check on return value of Map.get()".

If your review found it: good. If not: read the hunk again and note the exact line. Add it to your review.

---

## Key Insight

The bug was not subtle. `Map.get()` returning null when the key is absent is documented behavior. The reviewer should have asked: "What happens when `nodeId` is in the map but has an empty list? What happens when `nodeId` is not in the map at all?"

The answer to "What happens when nodeId is not in the map?" is: `requests.get(nodeId)` returns null, and `for (Request r : null)` throws `NullPointerException`. This is a one-question review that would have caught a production outage.

This is why Apache projects require: **every new method that takes user input must have a test for the empty/null/missing case**.

---

## Source Files

### `kafka-pr-17873.patch`

```diff
From 3f8a92b1c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9 Mon Sep 17 00:00:00 2001
From: alex.contributor <alex.contributor@example.com>
Date: Mon, 14 Oct 2024 11:22:03 +0000
Subject: [KAFKA-17873] RequestQueue: add drainRequests() for batch processing

Adds drainRequests(nodeId, callback) which processes all pending requests for
a given node in a single pass, invoking the callback for each. This replaces
the existing poll-in-a-loop pattern in NetworkClient, which had O(n) individual
synchronized calls.

Also fixes a minor issue where drainAll() was not returning the count of
drained requests, making it impossible for callers to distinguish "nothing
to drain" from "drained successfully."

Tested locally with NetworkClientTest — all existing tests pass.

Reviewers: b.reviewer@example.com
---
 .../kafka/clients/NetworkClient.java          |  18 ++-
 .../kafka/clients/RequestQueue.java           |  71 +++++++--
 .../kafka/clients/RequestQueueTest.java       |  29 +++-
 3 files changed, 104 insertions(+), 14 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/RequestQueue.java b/clients/src/main/java/org/apache/kafka/clients/RequestQueue.java
index a1b2c3d..f4e5c6d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/RequestQueue.java
+++ b/clients/src/main/java/org/apache/kafka/clients/RequestQueue.java
@@ -1,7 +1,7 @@
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
+ * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -18,6 +18,7 @@ package org.apache.kafka.clients;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -42,6 +43,14 @@ public class RequestQueue {
     private final Map<String, List<Request>> requests = new HashMap<>();
     private final int maxInFlight;
 
+    /**
+     * Callback invoked for each request drained from the queue.
+     */
+    public interface DrainCallback {
+        /**
+         * @param nodeId  the node the request was queued for
+         * @param request the drained request
+         */
+        void onDrain(String nodeId, Request request);
+    }
+
     public RequestQueue(int maxInFlight) {
         if (maxInFlight <= 0)
             throw new IllegalArgumentException("maxInFlight must be > 0");
@@ -68,16 +77,52 @@ public class RequestQueue {
     }
 
     /**
-     * Drains all pending requests across all nodes, returning the total count.
+     * Drains all pending requests across all nodes.
+     *
+     * @return count of requests drained
      */
-    public synchronized void drainAll() {
+    public synchronized int drainAll() {
+        int count = 0;
         for (Map.Entry<String, List<Request>> entry : requests.entrySet()) {
-            entry.getValue().clear();
+            count += entry.getValue().size();
+            entry.getValue().clear();
         }
+        return count;
     }
 
-    /** Returns the number of pending requests for the given node. */
-    public synchronized int pendingCount(String nodeId) {
-        List<Request> list = requests.get(nodeId);
-        return list == null ? 0 : list.size();
+    /**
+     * Drains all pending requests for a specific node, invoking the callback
+     * for each request before removing it from the queue.
+     *
+     * @param nodeId   the target node identifier
+     * @param callback invoked for each drained request
+     * @return the number of requests drained
+     */
+    public synchronized int drainRequests(String nodeId, DrainCallback callback) {
+        // ISSUE 3 — API regression: this method is declared on the RequestQueue
+        // interface (not shown in this diff) as 'void drainRequests(...)'.
+        // Changing the return type to 'int' breaks binary compatibility with
+        // callers compiled against the interface. The interface was not updated
+        // in this patch. Any caller using the interface type will get:
+        //   AbstractMethodError: RequestQueue.drainRequests(...)V
+        // at runtime. (The 'V' void descriptor is no longer present.)
+        //
+        // A reviewer should have checked: does this method appear on an interface?
+        // The answer is: yes, IRequestQueue.java — not included in this diff.
+
+        List<Request> pending = new ArrayList<>();
+        drain(nodeId, pending, callback);
+        return pending.size();
     }
 
+    private void drain(String nodeId, List<Request> out, DrainCallback callback) {
+        // ISSUE 1 — nodeId is never null-checked.
+        // If nodeId is null, requests.get(null) is valid (HashMap allows null keys)
+        // but only if null was ever used as a key. In practice, nodeIds are always
+        // non-null strings. The missing guard means a null nodeId silently produces
+        // an empty result rather than a fast-fail IllegalArgumentException.
+        // This is a MAJOR — not a BLOCKER — because it produces a wrong result
+        // (silent empty drain) rather than an exception in all cases.
+
+        // ISSUE 2 — THE PRODUCTION BUG
+        // requests.get(nodeId) returns null when nodeId is not in the map
+        // (i.e., no requests have ever been queued for this node, or all were
+        // previously drained). Iterating over null with an enhanced for loop:
+        //
+        //   for (Request r : null) { ... }
+        //
+        // throws NullPointerException. The fix is a single null check:
+        //
+        //   List<Request> list = requests.get(nodeId);
+        //   if (list == null) return;
+        //
+        // This is the bug that caused KAFKA-21044. It was not caught because
+        // there was no test for drainRequests() on a node with no pending requests.
+        for (Request r : requests.get(nodeId)) {
+            callback.onDrain(nodeId, r);
+            out.add(r);
+        }
+        requests.remove(nodeId);
+    }
+
+    /** Returns the number of pending requests for the given node. */
+    public synchronized int pendingCount(String nodeId) {
+        List<Request> list = requests.get(nodeId);
+        return list == null ? 0 : list.size();
+    }
 }
diff --git a/clients/src/test/java/org/apache/kafka/clients/RequestQueueTest.java b/clients/src/test/java/org/apache/kafka/clients/RequestQueueTest.java
index b2c3d4e..c5d6e7f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/RequestQueueTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/RequestQueueTest.java
@@ -52,6 +52,35 @@ public class RequestQueueTest {
         assertEquals(0, queue.pendingCount("node-1"));
     }
 
+    @Test
+    public void testDrainRequests_callsCallbackForEachRequest() {
+        RequestQueue queue = new RequestQueue(10);
+        queue.enqueue("node-1", new Request("req-1"));
+        queue.enqueue("node-1", new Request("req-2"));
+        queue.enqueue("node-1", new Request("req-3"));
+
+        List<String> drained = new ArrayList<>();
+        int count = queue.drainRequests("node-1",
+            (nodeId, req) -> drained.add(req.id()));
+
+        assertEquals(3, count);
+        assertEquals(3, drained.size());
+        assertEquals(0, queue.pendingCount("node-1"));
+    }
+
+    @Test
+    public void testDrainRequests_returnsZeroForNodeWithNoRequests() {
+        // ISSUE 4 — THIS TEST IS MISSING FROM THE PATCH.
+        //
+        // This comment is added here to show WHERE the missing test should have been.
+        // The PR added testDrainRequests_callsCallbackForEachRequest (above) but
+        // did NOT add a test for the case where the node has no pending requests.
+        //
+        // A reviewer who noticed "what happens when nodeId is not in the map?"
+        // would have requested this test. It would have failed immediately,
+        // revealing the NullPointerException before the patch was merged.
+        //
+        // Fix: add this test, then fix the drain() method to guard against
+        // requests.get(nodeId) returning null.
+        //
+        // RequestQueue queue = new RequestQueue(10);
+        // int count = queue.drainRequests("node-with-no-requests", (n, r) -> {});
+        // assertEquals(0, count);  // this line throws NPE before the fix
+    }
+
 }
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index d3e4f5a..e6f7g8h 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -187,12 +187,14 @@ public class NetworkClient {
 
     private void processCompletedReceives(String nodeId) {
-        // Old pattern: poll in a loop with individual synchronized calls
-        while (requestQueue.pendingCount(nodeId) > 0) {
-            Request r = requestQueue.poll(nodeId);
-            if (r != null) handleResponse(nodeId, r);
-        }
+        // New pattern: single drain call — fewer lock acquisitions
+        requestQueue.drainRequests(nodeId,
+            (nId, req) -> handleResponse(nId, req));
     }
 
 }