Skip to content

Commit a064f78

Browse files
sjarmakclaude
sjarmak
and
claude
committed
fix: coerce read_file start_line/end_line to int in tool dispatch
Haiku sometimes passes string values (e.g., "[430, 475]") for numeric tool parameters. Add _to_int() coercion to prevent TypeError on comparison. Includes 1-task pilot output (kafka-producer-bufpool-fix-001). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 98d304a commit a064f78

File tree

3 files changed

+155
-2
lines changed

3 files changed

+155
-2
lines changed

‎benchmarks/ccb_fix/kafka-producer-bufpool-fix-001/tests/ground_truth_agent.json‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
{
2+
"files": [
3+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
4+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java",
5+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java",
6+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java",
7+
"clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java"
8+
],
9+
"symbols": [
10+
{
11+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
12+
"symbol": "Sender.sendProducerData",
13+
"repo": null
14+
},
15+
{
16+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
17+
"symbol": "Sender.failBatch",
18+
"repo": null
19+
},
20+
{
21+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
22+
"symbol": "Sender.maybeRemoveAndDeallocateBatch",
23+
"repo": null
24+
},
25+
{
26+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
27+
"symbol": "Sender.sendProduceRequest",
28+
"repo": null
29+
},
30+
{
31+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java",
32+
"symbol": "ProducerBatch.records",
33+
"repo": null
34+
},
35+
{
36+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java",
37+
"symbol": "BufferPool.deallocate",
38+
"repo": null
39+
},
40+
{
41+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java",
42+
"symbol": "RecordAccumulator.deallocate",
43+
"repo": null
44+
}
45+
],
46+
"expected_edit_files": [
47+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
48+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java",
49+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java"
50+
],
51+
"expected_edit_files_source": "curator_agent_inference",
52+
"expected_edit_files_confidence": "medium",
53+
"chunks": [
54+
{
55+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
56+
"line_start": 357,
57+
"line_end": 443,
58+
"annotation": "sendProducerData(): Drains batches, checks for expired batches, and sends produce requests. Critical section where batches are added to in-flight and then immediately sent, but expiration/deallocation can occur before network write completes."
59+
},
60+
{
61+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
62+
"line_start": 405,
63+
"line_end": 423,
64+
"annotation": "getExpiredInflightBatches() and expiredBatches() are called AFTER sendProduceRequests() collates batches but BEFORE actually sending them via client.send(). This creates the race condition window."
65+
},
66+
{
67+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
68+
"line_start": 817,
69+
"line_end": 837,
70+
"annotation": "failBatch() with maybeRemoveAndDeallocateBatch() at line 835 - This deallocates the batch buffer while the batch may still have in-flight requests in the network layer."
71+
},
72+
{
73+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
74+
"line_start": 172,
75+
"line_end": 175,
76+
"annotation": "maybeRemoveAndDeallocateBatch() - Calls accumulator.deallocate() which returns the ByteBuffer to the pool immediately, even though the request may still be serializing/sending on the network."
77+
},
78+
{
79+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java",
80+
"line_start": 864,
81+
"line_end": 916,
82+
"annotation": "sendProduceRequest() - Extracts MemoryRecords from batch at line 874, which holds a reference to the underlying ByteBuffer. The request is sent but the reference to the buffer is not retained."
83+
},
84+
{
85+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java",
86+
"line_start": 480,
87+
"line_end": 482,
88+
"annotation": "records() method - Returns MemoryRecords wrapping the ByteBuffer from recordsBuilder. This is the object used in the network request, and its underlying buffer can be deallocated while the request is in-flight."
89+
},
90+
{
91+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java",
92+
"line_start": 540,
93+
"line_end": 542,
94+
"annotation": "buffer() method - Exposes the underlying ByteBuffer that can be deallocated. ProducerBatch needs to track if its buffer is in-flight."
95+
},
96+
{
97+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java",
98+
"line_start": 260,
99+
"line_end": 275,
100+
"annotation": "deallocate() method - Returns buffers to the free pool for reuse. Called while the original batch's request may still be in-flight, causing buffer reuse corruption."
101+
},
102+
{
103+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java",
104+
"line_start": 264,
105+
"line_end": 265,
106+
"annotation": "buffer.clear() and free.add(buffer) - The buffer is immediately available for reuse, creating the race condition window."
107+
},
108+
{
109+
"file": "clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java",
110+
"line_start": 1027,
111+
"line_end": 1033,
112+
"annotation": "deallocate() method - Calls free.deallocate() which immediately makes the buffer available for reuse, without checking if the request is still in-flight."
113+
}
114+
],
115+
"dependency_chain": [
116+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:sendProducerData()",
117+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:sendProduceRequests()",
118+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:sendProduceRequest()",
119+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java:records()",
120+
"clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java",
121+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:getExpiredInflightBatches()",
122+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:failBatch()",
123+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:maybeRemoveAndDeallocateBatch()",
124+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:deallocate()",
125+
"clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:deallocate()"
126+
]
127+
}

‎benchmarks/ccb_fix/kafka-producer-bufpool-fix-001/tests/ground_truth_meta.json‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
{
2+
"has_ground_truth": true,
3+
"has_chunk_ground_truth": true,
4+
"ground_truth_source": "curator_agent",
5+
"ground_truth_confidence": "high",
6+
"task_name": "kafka-producer-bufpool-fix-001",
7+
"curator_agent_version": "2.0",
8+
"model": "claude-haiku-4-5-20251001",
9+
"backend": "local",
10+
"timestamp": "2026-03-02T16:43:59.394149+00:00",
11+
"files_count": 5,
12+
"edit_files_count": 3,
13+
"chunks_count": 10,
14+
"symbols_count": 7,
15+
"cost_usd": 0.7242,
16+
"elapsed_sec": 80.4,
17+
"exploration_notes": "ROOT CAUSE ANALYSIS: This is a race condition in the Kafka producer's buffer pool memory management. The bug occurs when an in-flight ProducerBatch expires or its broker disconnects before the network layer completes serialization and transmission of the produce request. The execution flow is: (1) In sendProducerData(), batches are drained and added to inFlightBatches, (2) sendProduceRequests() creates ProduceRequest objects containing MemoryRecords that reference the batch's underlying ByteBuff"
18+
}

‎scripts/context_retrieval_agent.py‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -656,10 +656,18 @@ def dispatch_tool(
656656
input_args.get("cwd", ""),
657657
)
658658
elif name == "read_file":
659+
# Coerce start_line/end_line to int — models sometimes pass strings
660+
def _to_int(val: Any, default: int = 0) -> int:
661+
if isinstance(val, int):
662+
return val
663+
try:
664+
return int(val)
665+
except (TypeError, ValueError):
666+
return default
659667
return execute_read_file(
660668
input_args.get("path", ""),
661-
input_args.get("start_line", 0),
662-
input_args.get("end_line", 0),
669+
_to_int(input_args.get("start_line", 0)),
670+
_to_int(input_args.get("end_line", 0)),
663671
)
664672
elif name == "list_directory":
665673
return execute_list_directory(

0 commit comments

Comments
 (0)