usage of `minBytes` results in extreme Consumption bandwidth usage

[tobiasdcl]

Hi there,
I started to migrate a consumer to @platformatic/kafka that was using kafkajs before.
The DX is great! I noticed a minor type issue and submitted a PR for that ref - Thanks for the great work 🚀

However during validation in a pre prod stage I noticed an extreme spike in the Consumption bandwidth of the related kafka Topic. Usually Production and consumption bandwidth are somewhat equal but I observed an ~ 100x increase in the consumption bandwidth compared to the production bandwidth. I was able to identify setting minBytes: 250_000 as a property in the options of the consumer to be the cause of this. Once I removed the minBytes property the bandwidth normalized again (I even made another deployment afterwards with minBytes again to ensure the behaviour is consistent - which it was)

I was not yet able to identify the actual root cause for this but wanted to share the issue with you in case you already have a hunch what might be going on here.

[patrykwegrzyn]

Hi there
Is your setup piping data between topics (like A → B), or just consuming and producing on the same topic (A)?

Bit more detail on your pipeline would help me get a better picture.
Thanks for flagging this.

[tobiasdcl]

Hi there,
thanks for looking into this!

the pipeline is very basic, single Topic with three distinct producers and one consumer - no piping between topics. The Topic is hosted at Confluent Cloud.

here is the boiled down config that I used during testing:

   const consumer = new Consumer({
      clientId: `clientId`,
      groupId: `groupId`,
      bootstrapBrokers: ['broker-name'],
      tlsServerName: true,
      sasl: { 
         mechanism: 'PLAIN',
         username,
         password
      },
      tls: {},
      retries: 5,
      maxWaitTime: 10_000,
      sessionTimeout: 45_000,
      heartbeatInterval: 15_000,
      //minBytes: 250_000,
      },
    })

   const stream = await consumer.consume({
      autocommit: true,
      topics: ['topic-name'],
      mode: MessagesStreamModes.COMMITTED,
      fallbackMode: MessagesStreamFallbackModes.EARLIEST,
    });

    for await (const message of stream) {
      // process message
    }

[baac0]

Hello @tobiasdcl,

could you try whether #97 works for you and improves upon the situation? The change should avoid transferring records batches multiple times.

Cheers, Achim

[tobiasdcl]

Hello @tobiasdcl,

could you try whether #97 works for you and improves upon the situation? The change should avoid transferring records batches multiple times.

Cheers, Achim

I cant say for sure but looking at the PR I strongly believe this is the root cause!

Further evidence fir that is that omiting minBytes was enough to mitigate the issue in a pre-pro stage but not in production. With the difference beeing that prod has a lot more data then pre-prod. As a mitigation we set a really low value for maxBytes - this mitigates the issue since the batch size is significantly smaller resulting in less duplicate data beeing sent.

Unfortunately, I will only be able to verify that the PR fixed the issue once it is merged and a new version is released, sorry about that!

[patrykwegrzyn]

@baac0 i think you are right recordSize match maxBytes but always process single record
maxBytes: 250000
recordSize:250000

 {
  firstOffset: 944n,
  length: 108,
  partitionLeaderEpoch: 0,
  magic: 2,
  crc: 238395320,
  attributes: 0,
  lastOffsetDelta: 0,
  firstTimestamp: 1754050928068n,
  maxTimestamp: 1754050928068n,
  producerId: -1n,
  producerEpoch: 0,
  firstSequence: 0,
  records: [
    {
      length: 58,
      attributes: 0,
      timestampDelta: 0n,
      offsetDelta: 0,
      key: <Buffer 6b 65 79 2d 32 38 33 32>,
      value: <Buffer 76 61 6c 75 65 2d 32 38 33 32>,
      headers: [Array]
    }
  ]
}

this would suggest commit for single offset and a lot of data being re-fetched

Good Catch