Function getProducer

packages/queue/src/kafka.ts:107–143

()

let producerConnectPromise: Promise<Producer> | null = null;

const getProducer = async (): Promise<Producer> => {
  if (producer) {
    return producer;
  }
  if (!producerConnectPromise) {
    const client = getKafka();
    const p = client.producer({
      idempotent: true,
      // 1 (not 5) to avoid in-flight reordering after a transient broker hiccup:
      // with idempotency on and low retries, reordered batches trip
      // OUT_OF_ORDER_SEQUENCE_NUMBER and stick the producer per-partition.
      maxInFlightRequests: 1,
      allowAutoTopicCreation: true,
      retry: {
        retries: KAFKA_PRODUCER_RETRIES,
        initialRetryTime: KAFKA_PRODUCER_INITIAL_RETRY_MS,
        maxRetryTime: KAFKA_PRODUCER_MAX_RETRY_MS,
        factor: 2,
      },
    });
    producerConnectPromise = p
      .connect()
      .then(() => {
        producer = p;
        kafkaLogger.info(
          { brokers: KAFKA_BROKERS, topic: KAFKA_EVENTS_TOPIC },
          'kafka producer connected'
        );
        return p;
      })
      .catch((err) => {
        producerConnectPromise = null;
        throw err;
      });
  }
  return producerConnectPromise;
};

// Kafka error codes that mean the producer's PID/sequence state is
// permanently out of sync with the broker for some partition — only a