Merge pull request #16326 from suuuuuuminnnnnn/test/kafka-run-options-forwarding

test(microservices): verify kafka run options forwarding

Author: kamilmysliwiec

packages/microservices/interfaces/microservice-configuration.interface.ts

@@ -339,6 +339,11 @@ export interface KafkaOptions {
     postfixId?: string;
     client?: KafkaConfig;
     consumer?: ConsumerConfig;
+    /**
+     * Options passed to KafkaJS consumer.run().
+     * Note: `partitionsConsumedConcurrently` (KafkaJS parameter) controls
+     * concurrent processing at the partition level (not topic level).
+     */
     run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
     subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
     producer?: ProducerConfig;

packages/microservices/test/server/server-kafka.spec.ts

@@ -256,6 +256,39 @@ describe('ServerKafka', () => {
       expect(run.called).to.be.true;
       expect(connect.called).to.be.true;
     });
+    it('should pass run options with partitionsConsumedConcurrently to consumer.run()', async () => {
+      untypedServer.logger = new NoopLogger();
+      untypedServer.options.run = {
+        partitionsConsumedConcurrently: 5,
+      };
+      await server.listen(callback);
+      await server.bindEvents(untypedServer.consumer);
+
+      expect(run.called).to.be.true;
+      expect(run.firstCall.args[0]).to.include({
+        partitionsConsumedConcurrently: 5,
+      });
+      expect(run.firstCall.args[0]).to.have.property('eachMessage');
+    });
+    it('should pass multiple run options to consumer.run()', async () => {
+      untypedServer.logger = new NoopLogger();
+      untypedServer.options.run = {
+        partitionsConsumedConcurrently: 3,
+        autoCommit: false,
+        autoCommitInterval: 5000,
+      };
+      await server.listen(callback);
+      await server.bindEvents(untypedServer.consumer);
+
+      expect(run.called).to.be.true;
+      const callArg = run.firstCall.args[0];
+      expect(callArg).to.include({
+        partitionsConsumedConcurrently: 3,
+        autoCommit: false,
+        autoCommitInterval: 5000,
+      });
+      expect(callArg).to.have.property('eachMessage');
+    });
   });
 
   describe('getMessageHandler', () => {