feat(microservices): Removes unused exception and updates tests

Author: mkaufmaner

packages/microservices/client/client-kafka.ts

@@ -8,7 +8,6 @@ import {
 } from '../constants';
 import { KafkaResponseDeserializer } from '../deserializers/kafka-response.deserializer';
 import { KafkaHeaders } from '../enums';
-import { InvalidKafkaClientTopicPartitionException } from '../errors/invalid-kafka-client-topic-partition.exception';
 import { InvalidKafkaClientTopicException } from '../errors/invalid-kafka-client-topic.exception';
 import {
   BrokersFunction,

packages/microservices/errors/invalid-kafka-client-topic-partition.exception.ts

@@ -18,7 +18,6 @@ export class KafkaReplyPartitionAssigner {
   constructor(
     private readonly clientKafka: ClientKafka,
     private readonly config: {
-      groupId: string;
       cluster: Cluster;
     },
   ) {

packages/microservices/helpers/kafka-reply-partition-assigner.ts

@@ -3,7 +3,6 @@ import * as sinon from 'sinon';
 import { ClientKafka } from '../../client/client-kafka';
 import { NO_MESSAGE_HANDLER } from '../../constants';
 import { KafkaHeaders } from '../../enums';
-import { InvalidKafkaClientTopicPartitionException } from '../../errors/invalid-kafka-client-topic-partition.exception';
 import { InvalidKafkaClientTopicException } from '../../errors/invalid-kafka-client-topic.exception';
 import {
   ConsumerGroupJoinEvent,
@@ -269,6 +268,7 @@ describe('ClientKafka', () => {
 
       expect(createClientStub.calledOnce).to.be.true;
       expect(producerStub.calledOnce).to.be.true;
+
       expect(consumerStub.calledOnce).to.be.true;
 
       expect(on.calledOnce).to.be.true;
@@ -314,13 +314,19 @@ describe('ClientKafka', () => {
           memberId: 'member-1',
           memberAssignment: {
             'topic-a': [0, 1, 2],
+            'topic-b': [3, 4, 5],
           },
         },
       };
 
       client['setConsumerAssignments'](consumerAssignments);
+
       expect(client['consumerAssignments']).to.deep.eq(
-        consumerAssignments.payload.memberAssignment,
+        // consumerAssignments.payload.memberAssignment,
+        {
+          'topic-a': 0,
+          'topic-b': 3,
+        },
       );
     });
   });
@@ -493,30 +499,40 @@ describe('ClientKafka', () => {
     });
   });
 
+  describe('getConsumerAssignments', () => {
+    it('should get consumer assignments', () => {
+      client['consumerAssignments'] = {
+        [replyTopic]: 0,
+      };
+
+      const result = client.getConsumerAssignments();
+
+      expect(result).to.deep.eq(client['consumerAssignments']);
+    });
+  });
+
   describe('getReplyTopicPartition', () => {
     it('should get reply partition', () => {
       client['consumerAssignments'] = {
-        [replyTopic]: [0],
+        [replyTopic]: 0,
       };
 
       const result = client['getReplyTopicPartition'](replyTopic);
 
       expect(result).to.eq('0');
     });
 
-    it('should throw error when the topic is being consumed but is not assigned partitions', () => {
-      client['consumerAssignments'] = {
-        [replyTopic]: [],
-      };
+    it('should throw error when the topic is not being consumed', () => {
+      client['consumerAssignments'] = {};
 
       expect(() => client['getReplyTopicPartition'](replyTopic)).to.throw(
-        InvalidKafkaClientTopicPartitionException,
+        InvalidKafkaClientTopicException,
       );
     });
 
-    it('should throw error when the topic is not being consumer', () => {
+    it('should throw error when the topic is not being consumed', () => {
       client['consumerAssignments'] = {
-        [topic]: [],
+        [topic]: undefined,
       };
 
       expect(() => client['getReplyTopicPartition'](replyTopic)).to.throw(
@@ -568,7 +584,7 @@ describe('ClientKafka', () => {
 
       // set
       client['consumerAssignments'] = {
-        [replyTopic]: [parseFloat(replyPartition)],
+        [replyTopic]: parseFloat(replyPartition),
       };
     });
 

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

@@ -1,14 +1,16 @@
 import { expect } from 'chai';
 import * as Kafka from 'kafkajs';
-import { KafkaRoundRobinPartitionAssigner } from '../../helpers/kafka-round-robin-partition-assigner';
+import { KafkaReplyPartitionAssigner } from '../../helpers/kafka-reply-partition-assigner';
+import { ClientKafka } from '../../client/client-kafka';
 
-describe('kafka round robin by time', () => {
-  let cluster, topics, metadata, assigner;
+describe('kafka reply partition assigner', () => {
+  let cluster, topics, metadata, assigner, client;
 
   beforeEach(() => {
     metadata = {};
     cluster = { findTopicPartitionMetadata: topic => metadata[topic] };
-    assigner = new KafkaRoundRobinPartitionAssigner({ cluster });
+    client = new ClientKafka({});
+    assigner = new KafkaReplyPartitionAssigner(client, { cluster });
     topics = ['topic-A', 'topic-B'];
   });