kafka-go Fetch response negative record count panic

package compress

import (
	"encoding"
	"fmt"
	"io"
	"strconv"
)

type Compression int8

const (
	None   Compression = 0
	Gzip   Compression = 1
	Snappy Compression = 2
	Lz4    Compression = 3
	Zstd   Compression = 4
)

func (c Compression) Codec() Codec {
	if i := int(c); i >= 0 && i < len(Codecs) {
		return Codecs[i]
	}
	return nil
}

func (c Compression) String() string {
	if c == None {
		return "uncompressed"
	}
	return strconv.Itoa(int(c))
}

func (c Compression) MarshalText() ([]byte, error) { return []byte(c.String()), nil }

func (c *Compression) UnmarshalText(b []byte) error {
	if string(b) == "none" || string(b) == "uncompressed" {
		*c = None
		return nil
	}
	i, err := strconv.ParseInt(string(b), 10, 64)
	if err == nil && i >= 0 && i < int64(len(Codecs)) {
		*c = Compression(i)
		return nil
	}
	return fmt.Errorf("compression format must be none or numeric, not %q", b)
}

type Codec interface {
	Code() int8
	Name() string
	NewReader(io.Reader) io.ReadCloser
	NewWriter(io.Writer) io.WriteCloser
}

var (
	_ encoding.TextMarshaler   = Compression(0)
	_ encoding.TextUnmarshaler = (*Compression)(nil)

	Codecs = [...]Codec{
		None:   nil,
		Gzip:   nil,
		Snappy: nil,
		Lz4:    nil,
		Zstd:   nil,
	}
)

#!/usr/bin/env bash
set -euo pipefail

script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
source_dir="$script_dir/source"

UPSTREAM_REPO_URL="https://github.com/segmentio/kafka-go"
UPSTREAM_PINNED_SHA="2e0b3968aa51b16beb4e221876499a6ff816cd91"

_silent() {
  local label="$1"; shift
  local log="$script_dir/.${label}.log"
  if ! "$@" > "$log" 2>&1; then
    echo "[run.sh] $label failed; log follows:" >&2
    cat "$log" >&2
    return 1
  fi
}

setup() {
  if [[ -d "$source_dir/.git" ]] \
     && [[ "$(git -C "$source_dir" rev-parse HEAD 2>/dev/null)" == "$UPSTREAM_PINNED_SHA" ]]; then
    echo "[run.sh] reusing existing $source_dir at $UPSTREAM_PINNED_SHA"
    return 0
  fi
  _setup_inner() {
    rm -rf "$source_dir"
    mkdir -p "$source_dir"
    cd "$source_dir"
    git init -q
    git remote add origin "$UPSTREAM_REPO_URL"
    if git fetch --depth 1 origin "$UPSTREAM_PINNED_SHA" -q 2>/dev/null; then
      git checkout -q FETCH_HEAD
    else
      cd "$script_dir"
      rm -rf "$source_dir"
      git clone -q "$UPSTREAM_REPO_URL" "$source_dir"
      git -C "$source_dir" checkout -q "$UPSTREAM_PINNED_SHA"
    fi
    local head
    head="$(git -C "$source_dir" rev-parse HEAD)"
    if [[ "$head" != "$UPSTREAM_PINNED_SHA" ]]; then
      echo "setup: HEAD=$head but expected $UPSTREAM_PINNED_SHA" >&2
      exit 1
    fi
  }
  _silent setup _setup_inner
  # One-line confirmation reaches the terminal — mirrors the
  # 'Pinned ref:' line in the report README so the maintainer can see
  # exactly which commit they're exercising before the proof emits.
  echo "[run.sh] reproducing against $UPSTREAM_REPO_URL @ $UPSTREAM_PINNED_SHA"
}

build() {
  _silent build bash -c '
    set -euo pipefail
    scratch="$1/inputs/build-scratch/kafka-go"
    rm -rf "$scratch"
    mkdir -p "$(dirname "$scratch")"
    cp -a "$2" "$scratch"
    sed -i "s/^go 1\\.23.*/go 1.22.2/" "$scratch/go.mod"
    cp "$1/inputs/compress_stub.go" "$scratch/compress/compress.go"
    cd "$1/inputs/poc"
    GOTOOLCHAIN=local go build -o "$1/inputs/kafka-go-negative-record-count-poc" .
  ' _ "$script_dir" "$source_dir"
}

trigger() {
  set +e
  "$script_dir/inputs/kafka-go-negative-record-count-poc" > "$script_dir/.trigger.log" 2>&1
  status=$?
  set -e
  if [[ "$status" -eq 0 ]]; then
    echo "trigger: PoC returned successfully; expected panic" >&2
    cat "$script_dir/.trigger.log" >&2
    exit 1
  fi
  grep -F "panic: runtime error: makeslice: len out of range" "$script_dir/.trigger.log" \
    || { echo "trigger: no reproduction proof in run output" >&2; cat "$script_dir/.trigger.log" >&2; exit 1; }
}

cmd="${1:-all}"
case "$cmd" in
  setup)         setup ;;
  build)         setup; build ;;
  trigger|all|"") setup; build; trigger ;;
  *)
    echo "usage: $0 {setup|build|trigger|all}" >&2
    exit 2
    ;;
esac

bash ./poc/run.sh

panic: runtime error: makeslice: len out of range

diff -ruN '--exclude=.git' a/protocol/record_v2.go b/protocol/record_v2.go
--- a/protocol/record_v2.go	2026-06-11 19:41:40.381844166 +0000
+++ b/protocol/record_v2.go	2026-06-18 07:51:45.421110964 +0000
@@ -66,6 +66,10 @@
 	dec.reader = buffer
 	dec.remain = recordsLength
 
+	if numRecords < 0 {
+		return Errorf("invalid record batch with negative record count (%d)", numRecords)
+	}
+
 	records := make([]optimizedRecord, numRecords)
 	// These are two lazy allocators that will be used to optimize allocation of
 	// page references for keys and values.