Kafkajs: KafkaJSProtocolError: The broker received an out of order sequence number

[Delapouite]

Hi

I have a single producer:

const producer = kafka.producer({
  allowAutoTopicCreation: false,
  idempotent: true,
  maxInFlightRequests: 1
});

As you can see, I'm using the idempotent feature introduced in https://github.com/tulios/kafkajs/pull/203 I'm also forcing the maxInFlightRequests to 1

It works fine when my producer sends messages to the kafka server at a slow pace. By that I mean no requests are stored in the RequestQueue.

But during spike in activity, sometimes messages are automatically sent in batch.
Kafka then returns the following error:

ERROR [ReplicaManager broker=0] Error processing append operation on partition TOPIC (kafka.server.ReplicaManager)
kafka_1                 | org.apache.kafka.common.errors.OutOfOrderSequenceException: Out of order sequence number for producerId 74: 158 (incoming seq. number), 156 (current end sequence number)

In this bug case, we have 156 → 158 jump, because 2 messages where sent in the batch.

According the kafka java code, it expects only a 1 unit jump so 156 → 157:

https://github.com/apache/kafka/blob/fecb977b257888e2022c1b1e04dd7bf03e18720c/core/src/main/scala/kafka/log/ProducerStateManager.scala#L241-L251

My current understanding/intuition, is that the firstSequence number should always be incremented by 1 and not by the number of messages present in the batch request.

Which roughly means that the updateSequence method of the EOSManager should not take an increment param but always increment by 1 only.
https://github.com/tulios/kafkajs/blob/eae5ea42fc8ac035b0591a705b81829e3fedf8ad/src/producer/eosManager/index.js#L171-L197

I'm still quite noob on this codebase so my reasoning is probably flawed.

@ianwsperber , as you're the main implementer of the eos code, what's your take on this? Thanks a lot.

[Delapouite]

Here's a minimal reproducible case:

const { Kafka, logLevel } = require('kafkajs')

const delay = Number(process.argv[2]) || 100
console.log({ delay })

const kafka = new Kafka({
    clientId: 'test-app',
    brokers: ['localhost:9092'],
})

const createProducer = async () => {
    const producer = kafka.producer({
        allowAutoTopicCreation: false,
        idempotent: true,
        maxInFlightRequests: 1,
    })
    await producer.connect()
    producer.logger().setLogLevel(logLevel.INFO)
    return producer
}

const sendMessages = async (producer) => {
    setInterval(async () => {
        const messageId = Math.random()
        console.log(`sending message ${messageId}`)

        try {
            await producer.send({
                topic: 'test-topic',
                messages: [
                    { value: `Hello KafkaJS user! ${messageId}` },
                ],
            })
            console.log(`confirmed message ${messageId}`)
        } catch (err) {
            console.error(`crash for message ${messageId}`)
            console.error(err)
            process.exit(1)
        }
    }, delay)
}

createProducer()
    .then(sendMessages).catch((err) => {
        console.error(err)
        process.exit(1)
    })

If this code runs with a "large delay" between request (like the default 100ms), everything is fine:

node index.js

But if you run this code with a small delay, the out of OUT_OF_ORDER_SEQUENCE_NUMBER is raised.
Here I voluntarily set the delay to an extreme 1ms on my laptop, but on the real server the value is often higher:

node index.js 1

Here's the log output:

{ delay: 1 }
sending message 0.9237417995495543
sending message 0.32402946325370396
confirmed message 0.9237417995495543
sending message 0.9860404572033681
confirmed message 0.32402946325370396
sending message 0.06510684418463164
confirmed message 0.9860404572033681
sending message 0.599302604904195
confirmed message 0.06510684418463164
sending message 0.5404080025200682
confirmed message 0.599302604904195
sending message 0.22869052896543063
confirmed message 0.5404080025200682
sending message 0.29642055944336443
confirmed message 0.22869052896543063
confirmed message 0.29642055944336443
sending message 0.6128450568370936
{"level":"ERROR","timestamp":"2019-12-15T11:09:15.513Z","logger":"kafkajs","message":"[Connection] Response Produce(key: 0, version: 5)","broker":"localhost:9092","clientId":"test-app","error":"The broker received an out of order sequence number","correlationId":12,"size":58}
{"level":"ERROR","timestamp":"2019-12-15T11:09:15.514Z","logger":"kafkajs","message":"[Producer] The broker received an out of order sequence number","retryCount":0,"retryTime":355}
crash for message 0.6128450568370936
KafkaJSProtocolError: The broker received an out of order sequence number
    at createErrorFromCode (/home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/protocol/error.js:536:10)
    at Object.parse (/home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/protocol/requests/produce/v3/response.js:47:11)
    at Connection.send (/home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/network/connection.js:306:35)
    at processTicksAndRejections (internal/process/task_queues.js:97:5)
    at async Broker.produce (/home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/broker/index.js:252:12)
    at async /home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/producer/sendMessages.js:93:28
    at async Promise.all (index 0)
    at async makeRequests (/home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/producer/sendMessages.js:125:9)
    at async /home/delapouite/code/lab/kafka-crash/node_modules/kafkajs/src/producer/messageProducer.js:61:16
    at async Timeout._onTimeout (/home/delapouite/code/lab/kafka-crash/index.js:28:4) {
  name: 'KafkaJSProtocolError',
  retriable: false,
  helpUrl: undefined,
  type: 'OUT_OF_ORDER_SEQUENCE_NUMBER',
  code: 45
}

