Challenges with Current Implementation

Oct 13, 2025 by windailys.com 44 views

Introduction

In the current implementation of the consumer initialization, the batchReceivePolicy, which includes batchSize and timeout, must be specified when initializing the consumer. This approach, however, poses a challenge at high TPS because it necessitates checking the existence of a consumer each time the Read method is called, which can be time-consuming. In this article, we will re-evaluate the current approach and explore possible alternatives to initialize the consumer at the bin while utilizing the batchSize and timeout specified by the user in a not confusing, efficient way.

Current Implementation

The current implementation of the consumer initialization is as follows:

The consumer's batchReceivePolicy, which includes batchSize and timeout, must be specified when initializing the consumer.
These settings cannot be changed later.
Unlike PulsarClient and PulsarProducer, which are initialized at the bin using Spring beans, the consumer uses the batchSize and timeout defined in the ReadRequest object.
Therefore, we defer the creation of the consumer until the read method is first called.
If a consumer has not been created yet, the PulsarConsumerManager will create one at that time.
The ReadRequest object, containing the batchSize and timeout, is specified in the Limits section of the pipeline.

Challenges with Current Implementation

The current implementation poses a challenge at high TPS because it necessitates checking the existence of a consumer each time the Read method is called, which can be time-consuming. This can lead to performance issues and affect the overall efficiency of the system.

Possible Alternatives

There are two possible alternatives to the current implementation:

Alternative 1: Specifying Pulsar Batch Size and Timeout in ConfigMap

One possible alternative is to allow the user to specify the Pulsar batch size and timeout directly in the ConfigMap, while choosing the smaller values between the Pulsar's and Numaflow's settings. However, this approach has some drawbacks:

It gets complicated managing two batch sizes and two timeouts (the Pulsar one, and the Numaflow one).
For example, if the Pulsar batch size is 10, but the Numaflow batch size is 5, batchReceive would get 10 messages, but then we would need to ensure to not ack 5 of them and return them to the beginning of the backlog.

Alternative 2: Foregoing Numaflow’s Specified Values in the Pipeline

Another possible alternative is to forego Numaflow’s specified values in the Pipeline and only use those provided by the user in the ConfigMap to initialize the consumer at the bin. This method simplifies configuration but might not align with best practices.

Current Implementation Limitations

The current implementation does not permit user-specified batchSize and timeout in the ConfigMap, but these settings can be configured directly in the pipeline, which is consistent with typical settings in Numaflow. For more details, please refer to Numaflow’s API documentation: https://numaflow.numaproj.io/APIs/.

Conclusion

In conclusion, the current implementation of the consumer initialization poses a challenge at high TPS because it necessitates checking the existence of a consumer each time the Read method is called, which can be time-consuming. We need to re-evaluate if we want to continue with this approach or if there is a better way to initialize the consumer at the bin while utilizing the batchSize and timeout specified by the user in a not confusing, efficient way. The two possible alternatives discussed include specifying Pulsar batch size and timeout in ConfigMap and foregoing Numaflow’s specified values in the Pipeline. We need to carefully consider the pros and cons of each alternative and choose the best approach for our use case.

Recommendations

Based on the analysis, we recommend the following:

Re-evaluate the current implementation and explore possible alternatives to initialize the consumer at the bin while utilizing the batchSize and timeout specified by the user in a not confusing, efficient way.
Consider specifying Pulsar batch size and timeout in ConfigMap, but carefully manage two batch sizes and two timeouts.
Forego Numaflow’s specified values in the Pipeline and only use those provided by the user in the ConfigMap to initialize the consumer at the bin.
Simplify configuration by using user-specified values in the ConfigMap, but ensure that it aligns with best practices.

Future Work

Future work includes:

Implementing the recommended approach and testing its performance.
Continuously monitoring the system's performance and making adjustments as needed.
Exploring other possible alternatives and evaluating their pros and cons.

Introduction

In our previous article, we discussed the current implementation of the consumer initialization and its limitations. We also explored possible alternatives to initialize the consumer at the bin while utilizing the batchSize and timeout specified by the user in a not confusing, efficient way. In this article, we will answer some frequently asked questions (FAQs) related to the re-evaluation of the consumer initialization.

Q: What is the current implementation of the consumer initialization?

A: The current implementation of the consumer initialization is as follows:

The consumer's batchReceivePolicy, which includes batchSize and timeout, must be specified when initializing the consumer.
These settings cannot be changed later.
Unlike PulsarClient and PulsarProducer, which are initialized at the bin using Spring beans, the consumer uses the batchSize and timeout defined in the ReadRequest object.
Therefore, we defer the creation of the consumer until the read method is first called.
If a consumer has not been created yet, the PulsarConsumerManager will create one at that time.
The ReadRequest object, containing the batchSize and timeout, is specified in the Limits section of the pipeline.

Q: What are the challenges with the current implementation?

A: The current implementation poses a challenge at high TPS because it necessitates checking the existence of a consumer each time the Read method is called, which can be time-consuming. This can lead to performance issues and affect the overall efficiency of the system.

Q: What are the possible alternatives to the current implementation?

A: There are two possible alternatives to the current implementation:

Alternative 1: Specifying Pulsar Batch Size and Timeout in ConfigMap

It gets complicated managing two batch sizes and two timeouts (the Pulsar one, and the Numaflow one).
For example, if the Pulsar batch size is 10, but the Numaflow batch size is 5, batchReceive would get 10 messages, but then we would need to ensure to not ack 5 of them and return them to the beginning of the backlog.

Alternative 2: Foregoing Numaflow’s Specified Values in the Pipeline

Q: What are the pros and cons of each alternative?

A: The pros and cons of each alternative are as follows:

Alternative 1: Specifying Pulsar Batch Size and Timeout in ConfigMap

Pros:

Simplifies configuration by allowing users to specify batch size and timeout directly in the ConfigMap.
Aligns with best practices by using user-specified values.

Cons:

Gets complicated managing two batch sizes and two timeouts (the Pulsar one, and the Numaflow one).
May lead to performance issues if not implemented correctly.

Alternative 2: Foregoing Numaflow’s Specified Values in the Pipeline

Pros:

Simplifies configuration by using user-specified values in the ConfigMap.
Aligns with best practices by using user-specified values.

Cons:

May not align with best practices if not implemented correctly.
May lead to performance issues if not implemented correctly.

Q: What is the recommended approach?

A: Based on the analysis, we recommend the following:

Re-evaluate the current implementation and explore possible alternatives to initialize the consumer at the bin while utilizing the batchSize and timeout specified by the user in a not confusing, efficient way.
Consider specifying Pulsar batch size and timeout in ConfigMap, but carefully manage two batch sizes and two timeouts.
Forego Numaflow’s specified values in the Pipeline and only use those provided by the user in the ConfigMap to initialize the consumer at the bin.
Simplify configuration by using user-specified values in the ConfigMap, but ensure that it aligns with best practices.

Q: What is the future work?

A: Future work includes:

Implementing the recommended approach and testing its performance.
Continuously monitoring the system's performance and making adjustments as needed.
Exploring other possible alternatives and evaluating their pros and cons.

By re-evaluating the current implementation and exploring possible alternatives, we can improve the efficiency and performance of the system and provide a better user experience.