Refactor to always use a barrier for pending operation (#26)

Readers writer lock with reentrant calls

Refactor to always use a barrier for pending operation to avoid problems or regressions with async pause or resume calls that will be introduced later.
Uses linked lists for the cache.
Removes heap-js code.
Test name improvement to run single parametric tests

Seek signature and add partitionsConsumedConcurrently to
ConsumerRunConfig type definition

Make final max poll interval double of configured value
so even last message processed before cache cleanup can take that time to process.
Fix to restart max poll interval timer on fetch.

Marking batch stale after cache clear was requested and max poll interval is reached before it's cleared.
Add assignmentLost function to the rebalance callback.

Fix to nextN size, version with max.poll.interval.ms applied to each message or batch (only for messages after cache reset)

Performance test, removing outliers

Start performance timer from first message received after resuming

Author: emasab

LICENSE.heap-js

@@ -1,6 +1,6 @@
 The promisified API (lib/kafkajs) is inspired by kafkajs (github.com/tulios/kafkajs).
 The promisified tests (test/promisified) are also adapted from there.
-An async lock implementation and many error types are also adapted from there.
+Many error types are also adapted from there.
 The license notice is reproduced below.
 
 ----

LICENSE.kafkajs

@@ -37,12 +37,12 @@ module.exports = ts.config(
     {
         ...js.configs.recommended,
         files: ["lib/**/*.js", "test/promisified/**/*.js"],
-        ignores: ["lib/kafkajs/_heap.js"]
+        ignores: []
     },
     {
         ...ckjsSpecificSettings,
         files: ["lib/**/*.js", "test/promisified/**/*.js"],
-        ignores: ["lib/kafkajs/_heap.js"]
+        ignores: []
     },
     {
         ...jest.configs['flat/recommended'],

eslint.config.js

@@ -128,6 +128,8 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
             totalMessageSize += message.value.length;
             if (messagesReceived === 1) {
                 consumer.pause([{ topic }]);
+            } else if (messagesReceived === 2) {
+                startTime = hrtime();
             } else if (messagesReceived === totalMessageCnt) {
                 let elapsed = hrtime(startTime);
                 let durationNanos = elapsed[0] * 1e9 + elapsed[1];
@@ -153,7 +155,6 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
     console.log("Starting consumer.")
 
     totalMessageSize = 0;
-    startTime = hrtime();
     consumer.resume([{ topic }]);
     await new Promise((resolve) => {
         let interval = setInterval(() => {
@@ -203,6 +204,8 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, t
             totalMessageSize += message.value.length;
             if (messagesReceived === 1) {
                 consumer.pause([{ topic }]);
+            } else if (messagesReceived === 2) {
+                startTime = hrtime();
             } else if (messagesReceived === totalMessageCnt) {
                 let elapsed = hrtime(startTime);
                 let durationNanos = elapsed[0] * 1e9 + elapsed[1];
@@ -228,7 +231,6 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, t
     console.log("Starting consume-transform-produce.")
 
     totalMessageSize = 0;
-    startTime = hrtime();
     consumer.resume([{ topic: consumeTopic }]);
     await new Promise((resolve) => {
         let interval = setInterval(() => {

examples/performance/performance-primitives-kafkajs.js

@@ -134,6 +134,8 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
             totalMessageSize += message.value.length;
             if (messagesReceived === 1) {
                 consumer.pause([{ topic }]);
+            } else if (messagesReceived === 2) {
+                startTime = hrtime();
             } else if (messagesReceived === totalMessageCnt) {
                 let elapsed = hrtime(startTime);
                 let durationNanos = elapsed[0] * 1e9 + elapsed[1];
@@ -159,7 +161,6 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
     console.log("Starting consumer.")
 
     totalMessageSize = 0;
-    startTime = hrtime();
     consumer.resume([{ topic }]);
     await new Promise((resolve) => {
         let interval = setInterval(() => {
@@ -222,6 +223,8 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, t
             totalMessageSize += message.value.length;
             if (messagesReceived === 1) {
                 consumer.pause([{ topic }]);
+            } else if (messagesReceived === 2) {
+                startTime = hrtime();
             } else if (messagesReceived === totalMessageCnt) {
                 let elapsed = hrtime(startTime);
                 let durationNanos = elapsed[0] * 1e9 + elapsed[1];
@@ -247,7 +250,6 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, t
     console.log("Starting consume-transform-produce.")
 
     totalMessageSize = 0;
-    startTime = hrtime();
     consumer.resume([{ topic: consumeTopic }]);
     await new Promise((resolve) => {
         let interval = setInterval(() => {

examples/performance/performance-primitives.js

@@ -340,6 +340,19 @@ KafkaConsumer.prototype.assignments = function() {
   return this._errorWrap(this._client.assignments(), true);
 };
 
+/**
+ * Is current assignment in rebalance callback lost?
+ *
+ * @note This method should only be called from within the rebalance callback
+ * when partitions are revoked.
+ * 
+ * @return {boolean} true if assignment was lost
+ */
+
+KafkaConsumer.prototype.assignmentLost = function() {
+  return this._client.assignmentLost();
+};
+
 /**
  * Get the type of rebalance protocol used in the consumer group.
  *