You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by GitBox <gi...@apache.org> on 2020/03/25 01:40:26 UTC
[GitHub] [samza] bkonold opened a new pull request #1327: SAMZA-2493: Keep
checkpoint manager consumer open for repeated polling
bkonold opened a new pull request #1327: SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
URL: https://github.com/apache/samza/pull/1327
**Feature:** Allow control of whether the underlying consumer backing a checkpoint manager will be left open (until stopped), or closed after the initial read of checkpoints
**Changes:** Introduces an internal config, injected by SamzaContainer, that is injected dependent on whether the container is standby or not, to control whether the container's checkpoint manager should have it's underlying consumer stopped after initial read or not.
**Tests:** TBD
**API Changes:** None
**Upgrade Instructions:** None
**Usage Instructions:** None
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] mynameborat commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
mynameborat 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_r403293161
##########
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:
I agree. For now, the config approach is minimally invasive and backward compatible. We can always reevaluate this if there is a stronger justification to have container model exposed to the checkpoint manager.
Since it is not a user config, we should be able to deprecate it without much effort.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
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_r401947082
##########
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:
I spoke with @rmatharu and there is nothing in config we can use to infer standby vs not. If we're all ok with the internal config, I will change the prefix and we can go with that.
May also be worth pointing out that this PR doesn't do much to lock us into anything - if we later want to treat this information as more a first class citizen we can go ahead and change the factory API.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] mynameborat merged pull request #1327: SAMZA-2493: Keep
checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
mynameborat merged pull request #1327: SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
URL: https://github.com/apache/samza/pull/1327
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
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_r403364372
##########
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 INTERNAL_CHECKPOINT_MANAGER_CONSUMER_STOP_AFTER_FIRST_READ = "samza.internal.task.checkpoint.consumer.stop.after.first.read";
Review comment:
I initially thought of something like `samza.internal.task.checkpoint.consumer.keep.open` but thought that the name in the PR currently is more expressive of behavior.
Thoughts?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on issue #1327: SAMZA-2493: Keep
checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
bkonold commented on issue #1327: SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
URL: https://github.com/apache/samza/pull/1327#issuecomment-610095860
> Do you plan to add tests? I noticed TBD in the description.
@mynameborat yeah, added them
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
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_r404314979
##########
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:
Closing this thread then. Thanks.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] rmatharu commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
rmatharu 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_r402700000
##########
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 INTERNAL_CHECKPOINT_MANAGER_CONSUMER_STOP_AFTER_FIRST_READ = "samza.internal.task.checkpoint.consumer.stop.after.first.read";
Review comment:
Lets maybe add a one-line description of what this config does?
Also if it'd be possible to shrink the name somewhat.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on issue #1327: SAMZA-2493: Keep
checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
bkonold commented on issue #1327: SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
URL: https://github.com/apache/samza/pull/1327#issuecomment-606787955
@bharathkk @xinyuiscool can you take a look over this?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] prateekm commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
prateekm 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_r401138531
##########
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:
Why do this as a config instead of a parameter?
If internal config, let's use samza.internal prefix.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
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_r404314795
##########
File path: samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
##########
@@ -76,6 +76,8 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
val producerRef: AtomicReference[SystemProducer] = new AtomicReference[SystemProducer](getSystemProducer())
val producerCreationLock: Object = new Object
+ val stopConsumerAfterFirstRead: Boolean = new TaskConfig(config).getCheckpointManagerConsumerStopAfterFirstRead
Review comment:
Done
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
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_r401142737
##########
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:
@prateekm The rationale between @mynameborat and I is that the internal config was a lighter-weight change.
Passing this via construction parameter would mean changing the `CheckpointManagerFactory` API and underlying implementations. If we're ok with that, I'm open to either approach.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] mynameborat commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
mynameborat 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_r401217615
##########
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:
+1 to internal prefix. We were debating between passing the exposing task mode through parameter vs config. The config route seemed less invasive and also doesn't introduce breaking API changes.
IIRC, we want to know if the container is standby or not. Given KafkaCheckpointManager is a singleton for the container, is there a way to infer on the fly from the existing configurations if it is standby or active? for e.g. task.inputs is empty?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] rmatharu commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
rmatharu 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_r403315717
##########
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:
makes sense
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] mynameborat commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
mynameborat 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_r403304166
##########
File path: samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
##########
@@ -76,6 +76,8 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
val producerRef: AtomicReference[SystemProducer] = new AtomicReference[SystemProducer](getSystemProducer())
val producerCreationLock: Object = new Object
+ val stopConsumerAfterFirstRead: Boolean = new TaskConfig(config).getCheckpointManagerConsumerStopAfterFirstRead
Review comment:
can we put in a comment on when this config is set vs not?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] bkonold commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] rmatharu commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
rmatharu 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_r402701011
##########
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:
Is adding a parameter like JobModel or containermodel, etc to CheckpointManagerFactory a big deal?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] mynameborat commented on issue #1327: SAMZA-2493: Keep
checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
mynameborat commented on issue #1327: SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
URL: https://github.com/apache/samza/pull/1327#issuecomment-609975896
Do you plan to add tests? I noticed TBD in the description.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza] rmatharu commented on a change in pull request #1327:
SAMZA-2493: Keep checkpoint manager consumer open for repeated polling
Posted by GitBox <gi...@apache.org>.
rmatharu 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_r403364700
##########
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 INTERNAL_CHECKPOINT_MANAGER_CONSUMER_STOP_AFTER_FIRST_READ = "samza.internal.task.checkpoint.consumer.stop.after.first.read";
Review comment:
sure lets go with that one then
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services