apache/kafka · error · IllegalArgumentException

Invalid partition given with record: ${partition} is not in

Error message

Invalid partition given with record: ${partition} is not in the range [0...${numPartitions}].

What it means

Thrown by MockProducer.partition(ProducerRecord, Cluster) as an IllegalArgumentException when a ProducerRecord carries an explicit partition index that is not within [0, numPartitions) for the topic as known to the mock's Cluster. The mock computes the partition only when the record already specifies one (record.partition() != null); if it is negative or >= the topic's partition count from cluster.partitionsForTopic(topic).size(), the throw fires at line 641. It mirrors the real producer's validation so partition-targeting bugs surface in tests.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java:641

            completion.complete(e);
            return true;
        } else {
            return false;
        }
    }

    /**
     * computes partition for given record.
     */
    private int partition(ProducerRecord<K, V> record, Cluster cluster) {
        Integer partition = record.partition();
        String topic = record.topic();
        if (partition != null) {
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int numPartitions = partitions.size();
            // they have given us a partition, use it
            if (partition < 0 || partition >= numPartitions)
                throw new IllegalArgumentException("Invalid partition given with record: " + partition
                                                   + " is not in the range [0..."
                                                   + numPartitions
                                                   + "].");
            return partition;
        }
        byte[] keyBytes = keySerializer.serialize(topic, record.headers(), record.key());
        byte[] valueBytes = valueSerializer.serialize(topic, record.headers(), record.value());
        if (partitioner == null) {
            return this.cluster.partitionsForTopic(record.topic()).get(0).partition();
        }
        return this.partitioner.partition(topic, record.key(), keyBytes, record.value(), valueBytes, cluster);
    }

    private static class Completion {
        private final long offset;
        private final RecordMetadata metadata;
        private final ProduceRequestResult result;
        private final Callback callback;

View on GitHub (pinned to c31c9215e1)

Solutions

If you send records with explicit partitions, construct the MockProducer with a Cluster that advertises enough partitions for the topic (use Cluster.empty().append(...) or a helper that builds PartitionInfo).
If you do not need a specific partition, build ProducerRecord without the partition argument and let the partitioner/round-robin pick one.
Validate the partition against the cluster's partition count in production code before sending, surfacing the mismatch as a recoverable error.
Update the mock's Cluster setup whenever the real topic's partition count changes so the test reflects production.

Example fix

// before
MockProducer<String,String> p = new MockProducer<>(); // Cluster.empty()
p.send(new ProducerRecord<>("orders", 1, k, v)); // IllegalArgumentException

// after
Cluster c = Cluster.empty().withPartitions(
    Map.of(new TopicPartition("orders", 0), null,
           new TopicPartition("orders", 1), null));
MockProducer<String,String> p = new MockProducer<>(c, true, null, new StringSerializer(), new StringSerializer());
p.send(new ProducerRecord<>("orders", 1, k, v));

Defensive patterns

Strategy: validation

Validate before calling

import org.apache.kafka.common.PartitionInfo;
import java.util.List;

void safeSend(MockProducer<K,V> p, ProducerRecord<K,V> r) {
    Integer part = r.partition();
    if (part != null) {
        List<PartitionInfo> parts = p.partitionsFor(r.topic());
        int n = parts == null ? 0 : parts.size();
        if (part < 0 || part >= n) {
            throw new IllegalArgumentException(
                "Refusing send: partition " + part + " out of [0.." + n + ")");
        }
    }
    p.send(r);
}

Prevention

If you set ProducerRecord.partition(), always validate it against the current cluster metadata before send.
Prefer letting the partitioner choose; only pin a partition when you have authoritative knowledge of partition count.
Refresh and cache partitionsFor(topic) when topics are created/expanded; stale counts cause this error.
In tests, configure the MockProducer's Cluster with the exact partition count you intend to use.

When it happens

Trigger: Sending a ProducerRecord constructed with an explicit partition (e.g. new ProducerRecord<>(topic, partition, key, value)) where partition < 0 or partition >= number of partitions the mock's Cluster reports for that topic. The mock Cluster defaults to Cluster.empty() (zero partitions for every topic) unless a Cluster with partition metadata was supplied to the constructor.

Common situations: Using the default no-arg MockProducer() (Cluster.empty()) and sending a record with an explicit partition — there are zero partitions so any index is out of range; a test that hardcodes partition=2 against a cluster mock with fewer partitions; a topic-partition count change in production metadata not reflected in the mock cluster setup.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/049d46794e4bad8d.json. Report an issue: GitHub.