TimeoutNegativeWarning emitted when running on Node 24.0.1

[wilsonjackson]

Describe the bug
When running on Node 24.0.1, a TimeoutNegativeWarning is emitted when the client makes a connection. The timeout value appears to be a negative timestamp, presumably resulting from this subtraction.

To Reproduce

With a trivial app:

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  brokers: ['localhost:8092']
});

const producer = kafka.producer();
producer.connect();

Run with warnings enabled:

KAFKAJS_NO_PARTITIONER_WARNING=1 node --trace-warnings index.js

Creating a producer, consumer, or admin client all cause the same emitted warning.

Expected behavior
No warnings or errors emitted.

Observed behavior

(node:25173) TimeoutNegativeWarning: -1747359280440 is a negative number.
Timeout duration was set to 1.
    at new Timeout (node:internal/timers:195:17)
    at setTimeout (node:timers:120:19)
    at RequestQueue.scheduleCheckPendingRequests (/tmp/kafkajs-warning/node_modules/kafkajs/src/network/requestQueue/index.js:317:37)
    at RequestQueue.checkPendingRequests (/tmp/kafkajs-warning/node_modules/kafkajs/src/network/requestQueue/index.js:297:10)
    at RequestQueue.fulfillRequest (/tmp/kafkajs-warning/node_modules/kafkajs/src/network/requestQueue/index.js:206:10)
    at Connection.processData (/tmp/kafkajs-warning/node_modules/kafkajs/src/network/connection.js:529:25)
    at Socket.onData (/tmp/kafkajs-warning/node_modules/kafkajs/src/network/connection.js:184:14)
    at Socket.emit (node:events:507:28)
    at addChunk (node:internal/streams/readable:559:12)
    at readableAddChunkPushByteMode (node:internal/streams/readable:510:3)

Environment:

• OS: MacOS 13.7.2
• KafkaJS version: 2.2.4
• Kafka version: 3.7.2
• NodeJS version: 24.0.1

[eduinlight]

Same for me

[yogesh-aggarwal]

Same for me

Code for kafka:

Output:

[velgardey]

any fixes ?

[Vinayak-Soni]

I used copilot to fix this issue and below code works for me.

const { EventEmitter } = require('events')
const SocketRequest = require('./socketRequest')
const events = require('../instrumentationEvents')
const { KafkaJSInvariantViolation } = require('../../errors')

const PRIVATE = {
  EMIT_QUEUE_SIZE_EVENT: Symbol('private:RequestQueue:emitQueueSizeEvent'),
  EMIT_REQUEST_QUEUE_EMPTY: Symbol('private:RequestQueue:emitQueueEmpty'),
}

const REQUEST_QUEUE_EMPTY = 'requestQueueEmpty'
const CHECK_PENDING_REQUESTS_INTERVAL = 10

