Publishing more than actual messages in Google Pub/Sub using Node.js and csv-parse

Question

Using Node.js, Google Pub/Sub, csv-parse.

Use case - I have a large csv file to process and import in my DB. It has few third party APIs which take 1 second to process each row. So process flow is below -

1. User uploads the file
2. node server upload the file in storage and send message to PubSubNo.1
3. Now My listener listens to above pubsub and starts processing these messages , it download the file and start breaking each row and publishes to another PubSub for further processing
4. In the end I parallel process these smaller row messages and achieve faster processing.

Problem - As soon as my listener downloads the file it send x no. of row messages to next PubSubNo2 but when i check its subscription it shows more than x messages. e.g. I upload a 6000 record csv and on subscriber it shows more than 40K-50K messages.

Package.json

"dependencies": {
    "@google-cloud/pubsub": "1.5.0",
    "axios": "^0.19.2",
    "csv-parse": "^4.8.5",
    "dotenv": "^8.2.0",
    "google-gax": "1.14.1",
    "googleapis": "47.0.0",
    "moment": "^2.24.0",
    "path": "^0.12.7",
    "pg": "^7.18.1",
    "winston": "^3.0.0"
  }

Publisher Code

async processFile(filename) {
    let cnt = 0;
    let index = null;
    let rowCounter = 0;
    const handler = (resolve, reject) => {
      const parser = CsvParser({
          delimiter: ',',
        })
        .on('readable', () => {
          let row;
          let hello = 0;
          let busy = false;
          this.meta.totalRows = (parser.info.records - 1);
          while (row = parser.read()) {
            if (cnt++ === 0) {
              index = row;
              continue;
            }
            let messageObject = {
              customFieldsMap: this.customFieldsMap,
              importAttributes: this.jc.attrs,
              importColumnData: row,
              rowCount: cnt,
              importColumnList: index,
              authToken: this.token
            }
            let topicPublishResult = PubSubPublish.publishToTopic(process.env.IMPORT_CSV_ROW_PUBLISHING_TOPIC, messageObject);
            topicPublishResult.then((response) => {
              rowCounter += 1;
              const messageInfo = "Row " + rowCounter + " published" +
                " | MessageId = " + response +
                " | importId = " + this.data.importId +
                " | fileId = " + this.data.fileId +
                " | orgId = " + this.data.orgId;
              console.info(messageInfo);
            })
          }
        })
        .on('end', () => {
          console.log("File consumed!");
          resolve(this.setStatus("queued"))
        })
        .on('error', reject);
      fs.createReadStream(filename).pipe(parser);
    };
    await new Promise(handler);
  }

And Publish module code

const {
  PubSub
} = require('@google-cloud/pubsub');

const pubsub = new PubSub({
  projectId: process.env.PROJECT_ID
});
module.exports = {
  publishToTopic: function(topicName, data) {
    return pubsub.topic(topicName, {
      batching: {
        maxMessages: 500,
        maxMilliseconds: 5000,
      }
    }).publish(Buffer.from(JSON.stringify(data)));
  },
};

This works without any issues for file os 10, 100,200,2000 records but giving trouble with more as in for 6K records. After I publish 6K records there is an error of UnhandledPromiseRejection for all 6K records e.g.

(node:49994) UnhandledPromiseRejectionWarning: Error: Retry total timeout exceeded before any response was received
    at repeat (/Users/tarungupta/office/import-processor/node_modules/google-gax/build/src/normalCalls/retries.js:65:31)
    at Timeout._onTimeout (/Users/tarungupta/office/import-processor/node_modules/google-gax/build/src/normalCalls/retries.js:100:25)
    at listOnTimeout (internal/timers.js:531:17)
    at processTimers (internal/timers.js:475:7)
(node:49994) UnhandledPromiseRejectionWarning: Unhandled promise rejection. This error originated either by throwing inside of an async function without a catch block, or by rejecting a promise which was not handled with .catch(). (rejection id: 6000)

Any help is appreciated!

Answer 1

It's possible that your publisher is getting overwhelmed when you have 6,000 messages to publish. The reason is that you create a new instance of the publisher for each message that you create in your publishToTopic method. Consequently, you are not getting to take advantage of any batching and you are waiting 5 seconds to send every message. That's a lot of overhead for each message. It could mean that callbacks are not getting processed in a timely fashion, resulting in timeouts and attempts to resend. You want to create your pubsub.topic object a single time and then reuse it across publish calls.