message: Improve rdkafka producing in error scenarios
diff: @@ -219,18 +219,17 @@ export class RdkafkaProducer extends RdkafkaBase implements IProducer {
  * Sends all pending messages
  */
  private sendPendingMessages() {
  - for (const [, value] of this.messages) {
  - this.sendBoxcars(value);
  - }
  + const messages = Array.from(this.messages.values());
  + // clear messages now because sendBoxcars may insert some
  this.messages.clear();
  - }
  - private sendBoxcars(boxcars: IPendingBoxcar[]) {
  - if (!this.producer) {
  - throw new Error("Invalid producer");
  + for (const message of messages) {
  + this.sendBoxcars(message);
  + }
  }
  + private sendBoxcars(boxcars: IPendingBoxcar[]) {
  for (const boxcar of boxcars) {
  const boxcarMessage: IBoxcarMessage = {
  contents: boxcar.messages,
  @@ -242,6 +241,7 @@ export class RdkafkaProducer extends RdkafkaBase implements IProducer {
  const message = Buffer.from(JSON.stringify(boxcarMessage));
  try {
  + if (this.producer && this.connected) {
  this.producer.produce(
  this.topic, // topic
  boxcar.partitionId ?? null, // partition id or null for consistent random for keyed messages
  @@ -260,6 +260,13 @@ export class RdkafkaProducer extends RdkafkaBase implements IProducer {
  }
  },
  );
  + } else {
  + // we don't have a producer or we are not connected
  + // normally sendBoxcars would not be called in this scenario, but it could happen if
  + // the above this.producer.produce call errors out and calls this.handleError within this for loop
  + // when this happens, let's requeue the messages for later
  + void this.send(boxcar.messages, boxcar.tenantId, boxcar.documentId, boxcar.partitionId);
  + }
  } catch (ex) {
  // produce can throw if the outgoing message queue is full
  boxcar.deferred.reject(ex);
  