module.exports = class RequestQueue extends EventEmitter {
  /**
   * @param {Object} options
   * @param {number} options.maxInFlightRequests
   * @param {number} options.requestTimeout
   * @param {boolean} options.enforceRequestTimeout
   * @param {string} options.clientId
   * @param {string} options.broker
   * @param {import("../../../types").Logger} options.logger
   * @param {import("../../instrumentation/emitter")} [options.instrumentationEmitter=null]
   * @param {() => boolean} [options.isConnected]
   */
  constructor({
    instrumentationEmitter = null,
    maxInFlightRequests,
    requestTimeout,
    enforceRequestTimeout,
    clientId,
    broker,
    logger,
    isConnected = () => true,
  }) {
    super()
    this.instrumentationEmitter = instrumentationEmitter
    this.maxInFlightRequests = maxInFlightRequests
    this.requestTimeout = requestTimeout
    this.enforceRequestTimeout = enforceRequestTimeout
    this.clientId = clientId
    this.broker = broker
    this.logger = logger
    this.isConnected = isConnected

    this.inflight = new Map()
    this.pending = []

    /**
     * Until when this request queue is throttled and shouldn't send requests
     *
     * The value represents the timestamp of the end of the throttling in ms-since-epoch. If the value
     * is smaller than the current timestamp no throttling is active.
     *
     * @type {number}
     */
    this.throttledUntil = -1

    /**
     * Timeout id if we have scheduled a check for pending requests due to client-side throttling
     *
     * @type {null|NodeJS.Timeout}
     */
    this.throttleCheckTimeoutId = null

    this[PRIVATE.EMIT_REQUEST_QUEUE_EMPTY] = () => {
      if (this.pending.length === 0 && this.inflight.size === 0) {
        this.emit(REQUEST_QUEUE_EMPTY)
      }
    }

    this[PRIVATE.EMIT_QUEUE_SIZE_EVENT] = () => {
      instrumentationEmitter &&
        instrumentationEmitter.emit(events.NETWORK_REQUEST_QUEUE_SIZE, {
          broker: this.broker,
          clientId: this.clientId,
          queueSize: this.pending.length,
        })

      this[PRIVATE.EMIT_REQUEST_QUEUE_EMPTY]()
    }
  }

  /**
   * @public
   */
  scheduleRequestTimeoutCheck() {
    if (this.enforceRequestTimeout) {
      this.destroy()

      this.requestTimeoutIntervalId = setInterval(() => {
        this.inflight.forEach(request => {
          if (Date.now() - request.sentAt > request.requestTimeout) {
            request.timeoutRequest()
          }
        })

        if (!this.isConnected()) {
          this.destroy()
        }
      }, Math.min(this.requestTimeout, 100))
    }
  }

  maybeThrottle(clientSideThrottleTime) {
    // Validate incoming throttle value to avoid corrupt/invalid numbers
    if (clientSideThrottleTime == null) return

    if (typeof clientSideThrottleTime !== 'number' || !Number.isFinite(clientSideThrottleTime)) {
      this.logger.warn('Invalid clientSideThrottleTime received', {
        clientSideThrottleTime,
        broker: this.broker,
        clientId: this.clientId,
      })
      return
    }

    // Ignore non-positive values
    if (clientSideThrottleTime <= 0) return

    // Clamp excessively large throttle values (defensive)
    const MAX_REASONABLE_THROTTLE = 24 * 60 * 60 * 1000 // 24 hours
    let safeThrottle = clientSideThrottleTime
    if (clientSideThrottleTime > MAX_REASONABLE_THROTTLE) {
      this.logger.warn('Clamping overly large clientSideThrottleTime', {
        clientSideThrottleTime,
        clampedTo: MAX_REASONABLE_THROTTLE,
        broker: this.broker,
        clientId: this.clientId,
      })
      safeThrottle = MAX_REASONABLE_THROTTLE
    }

    this.logger.debug(`Client side throttling in effect for ${safeThrottle}ms`)
    const minimumThrottledUntil = Date.now() + safeThrottle
    this.throttledUntil = Math.max(minimumThrottledUntil, this.throttledUntil)

  }

  createSocketRequest(pushedRequest) {
    const { correlationId } = pushedRequest.entry
    const defaultRequestTimeout = this.requestTimeout
    const customRequestTimeout = pushedRequest.requestTimeout

    // Some protocol requests have custom request timeouts (e.g JoinGroup, Fetch, etc). The custom
    // timeouts are influenced by user configurations, which can be lower than the default requestTimeout
    const requestTimeout = Math.max(defaultRequestTimeout, customRequestTimeout || 0)

    const socketRequest = new SocketRequest({
      entry: pushedRequest.entry,
      expectResponse: pushedRequest.expectResponse,
      broker: this.broker,
      clientId: this.clientId,
      instrumentationEmitter: this.instrumentationEmitter,
      requestTimeout,
      send: () => {
        if (this.inflight.has(correlationId)) {
          throw new KafkaJSInvariantViolation('Correlation id already exists')
        }
        this.inflight.set(correlationId, socketRequest)
        pushedRequest.sendRequest()
      },
      timeout: () => {
        this.inflight.delete(correlationId)
        this.checkPendingRequests()
        // Try to emit REQUEST_QUEUE_EMPTY. Otherwise, waitForPendingRequests may stuck forever
        this[PRIVATE.EMIT_REQUEST_QUEUE_EMPTY]()
      },
    })

    return socketRequest
  }

  /**
   * @typedef {Object} PushedRequest
   * @property {import("./socketRequest").RequestEntry} entry
   * @property {boolean} expectResponse
   * @property {Function} sendRequest
   * @property {number} [requestTimeout]
   *
   * @public
   * @param {PushedRequest} pushedRequest
   */
  push(pushedRequest) {
    const { correlationId } = pushedRequest.entry
    const socketRequest = this.createSocketRequest(pushedRequest)

    if (this.canSendSocketRequestImmediately()) {
      this.sendSocketRequest(socketRequest)
      return
    }

    this.pending.push(socketRequest)
    this.scheduleCheckPendingRequests()

    this.logger.debug('Request enqueued', {
      clientId: this.clientId,
      broker: this.broker,
      correlationId,
    })

    this[PRIVATE.EMIT_QUEUE_SIZE_EVENT]()
  }

  /**
   * @param {SocketRequest} socketRequest
   */
  sendSocketRequest(socketRequest) {
    socketRequest.send()

    if (!socketRequest.expectResponse) {
      this.logger.debug('Request does not expect a response, resolving immediately', {
        clientId: this.clientId,
        broker: this.broker,
        correlationId: socketRequest.correlationId,
      })

      this.inflight.delete(socketRequest.correlationId)
      socketRequest.completed({ size: 0, payload: null })
    }
  }

  /**
   * @public
   * @param {object} response
   * @param {number} response.correlationId
   * @param {Buffer} response.payload
   * @param {number} response.size
   */
  fulfillRequest({ correlationId, payload, size }) {
    const socketRequest = this.inflight.get(correlationId)
    this.inflight.delete(correlationId)
    this.checkPendingRequests()

    if (socketRequest) {
      socketRequest.completed({ size, payload })
    } else {
      this.logger.warn('Response without match', {
        clientId: this.clientId,
        broker: this.broker,
        correlationId,
      })
    }

    this[PRIVATE.EMIT_REQUEST_QUEUE_EMPTY]()
  }

  /**
   * @public
   * @param {Error} error
   */
  rejectAll(error) {
    const requests = [...this.inflight.values(), ...this.pending]

    for (const socketRequest of requests) {
      socketRequest.rejected(error)
      this.inflight.delete(socketRequest.correlationId)
    }

    this.pending = []
    this.inflight.clear()
    this[PRIVATE.EMIT_QUEUE_SIZE_EVENT]()
  }

  /**
   * @public
   */
  waitForPendingRequests() {
    return new Promise(resolve => {
      if (this.pending.length === 0 && this.inflight.size === 0) {
        return resolve()
      }

      this.logger.debug('Waiting for pending requests', {
        clientId: this.clientId,
        broker: this.broker,
        currentInflightRequests: this.inflight.size,
        currentPendingQueueSize: this.pending.length,
      })

      this.once(REQUEST_QUEUE_EMPTY, () => resolve())
    })
  }

  /**
   * @public
   */
  destroy() {
    clearInterval(this.requestTimeoutIntervalId)
    clearTimeout(this.throttleCheckTimeoutId)
    this.throttleCheckTimeoutId = null
  }

  canSendSocketRequestImmediately() {
    const shouldEnqueue =
      (this.maxInFlightRequests != null && this.inflight.size >= this.maxInFlightRequests) ||
      this.throttledUntil > Date.now()

    return !shouldEnqueue
  }

  /**
   * Check and process pending requests either now or in the future
   *
   * This function will send out as many pending requests as possible taking throttling and
   * in-flight limits into account.
   */
  checkPendingRequests() {
    while (this.pending.length > 0 && this.canSendSocketRequestImmediately()) {
      const pendingRequest = this.pending.shift() // first in first out
      this.sendSocketRequest(pendingRequest)

      this.logger.debug('Consumed pending request', {
        clientId: this.clientId,
        broker: this.broker,
        correlationId: pendingRequest.correlationId,
        pendingDuration: pendingRequest.pendingDuration,
        currentPendingQueueSize: this.pending.length,
      })

      this[PRIVATE.EMIT_QUEUE_SIZE_EVENT]()
    }

    this.scheduleCheckPendingRequests()
  }

  /**
   * Ensure that pending requests will be checked in the future
   *
   * If there is a client-side throttling in place this will ensure that we will check
   * the pending request queue eventually.
   */
  scheduleCheckPendingRequests() {
    // If we're throttled: Schedule checkPendingRequests when the throttle
    // should be resolved. If there is already something scheduled we assume that that
    // will be fine, and potentially fix up a new timeout if needed at that time.
    // Note that if we're merely "overloaded" by having too many inflight requests
    // we will anyways check the queue when one of them gets fulfilled.
    let scheduleAt = this.throttledUntil - Date.now()
    // If scheduleAt is not a finite number, log and use the interval
    if (!Number.isFinite(scheduleAt)) {
      this.logger.warn('Invalid scheduleAt computed for throttle; using fallback interval', {
        scheduleAt,
        throttledUntil: this.throttledUntil,
        broker: this.broker,
        clientId: this.clientId,
      })
      scheduleAt = CHECK_PENDING_REQUESTS_INTERVAL
    }

    if (!this.throttleCheckTimeoutId) {
      // If nothing pending and no future throttle, don't schedule
      if (this.pending.length === 0 && scheduleAt <= 0) {
        return
      }

      if (this.pending.length > 0) {
        scheduleAt = scheduleAt > 0 ? scheduleAt : CHECK_PENDING_REQUESTS_INTERVAL
      } else {
        scheduleAt = Math.max(0, scheduleAt)
      }

      this.throttleCheckTimeoutId = setTimeout(() => {
        this.throttleCheckTimeoutId = null
        this.checkPendingRequests()
      }, scheduleAt)
    }
  }
}

i am using bunjs@latest version and v2.4.0 of kafkajs
node -v = v22.16.0

[hp8wvvvgnj6asjm7]

got this message too. will this pull request ever be merged?

[kirit0s]

@Vinayak-Soni maybe next time copilot can write only relevant changes. Something like this:

      if (this.pending.length > 0) {
        scheduleAt = scheduleAt > 0 ? scheduleAt : CHECK_PENDING_REQUESTS_INTERVAL
      } else {
        scheduleAt = Math.max(0, scheduleAt)
      }

Fix in else block.