bkonold commented on a change in pull request #1327: SAMZA-2493: Keep checkpoint manager consumer open for repeated polling URL: https://github.com/apache/samza/pull/1327#discussion_r403261714
########## File path: samza-core/src/main/java/org/apache/samza/config/TaskConfig.java ########## @@ -105,6 +105,7 @@ private static final String BROADCAST_STREAM_PATTERN = "^[\\d]+$"; private static final String BROADCAST_STREAM_RANGE_PATTERN = "^\\[[\\d]+\\-[\\d]+\\]$"; public static final String CHECKPOINT_MANAGER_FACTORY = "task.checkpoint.factory"; + public static final String CHECKPOINT_MANAGER_CONSUMER_STOP_AFTER_FIRST_READ = "task.checkpoint.consumer.stop.after.first.read"; Review comment: @rmatharu @mynameborat @prateekm You'll need some visibility on the calling-container, so jobModel + containerId or containerModel would work. At least one usage of CheckpointManagerFactory does not have this information available (it is used in StreamManager during execution planning, before the job model is created). I suppose we could make the parameter nullable in this case. There is also usage in our tooling (recovery and checkpoint tools) that expects an instance of CheckpointManager to be shared within the same job. It does however seem like these can be easily refactored. In addition to the two considerations above, changing the API would impact any existing open source customers who have implemented CheckpointManagerFactory for their own use cases. In this case perhaps we mark the existing method as deprecated and remove in a future version. IMO those are heavy changes to make just for one implementer of CheckpointManagerFactory to be able to toggle its consumer behavior. For context, stopping the consumer after initial read is an optimization (https://github.com/apache/samza/pull/993/files) because BlockingEnvelopeMap buffers were being filled with messages that would never be read. An alternative to this PR could be tackling that optimization differently. Fetch thresholds in BEM are currently configured per system. We wouldn't want to override this fetch threshold for the whole checkpoint system since it may overlap with other streams and impact job perf. However, if we implemented fetch thresholds at the system-stream / SSP granularity we could avoid this stop in some cases vs others bifurcation and instead limit buffering specifically for the checkpoint stream. TL;DR, my opinion: 1. This PR is the shortest path forward. 2. Modifying CheckpointManagerFactory API is unjustified. 3. Alternative to this PR would be to approach the original problem of optimization a bit differently. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
