Add parametrization to tests for auto commit

Author: milindl

test/promisified/consumer/consumeMessages.spec.js

@@ -13,9 +13,9 @@ const {
     waitForConsumerToJoinGroup,
     sleep,
     generateMessages,
-} = require('../testhelpers')
+} = require('../testhelpers');
 
-describe('Consumer', () => {
+describe.each([[true], [false]])('Consumer', (isAutoCommit) => {
     let topicName, groupId, producer, consumer;
 
     beforeEach(async () => {
@@ -30,6 +30,7 @@ describe('Consumer', () => {
             groupId,
             maxWaitTimeInMs: 100,
             fromBeginning: true,
+            autoCommit: isAutoCommit,
         });
     });
 
@@ -307,6 +308,7 @@ describe('Consumer', () => {
                 minBytes: 1024,
                 maxWaitTimeInMs: 500,
                 fromBeginning: true,
+                autoCommit: isAutoCommit,
             })
 
             const messages = Array(10)
@@ -492,6 +494,7 @@ describe('Consumer', () => {
                 groupId,
                 maxWaitTimeInMs: 100,
                 fromBeginning: true,
+                autoCommit: isAutoCommit
             });
 
             await consumer.connect();
@@ -528,7 +531,8 @@ describe('Consumer', () => {
             consumer = createConsumer({
                 groupId,
                 maxWaitTimeInMs: 100,
-                fromBeginning: true
+                fromBeginning: true,
+                autoCommit: isAutoCommit,
             });
 
             await consumer.connect();
@@ -585,6 +589,7 @@ describe('Consumer', () => {
                 groupId,
                 maxWaitTimeInMs: 100,
                 fromBeginning: true,
+                autoCommit: isAutoCommit,
             });
 
             await consumer.connect();
@@ -642,6 +647,7 @@ describe('Consumer', () => {
                     maxWaitTimeInMs: 100,
                     readUncommitted: true,
                     fromBeginning: true,
+                    autoCommit: isAutoCommit,
                 })
 
                 await consumer.connect();
@@ -678,6 +684,9 @@ describe('Consumer', () => {
         it(
             'respects offsets sent by a committed transaction ("consume-transform-produce" flow)',
             async () => {
+                if (isAutoCommit) { /* only autoCommit: false makes sense for this test. */
+                    return;
+                }
                 // Seed the topic with some messages. We don't need a tx producer for this.
                 await producer.connect();
                 const partition = 0;
@@ -785,6 +794,10 @@ describe('Consumer', () => {
         it(
             'does not respect offsets sent by an aborted transaction ("consume-transform-produce" flow)',
             async () => {
+                if (isAutoCommit) { /* only autoCommit: false makes sense for this test. */
+                    return;
+                }
+
                 // Seed the topic with some messages. We don't need a tx producer for this.
                 await producer.connect();
 

test/promisified/consumer/consumerCacheTests.spec.js

@@ -1,5 +1,6 @@
 jest.setTimeout(30000)
 
+const { is } = require('bluebird');
 const {
     secureRandom,
     createTopic,
@@ -10,7 +11,7 @@ const {
     sleep,
 } = require('../testhelpers');
 
-describe('Consumer message cache', () => {
+describe.each([[false], [true]])('Consumer message cache', (isAutoCommit) => {
     let topicName, groupId, producer, consumer;
 
     beforeEach(async () => {
@@ -21,10 +22,12 @@ describe('Consumer message cache', () => {
 
         producer = createProducer({});
 
+        const common = {};
         consumer = createConsumer({
             groupId,
             maxWaitTimeInMs: 100,
             fromBeginning: true,
+            autoCommit: isAutoCommit,
         });
     });
 
@@ -127,6 +130,7 @@ describe('Consumer message cache', () => {
             groupId,
             maxWaitTimeInMs: 100,
             fromBeginning: true,
+            autoCommit: isAutoCommit,
         });
 
         await consumer.connect();
@@ -142,6 +146,10 @@ describe('Consumer message cache', () => {
             eachMessage: async event => {
                 messagesConsumed.push(event);
                 messagesConsumedConsumer1.push(event);
+                if (!isAutoCommit)
+                    await consumer.commitOffsets([
+                        { topic: event.topic, partition: event.partition, offset: Number(event.message.offset) + 1 },
+                    ]);
 
                 /* Until the second consumer joins, consume messages slowly so as to not consume them all
                  * before the rebalance triggers. */
@@ -204,6 +212,7 @@ describe('Consumer message cache', () => {
             rebalanceTimeout: 10000,
             sessionTimeout: 10000,
             clientId: "impatientConsumer",
+            autoCommit: isAutoCommit,
         });
 
         await producer.connect();
@@ -219,6 +228,11 @@ describe('Consumer message cache', () => {
             eachMessage: async event => {
                 messagesConsumed.push(event);
                 impatientConsumerMessages.push(event);
+                if (!isAutoCommit)
+                    await impatientConsumer.commitOffsets([
+                        { topic: event.topic, partition: event.partition, offset: Number(event.message.offset) + 1 },
+                    ]);
+
                 /* When the second consumer is joining, deliberately slow down message consumption.
                  * We should still have a rebalance very soon, since we must expire the cache and
                  * trigger a rebalance before max.poll.interval.ms.
@@ -268,5 +282,5 @@ describe('Consumer message cache', () => {
         expect(impatientConsumerMessages.length).toBeGreaterThan(0);
 
         await impatientConsumer.disconnect();
-    });
+    }, 60000);
 });

test/promisified/consumer/subscribe.spec.js

@@ -3,6 +3,7 @@ jest.setTimeout(30000);
 const { Kafka, ErrorCodes } = require('../../../lib').KafkaJS;
 const { secureRandom,
     createTopic,
+    waitFor,
     waitForMessages,
     waitForConsumerToJoinGroup,
     createProducer,
@@ -13,7 +14,6 @@ describe('Consumer', () => {
 
     beforeEach(async () => {
         groupId = `consumer-group-id-${secureRandom()}`;
-
         consumer = createConsumer({
             groupId,
             maxWaitTimeInMs: 1,
@@ -52,7 +52,7 @@ describe('Consumer', () => {
             });
 
             consumer.run({ eachMessage: async event => messagesConsumed.push(event) });
-            await waitForConsumerToJoinGroup(consumer);
+            await waitFor(() => consumer.assignment().length > 0, () => null);
 
             await producer.connect();
             await producer.sendBatch({

test/promisified/testhelpers.js

@@ -14,28 +14,27 @@ const clusterInformation = {
 
 const debug = process.env.TEST_DEBUG;
 
-function makeConfig(config) {
+function makeConfig(config, common) {
     const kafkaJS =  Object.assign(config, clusterInformation.kafkaJS);
-    const common = {};
     if (debug) {
         common['debug'] = debug;
     }
 
     return Object.assign(common, { kafkaJS });
 }
 
-function createConsumer(config) {
-    const kafka = new Kafka(makeConfig(config));
+function createConsumer(config, common = {}) {
+    const kafka = new Kafka(makeConfig(config, common));
     return kafka.consumer();
 }
 
-function createProducer(config) {
-    const kafka = new Kafka(makeConfig(config));
+function createProducer(config, common = {}) {
+    const kafka = new Kafka(makeConfig(config, common));
     return kafka.producer();
 }
 
-function createAdmin(config) {
-    const kafka = new Kafka(makeConfig(config));
+function createAdmin(config, common = {}) {
+    const kafka = new Kafka(makeConfig(config, common));
     return kafka.admin();
 }