apache/kafka · error · IllegalArgumentException

Invalid message timestamp {}

Error message

Invalid message timestamp {}

What it means

Thrown by DefaultRecordBatch.writeHeader when baseTimestamp is negative and not equal to NO_TIMESTAMP (-1L). The v2 batch header reserves 8 bytes for baseTimestamp; the only permitted non-negative values are real epoch-millis timestamps, with NO_TIMESTAMP as the special 'unset' sentinel. Any other negative value indicates a caller bug. IllegalArgumentException.

Source

Thrown at clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java:480

                                   int lastOffsetDelta,
                                   int sizeInBytes,
                                   byte magic,
                                   CompressionType compressionType,
                                   TimestampType timestampType,
                                   long baseTimestamp,
                                   long maxTimestamp,
                                   long producerId,
                                   short epoch,
                                   int sequence,
                                   boolean isTransactional,
                                   boolean isControlBatch,
                                   boolean isDeleteHorizonSet,
                                   int partitionLeaderEpoch,
                                   int numRecords) {
        if (magic < RecordBatch.CURRENT_MAGIC_VALUE)
            throw new IllegalArgumentException("Invalid magic value " + magic);
        if (baseTimestamp < 0 && baseTimestamp != NO_TIMESTAMP)
            throw new IllegalArgumentException("Invalid message timestamp " + baseTimestamp);

        short attributes = computeAttributes(compressionType, timestampType, isTransactional, isControlBatch, isDeleteHorizonSet);

        int position = buffer.position();
        buffer.putLong(position + BASE_OFFSET_OFFSET, baseOffset);
        buffer.putInt(position + LENGTH_OFFSET, sizeInBytes - LOG_OVERHEAD);
        buffer.putInt(position + PARTITION_LEADER_EPOCH_OFFSET, partitionLeaderEpoch);
        buffer.put(position + MAGIC_OFFSET, magic);
        buffer.putShort(position + ATTRIBUTES_OFFSET, attributes);
        buffer.putLong(position + BASE_TIMESTAMP_OFFSET, baseTimestamp);
        buffer.putLong(position + MAX_TIMESTAMP_OFFSET, maxTimestamp);
        buffer.putInt(position + LAST_OFFSET_DELTA_OFFSET, lastOffsetDelta);
        buffer.putLong(position + PRODUCER_ID_OFFSET, producerId);
        buffer.putShort(position + PRODUCER_EPOCH_OFFSET, epoch);
        buffer.putInt(position + BASE_SEQUENCE_OFFSET, sequence);
        buffer.putInt(position + RECORDS_COUNT_OFFSET, numRecords);
        long crc = Crc32C.compute(buffer, ATTRIBUTES_OFFSET, sizeInBytes - ATTRIBUTES_OFFSET);
        buffer.putInt(position + CRC_OFFSET, (int) crc);

View on GitHub (pinned to c31c9215e1)

Solutions

1. Pass a non-negative epoch-millis baseTimestamp, or RecordBatch.NO_TIMESTAMP (-1L) if you have no meaningful base time.
2. Guard subtractions that compute deltas: clamp the base so baseTimestamp >= 0 before calling writeHeader.
3. Sanitize inbound timestamps (e.g. from external systems) to either a valid millis value or NO_TIMESTAMP before producing.

Example fix

// before
long base = firstRecordTs - clockSkewBase; // can go negative
writeHeader(buf, ..., base, ...);

// after
long base = firstRecordTs - clockSkewBase;
if (base < 0) base = RecordBatch.NO_TIMESTAMP;
writeHeader(buf, ..., base, ...);

// Timestamp is user-supplied via ProducerRecord or batch writer. Validate before send:
long requireValidTimestamp(long ts) {
    if (ts < 0 && ts != RecordBatch.NO_TIMESTAMP)
        throw new IllegalArgumentException("timestamp must be >= 0 or RecordBatch.NO_TIMESTAMP, got " + ts);
    return ts;
}
ProducerRecord<K,V> safe(String t, K k, V v, long ts) {
    return new ProducerRecord<>(t, null, requireValidTimestamp(ts), k, v);
}

// Box the timestamp in a value type that cannot hold an invalid value:
static final class RecordTimestamp {
    private final long ms;
    private RecordTimestamp(long ms) { this.ms = ms; }
    static RecordTimestamp now() { return new RecordTimestamp(System.currentTimeMillis()); }
    static RecordTimestamp of(long in) {
        if (in < 0 && in != RecordBatch.NO_TIMESTAMP) throw new IllegalArgumentException("bad ts");
        return new RecordTimestamp(in);
    }
    long millis() { return ms; }
}

try {
    producer.send(new ProducerRecord<>(topic, ts, k, v));
} catch (IllegalArgumentException e) { // invalid timestamp
    log.warn("Bad timestamp {}; sending with broker-assigned time", ts, e);
    producer.send(new ProducerRecord<>(topic, k, v)); // no timestamp → CREATE_TIME/broker default
}

When it happens

Trigger: Produced when writeHeader is called with a baseTimestamp such as -2 or Long.MIN_VALUE — e.g. a producer/test computing baseTimestamp as (recordTs - earliestTs) that underflows, or code forwarding an uninitialized timestamp field. The check at DefaultRecordBatch.java:479-480 fires before any bytes are written.

Common situations: Tests that pass arbitrary negative numbers for timestamps; clock-skew handling code that subtracts a larger base from a smaller record timestamp; serializing a record whose timestamp was loaded from a corrupt source without sanitization; mocking frameworks that default numeric fields to negative sentinels.

Related errors

• Timestamp type must be provided to compute attributes for me
• Invalid timestamp: %d. Timestamp should always be non-negati
• Invalid magic value {}
• Invalid configuration value for 'acks': {acksString}
• Topic cannot be null.

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