error in the kafka logs:

kafka_1      | [2019-12-15 11:09:15,513] ERROR [ReplicaManager broker=1001] Error processing append operation on partition test-topic-0 (kafka.server.ReplicaManager)
kafka_1      | org.apache.kafka.common.errors.OutOfOrderSequenceException: Out of order sequence number for producerId 68 at offset 8176 in partition test-topic-0: 8 (incoming seq. number), 6 (current end sequence number)

As soon as 2 consecutive "confirmed message id" log lines appear, the next send crash.

In the end, I don't think the problem is about "incrementing by one" in updateSequence(), as I said in the first comment of this thread. It's more than when a request has been parked in the RequestQueue, it should not call updateSequence() at all.

[Delapouite]

I've registered to the Slack channel and found some other users who faced the same issue in the past. I'm adding links here so they can be contacted at a later time if progress is made:

by Vykimo https://app.slack.com/client/TF528A1BJ/CF6RFPF6K/thread/CF6RFPF6K-1556120666.002100
by Arpit https://app.slack.com/client/TF528A1BJ/CF6RFPF6K/thread/CF6RFPF6K-1556536043.003600
by Dave Bohl https://app.slack.com/client/TF528A1BJ/CF6RFPF6K/thread/CF6RFPF6K-1569971941.056400

[tulios]

Hi @Delapouite, thanks for the detailed investigation. Most of the team is already on vacation, so bear with me 😄

We implemented transactions right after it was out, so we might have gotten some things wrong. I will try to get the docs around it to confirm some things, can you make the changes you proposed locally and try that? I could also help you land a PR

[iammathew]

Hey @Delapouite @tulios,
facing the same issues on a heavy traffic edge installation for Kafka. Is there a solution for this already in place? Or updates?

[Delapouite]

No solutions found on my side yet. I had to disable idempotency for the moment.

[bpetty-formfast]

No solutions found on my side yet. I had to disable idempotency for the moment.

Thank you, just started using KafkaJS and started seeing this issue. Disabling idempotency got us through. I noticed that the producer is pretty much dead after receiving this error.

[cabelitos]

Yeah. This is also affecting us. Which is kinda bad, since the only solution right know is to disable idempotency. :/.
Tried to understand what's happening, but no progress so far.

[ArnaudovSt]

Any progress on that issue?

[Nevon]

Any progress on that issue?

Not as far as I'm aware, but we are open to contributions if anyone is willing to invest some time into it.

[dearsaturn]

I was able to circumvent this issue with some naive throttling:

import {Producer, Kafka, CompressionTypes, ProducerConfig} from 'kafkajs';
import {IAction, IQueuedRecord} from './types';
import {createActionMessage} from './create_action_message';
import {TopicAdministrator} from './topic_administrator';
import {isKafkaJSProtocolError} from './type_guard';
import Bluebird from 'bluebird';
import pino from 'pino';
import uuid from 'uuid';

export class ThrottledProducer {
    public recordsSent = 0;

    private producer: Producer;
    private topicAdministrator: TopicAdministrator;
    private isConnected: boolean = false;
    private intervalTimeout: NodeJS.Timeout;
    private createdTopics = new Set<string>();
    private recordQueue: IQueuedRecord[] = [];
    private isFlushing = false;
    private logger: pino.Logger;

    constructor(
        protected kafka: Kafka,
        protected producerConfig: Omit<
            ProducerConfig,
            'allowAutoTopicCreation' | 'maxInFlightRequests' | 'idempotent'
        > & {maxOutgoingBatchSize?: number; flushIntervalMs?: number} = {
            maxOutgoingBatchSize: 10000,
            flushIntervalMs: 1000
        },
        topicAdministrator?: TopicAdministrator,
        logger?: pino.Logger
    ) {
        this.topicAdministrator = topicAdministrator || new TopicAdministrator(kafka);
        this.logger = logger
            ? logger.child({class: 'KafkaSagasThrottledProducer'})
            : pino().child({class: 'KafkaSagasThrottledProducer'});
        this.createProducer();
    }

