Test conflicts of messageRoutingMode and messageRouter

Author: BewareMyPower

src/ProducerConfig.cc

@@ -157,8 +157,10 @@ ProducerConfig::ProducerConfig(const Napi::Object& producerConfig) : topic("") {
     pulsar_producer_configuration_set_block_if_queue_full(this->cProducerConfig.get(), blockIfQueueFull);
   }
 
+  bool useCustomPartition = false;
   if (producerConfig.Has(CFG_ROUTING_MODE) && producerConfig.Get(CFG_ROUTING_MODE).IsString()) {
     std::string messageRoutingMode = producerConfig.Get(CFG_ROUTING_MODE).ToString().Utf8Value();
+    useCustomPartition = (messageRoutingMode == "CustomPartition");
     if (MESSAGE_ROUTING_MODE.count(messageRoutingMode))
       pulsar_producer_configuration_set_partitions_routing_mode(this->cProducerConfig.get(),
                                                                 MESSAGE_ROUTING_MODE.at(messageRoutingMode));
@@ -251,15 +253,12 @@ ProducerConfig::ProducerConfig(const Napi::Object& producerConfig) : topic("") {
     this->cProducerConfig.get()->conf.setBatchingType(PRODUCER_BATCHING_TYPE.at(batchingType));
   }
 
-  if (producerConfig.Has(CFG_MESSAGE_ROUTER)) {
+  if (useCustomPartition && producerConfig.Has(CFG_MESSAGE_ROUTER)) {
     auto value = producerConfig.Get(CFG_MESSAGE_ROUTER);
     if (value.IsFunction()) {
       messageRouter = Napi::Persistent(value.As<Napi::Function>());
       pulsar_producer_configuration_set_message_router(this->cProducerConfig.get(), choosePartition,
                                                        &messageRouter);
-    } else {
-      Napi::TypeError::New(producerConfig.Env(), "messageRouter should be a function")
-          .ThrowAsJavaScriptException();
     }
   }
 }

tests/producer.test.js

@@ -20,6 +20,11 @@
 const Pulsar = require('../index');
 const httpUtils = require('./http_utils');
 
+function getPartition(msgId) {
+  // The message id string is in the format of "entryId,ledgerId,partition,batchIndex"
+  return Number(msgId.toString().split(',')[2]);
+}
+
 (() => {
   describe('Producer', () => {
     let client;
@@ -187,7 +192,7 @@ const httpUtils = require('./http_utils');
           console.log(`All messages have been sent. IDs: ${allMsgIds.join(', ')}`);
           for (let i = 0; i < allMsgIds.length; i += 1) {
             // The message id string is in the format of "entryId,ledgerId,partition,batchIndex"
-            const partition = Number(allMsgIds[i].toString().split(',')[2]);
+            const partition = getPartition(allMsgIds[i]);
             expect(i % numPartitions).toBe(partition);
           }
         } catch (error) {
@@ -210,6 +215,29 @@ const httpUtils = require('./http_utils');
           producer.send({ data: Buffer.from('test') }),
         ).rejects.toThrow('Failed to send message: UnknownError');
       }, 30000);
+      test('Not CustomPartition', async () => {
+        const topic = `test-not-custom-part-${Date.now()}`;
+        const numPartitions = 2;
+        const response = await httpUtils.createPartitionedTopic(topic, numPartitions);
+        expect(response.statusCode).toBe(204);
+
+        let index = 0;
+        const producer = await client.createProducer({
+          topic,
+          messageRouter: (_, topicMetadata) => {
+            const result = index % topicMetadata.numPartitions;
+            index += 1;
+            return result;
+          },
+          messageRoutingMode: 'UseSinglePartition',
+        });
+        const partitions = new Set();
+        for (let i = 0; i < 10; i += 1) {
+          const msgId = await producer.send({ data: Buffer.from('msg') });
+          partitions.add(getPartition(msgId));
+        }
+        expect(partitions.size).toBe(1);
+      }, 30000);
     });
   });
 })();