Need help with adding Key_Based batching support to producer config

[joshua-motorq]

#267

I am working on adding Key_Based batching to the producer configuration. @BewareMyPower thank you for pointing me in this direction

I just have some questions regarding the C function use

I have updated the type in index.d.ts

export interface ProducerConfig {
  topic: string;
  producerName?: string;
  sendTimeoutMs?: number;
  initialSequenceId?: number;
  maxPendingMessages?: number;
  maxPendingMessagesAcrossPartitions?: number;
  blockIfQueueFull?: boolean;
  messageRoutingMode?: MessageRoutingMode;
  hashingScheme?: HashingScheme;
  compressionType?: CompressionType;
  batchingEnabled?: boolean;
  batchingMaxPublishDelayMs?: number;
  batchingMaxMessages?: number;
  _**batchingType?: BatchingType;**_
  properties?: { [key: string]: string };
  publicKeyPath?: string;
  encryptionKey?: string;
  cryptoFailureAction?: ProducerCryptoFailureAction;
  chunkingEnabled?: boolean;
  schema?: SchemaInfo;
  accessMode?: ProducerAccessMode;
}

export type BatchingType =
  'DefaultBatching' |
  'KeyBasedBatching';

I am having some difficulty with the setting of this property.

I was not able to find a direct function like pulsar_producer_configuration_set_batching_type similar to how pulsar_producer_configuration_set_access_mode exists

So i went with using the pulsar_producer_configuration_set_property function.

if (producerConfig.Has(CFG_BATCHING_TYPE) && producerConfig.Get(CFG_BATCHING_TYPE).IsString()) {
    std::string batchingType = producerConfig.Get(CFG_BATCHING_TYPE).ToString().Utf8Value();
    pulsar_producer_configuration_set_property(this->cProducerConfig.get(), "batchingType",
                                               batchingType.c_str());
  }

However, i am not able to see the expected behaviour when i test after installing the library locally. The entire batch gets sent to one consumer (based on the first key) behaving the same as default batching

I will create a PR for this once i verify it working

Could i please get some help with this?

[shibd]

I was not able to find a direct function like pulsar_producer_configuration_set_batching_type similar to how pulsar_producer_configuration_set_access_mode exists

hi, @joshua-motorq

Yes, the C pulsar client hasn't exposed this method yet. A workaround is to directly use the CPP method in node.js, which requires redefining the class, similar to this approach.

struct _pulsar_producer_configuration {
  pulsar::ProducerConfiguration conf;
};

For detailed, check the PR: #418

[joshua-motorq]

Got it, will check this out!

[joshua-motorq]

Thanks for the PR as well and showing how its done!