    // tslint:disable-next-line: cyclomatic-complexity
    public putAction = async <Action extends IAction>(action: Action) => {
        if (!this.isConnected) {
            throw new Error('You must .connect before producing actions');
        }

        if (!this.createdTopics.has(action.topic)) {
            this.logger.debug({topic: action.topic}, 'Creating topic');
            await this.topicAdministrator.createTopic(action.topic);
            this.createdTopics.add(action.topic);
        }

        return new Promise<void>((resolve, reject) => {
            this.recordQueue = [
                ...this.recordQueue,
                {
                    resolve,
                    reject,
                    record: {
                        topic: action.topic,
                        messages: [createActionMessage({action})]
                    }
                }
            ];

            return;
        });
    };

    public connect = async () => {
        if (this.isConnected) {
            return;
        }

        const flushIntervalMs = this.producerConfig.flushIntervalMs || 1000;

        this.logger.debug('Connecting producer');
        await this.producer.connect();
        this.logger.debug('Connected producer');

        this.logger.debug({flushIntervalMs}, 'Creating flush interval');
        this.intervalTimeout = setInterval(this.flush, flushIntervalMs);
        this.logger.debug('Created flush interval');

        this.isConnected = true;
    };

    public disconnect = async () => {
        if (!this.isConnected) {
            return;
        }

        this.logger.debug('Disconnecting');
        clearInterval(this.intervalTimeout);

        await this.producer.disconnect();
        this.logger.debug('Disconnected');
        this.isConnected = false;
    };

    private createProducer = () => {
        this.logger.debug('Creating a new producer');
        this.producer = this.kafka.producer({
            maxInFlightRequests: 1,
            idempotent: true,
            ...this.producerConfig
        });
        this.logger.debug('Created a new producer');
    };

    // tslint:disable-next-line: cyclomatic-complexity
    private flush = async (
        retryRecords?: IQueuedRecord[],
        retryCounter = 0,
        retryBatchId?: string
    ) => {
        if (!retryRecords && this.isFlushing) {
            return;
        }

        if (retryCounter) {
            /** Wait for a max of 30 seconds before retrying */
            const retryDelay = Math.min(retryCounter * 1000, 30000);
            this.logger.debug({retryDelay}, 'Waiting before attempting retry');
            await Bluebird.delay(retryDelay);
        }

        /**
         * Ensures that if the interval call ends up being concurrent due latency in sendBatch,
         * unintentinally overlapping cycles are deferred to the next interval.
         */
        this.isFlushing = true;

        const batchSize = this.producerConfig.maxOutgoingBatchSize || 1000;
        const outgoingRecords = retryRecords || this.recordQueue.slice(0, batchSize);
        this.recordQueue = this.recordQueue.slice(batchSize);
        const batchId = retryBatchId || uuid.v4();

        if (!outgoingRecords.length) {
            this.logger.debug({batchId}, 'No records to flush');
            this.isFlushing = false;
            return;
        }

        this.logger.debug(
            {
                remaining: this.recordQueue.length,
                records: outgoingRecords.length,
                batchId
            },
            'Flushing queue'
        );

        try {
            await this.producer.sendBatch({
                topicMessages: outgoingRecords.map(({record}) => record),
                acks: -1,
                compression: CompressionTypes.GZIP
            });

            this.recordsSent += outgoingRecords.length;
            this.logger.debug({batchId}, 'Flushed queue');
            outgoingRecords.map(({resolve}) => resolve());
            this.isFlushing = false;
            return;
        } catch (error) {
            /**
             * If for some reason this producer is no longer recognized by the broker,
             * create a new producer.
             */
            if (isKafkaJSProtocolError(error) && error.type === 'UNKNOWN_PRODUCER_ID') {
                await this.producer.disconnect();
                this.createProducer();
                await this.producer.connect();
                this.logger.debug(
                    {batchId},
                    'Retrying failed flush attempt due to UNKNOWN_PRODUCER_ID'
                );
                await this.flush(outgoingRecords, retryCounter + 1, batchId);
                return;
            }

            outgoingRecords.map(({reject}) => reject(error));
            this.isFlushing = false;
            return;
        }
    };
}

[jducoeur]

Also hitting this issue. (Using Kafka Streams, when trying to pre-seed a KTable.) I'm going to try the throttle approach to get past this immediate hurdle, but this is a serious problem -- if idempotent mode falls over under load, that calls its usefulness into question...

[matthewadams]

👀 seeing with latest kafkajs

[Va1]

experiencing the same issue in idempotent mode