Function produceIncomingEvent

packages/queue/src/kafka.ts:186–212

(
  payload: EventsQueuePayloadIncomingEvent['payload'],
  partitionKey: string
)

};

export const produceIncomingEvent = async (
  payload: EventsQueuePayloadIncomingEvent['payload'],
  partitionKey: string
): Promise<void> => {
  const p = await getProducer();
  try {
    await p.send({
      topic: KAFKA_EVENTS_TOPIC,
      timeout: KAFKA_REQUEST_TIMEOUT_MS,
      messages: [
        {
          key: Buffer.from(partitionKey),
          value: Buffer.from(JSON.stringify(payload)),
        },
      ],
    });
  } catch (err) {
    if (isFatalProducerError(err)) {
      kafkaLogger.warn(
        { err },
        'kafka producer in fatal state; resetting for next call'
      );
      resetProducer(p);
    }
    throw err;
  }
};

const consumers = new Set<Consumer>();