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 2022/06/23 17:42:20 UTC
[GitHub] [samza] ajothomas opened a new pull request, #1616: SAMZA-2751: [Pipeline Drain] [WIP] RunLoop and High Level API changes for Drain
ajothomas opened a new pull request, #1616:
URL: https://github.com/apache/samza/pull/1616
**Feature:**
[Pipeline Drain] SAMZA-2741 aims to support the ability to drain pipelines to allow incompatible intermediate schema changes.
This PR is a part of the Pipeline Drain work and aims to make changes to RunLoop processing and high level API to allow pipelines to drain
**Changes:**
- DrainMessage is the control message to trigger drain
- RunLoop Changes
- SystemConsumers adds DrainMessage to it's queue for the chooser to choose from
- DrainMessages are processed by RunLoopTask and shuts down the task when all SSPs for a task have encountered drain
- Added DrainListenerTask modifier interface for StreamTask
- Sync commits on the RunLoop thread on Drain
- High level API changes
- Handling and propagation of DrainMessages through the DAG in StreamOperatorTask
**Tests:**
- Unit tests for Drain in RunLoop
- (Pending) integ test for Drain in SamzaContainer
- (Pending) integ test for High level API
API changes:
No public API change.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r930410730
##########
samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala:
##########
@@ -72,6 +72,8 @@ class TaskInstance(
val taskName: TaskName = taskModel.getTaskName
val isInitableTask = task.isInstanceOf[InitableTask]
+ //val isDrainTask = classOf[DrainListenerTask].isAssignableFrom(task.getClass)
Review Comment:
Done. Thanks for catching 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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r931670317
##########
samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java:
##########
@@ -34,14 +34,15 @@
public class RunLoopFactory {
private static final Logger log = LoggerFactory.getLogger(RunLoopFactory.class);
- public static Runnable createRunLoop(scala.collection.immutable.Map<TaskName, RunLoopTask> taskInstances,
+ public static RunLoop createRunLoop(scala.collection.immutable.Map<TaskName, RunLoopTask> taskInstances,
Review Comment:
Made some changes which don't require the return type to be `RunLoop` anymore. Reverted to `Runnable`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r923696627
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -514,18 +548,46 @@ private void run() {
case END_OF_STREAM:
endOfStream();
break;
+ case DRAIN:
+ drain();
+ break;
default:
//no op
break;
}
}
+ /**
+ * Called when a task has drained i.e all SSPs for the task have received a drain message.
+ * */
+ private void drain() {
+ state.complete = true;
+ state.startDrain();
+ try {
+ ReadableCoordinator coordinator = new ReadableCoordinator(task.taskName());
+
+ task.drain(coordinator);
+
+ // issue a shutdown request for the task
+ coordinator.shutdown(TaskCoordinator.RequestScope.CURRENT_TASK);
+ coordinatorRequests.update(coordinator);
+
+ // issue a commit explicitly before we shutdown the task
+ // Adding commit to coordinator will not work as the state is marked complete and NO_OP will always be the
+ // next operation for this task
+ commit();
Review Comment:
Yes- We do the following:
- set the isAsyncCommitEnabled to false
- commits will happen on the RunLoop thread instead of the commit executor
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r927158298
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -805,6 +877,52 @@ private boolean checkEndOfStream() {
return processingSspSet.isEmpty();
}
+ private boolean shouldDrain() {
+ if (endOfStream) {
+ return false;
+ }
+
+ if (!pendingEnvelopeQueue.isEmpty()) {
+ PendingEnvelope pendingEnvelope = pendingEnvelopeQueue.peek();
+ IncomingMessageEnvelope envelope = pendingEnvelope.envelope;
+
+ if (envelope.isDrain()) {
+ final DrainMessage message = (DrainMessage) envelope.getMessage();
+ if (!message.getDeploymentId().equals(deploymentId)) {
+ // Removing the drain message from the pending queue as it doesn't match with the current deploymentId
+ // Removing it will ensure that it is not picked up by process()
+ pendingEnvelopeQueue.remove();
+ } else {
+ if (elasticityFactor <= 1) {
+ SystemStreamPartition ssp = envelope.getSystemStreamPartition();
+ processingSspSetToDrain.remove(ssp);
+ } else {
+ // if envelope contains drain message,
+ // the ssp of envelope should be removed from task's processing set irresp of keyBucket
+ SystemStreamPartition sspOfEnvelope = envelope.getSystemStreamPartition();
+ Optional<SystemStreamPartition> ssp = processingSspSetToDrain.stream()
+ .filter(sspInSet -> sspInSet.getSystemStream().equals(sspOfEnvelope.getSystemStream())
+ && sspInSet.getPartition().equals(sspOfEnvelope.getPartition()))
+ .findFirst();
+ ssp.ifPresent(processingSspSetToDrain::remove);
+ }
+
+ if (!hasIntermediateStreams) {
Review Comment:
This condition is necessary for high level API otherwise RunLoop will simply stop the task when drain message is seen for the final SSP. We want the drain message to go be processed by the high-level API DAG. This holds true for source SSPs and intermediate SSPs alike.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r927158471
##########
samza-api/src/main/java/org/apache/samza/system/DrainMessage.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.system;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * The DrainMessage is a control message that is sent out to next stage
+ * once the task has consumed to the end of a bounded stream.
+ */
+public class DrainMessage extends ControlMessage {
+ private final String deploymentId;
Review Comment:
Changed it.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] xinyuiscool merged pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
xinyuiscool merged PR #1616:
URL: https://github.com/apache/samza/pull/1616
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] mynameborat commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
mynameborat commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r927135337
##########
samza-core/src/main/java/org/apache/samza/config/JobConfig.java:
##########
@@ -169,6 +169,9 @@ public class JobConfig extends MapConfig {
public static final String DRAIN_MONITOR_ENABLED = "samza.drain-monitor.enabled";
public static final boolean DRAIN_MONITOR_ENABLED_DEFAULT = false;
+ public static final String DRAIN_MONITOR_POLL_INTERVAL_MILLIS = "samza.drain-monitor.poll.interval.ms";
+ public static final long DRAIN_MONITOR_POLL_INTERVAL_MILLIS_DEFAULT = 60_000;
Review Comment:
Can we be consistent and call these as `job.drain-monitor...` instead of `samza.drain-monitor...`? Same for enabled vs disabled config.
##########
samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala:
##########
@@ -613,14 +638,20 @@ class TaskInstance(
// if elasticityFactor > 1, find the SSP with keyBucket
var incomingMessageSsp = envelope.getSystemStreamPartition(elasticityFactor)
- // if envelope is end of stream or watermark, it needs to be routed to all tasks consuming the ssp irresp of keyBucket
+ // if envelope is end of stream or watermark or drain,
+ // it needs to be routed to all tasks consuming the ssp irresp of keyBucket
val messageType = MessageType.of(envelope.getMessage)
- if (envelope.isEndOfStream() || MessageType.END_OF_STREAM == messageType || MessageType.WATERMARK == messageType) {
+ if (envelope.isEndOfStream()
+ || envelope.isDrain()
+ || messageType == MessageType.END_OF_STREAM
+ || messageType == MessageType.DRAIN
+ || messageType == MessageType.WATERMARK) {
Review Comment:
what is the scenario where the message envelope will have `envelope.isDrain()` vs when `messageType == MessageType.DRAIN`?
Wondering why you need both.
##########
samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java:
##########
@@ -34,14 +34,15 @@
public class RunLoopFactory {
private static final Logger log = LoggerFactory.getLogger(RunLoopFactory.class);
- public static Runnable createRunLoop(scala.collection.immutable.Map<TaskName, RunLoopTask> taskInstances,
+ public static RunLoop createRunLoop(scala.collection.immutable.Map<TaskName, RunLoopTask> taskInstances,
Review Comment:
why is this change in return type needed?
##########
samza-api/src/main/java/org/apache/samza/system/DrainMessage.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.system;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * The DrainMessage is a control message that is sent out to next stage
+ * once the task has consumed to the end of a bounded stream.
+ */
+public class DrainMessage extends ControlMessage {
+ /**
+ * Id used to invalidate DrainMessages between runs. Ties to app.run.id from config.
+ */
+ private final String runId;
+
+ public DrainMessage(String runId) {
+ this(null, runId);
+ }
+
+ public DrainMessage(@JsonProperty("task-name") String taskName, @JsonProperty("run-id") String runId) {
Review Comment:
can you follow the existing pattern of using Mixin instead of adding annotations to the model classes? E.g., https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/serializers/model/JsonJobCoordinatorMetadataMixIn.java
##########
samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/NamespaceAwareCoordinatorStreamStore.java:
##########
@@ -35,6 +35,10 @@ public class NamespaceAwareCoordinatorStreamStore implements MetadataStore {
private final MetadataStore metadataStore;
private final String namespace;
+ public MetadataStore getMetadataStore() {
+ return metadataStore;
+ }
+
Review Comment:
Why is this needed? If its used in tests, can we make it package private and add annotation `VisibleForTesting` instead?
##########
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java:
##########
@@ -934,7 +941,9 @@ private void startSideInputs() {
taskConfig.getMaxIdleMs(),
sideInputContainerMetrics,
System::nanoTime,
- false); // commit must be synchronous to ensure integrity of state flush
+ false,
+ DEFAULT_SIDE_INPUT_ELASTICITY_FACTOR,
Review Comment:
does this change have any impact? what is the elasticity factor defaults to with the constructor prior to this change? Is this the same as `DEFAULT_SIDE_INPUT_ELASTICITY_FACTOR`
##########
samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala:
##########
@@ -72,6 +72,8 @@ class TaskInstance(
val taskName: TaskName = taskModel.getTaskName
val isInitableTask = task.isInstanceOf[InitableTask]
+ //val isDrainTask = classOf[DrainListenerTask].isAssignableFrom(task.getClass)
Review Comment:
Remove commented code?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r930412218
##########
samza-api/src/main/java/org/apache/samza/system/DrainMessage.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.system;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * The DrainMessage is a control message that is sent out to next stage
+ * once the task has consumed to the end of a bounded stream.
+ */
+public class DrainMessage extends ControlMessage {
+ /**
+ * Id used to invalidate DrainMessages between runs. Ties to app.run.id from config.
+ */
+ private final String runId;
+
+ public DrainMessage(String runId) {
+ this(null, runId);
+ }
+
+ public DrainMessage(@JsonProperty("task-name") String taskName, @JsonProperty("run-id") String runId) {
Review Comment:
I wasn't aware that we were using jackson mixin to model classes. Thanks.
DrainMessage, however, extends a ControlMessage base class which in turn is used by a couple of other classes. Should we still be using mixin for cases like these?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] xinyuiscool commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
xinyuiscool commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r926976861
##########
samza-api/src/main/java/org/apache/samza/system/DrainMessage.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.system;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * The DrainMessage is a control message that is sent out to next stage
+ * once the task has consumed to the end of a bounded stream.
+ */
+public class DrainMessage extends ControlMessage {
+ private final String deploymentId;
Review Comment:
As discussed offline, let's make it consistent with the current Samza term which is "runId".
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -805,6 +877,52 @@ private boolean checkEndOfStream() {
return processingSspSet.isEmpty();
}
+ private boolean shouldDrain() {
+ if (endOfStream) {
+ return false;
+ }
+
+ if (!pendingEnvelopeQueue.isEmpty()) {
+ PendingEnvelope pendingEnvelope = pendingEnvelopeQueue.peek();
+ IncomingMessageEnvelope envelope = pendingEnvelope.envelope;
+
+ if (envelope.isDrain()) {
+ final DrainMessage message = (DrainMessage) envelope.getMessage();
+ if (!message.getDeploymentId().equals(deploymentId)) {
+ // Removing the drain message from the pending queue as it doesn't match with the current deploymentId
+ // Removing it will ensure that it is not picked up by process()
+ pendingEnvelopeQueue.remove();
+ } else {
+ if (elasticityFactor <= 1) {
+ SystemStreamPartition ssp = envelope.getSystemStreamPartition();
+ processingSspSetToDrain.remove(ssp);
+ } else {
+ // if envelope contains drain message,
+ // the ssp of envelope should be removed from task's processing set irresp of keyBucket
+ SystemStreamPartition sspOfEnvelope = envelope.getSystemStreamPartition();
+ Optional<SystemStreamPartition> ssp = processingSspSetToDrain.stream()
+ .filter(sspInSet -> sspInSet.getSystemStream().equals(sspOfEnvelope.getSystemStream())
+ && sspInSet.getPartition().equals(sspOfEnvelope.getPartition()))
+ .findFirst();
+ ssp.ifPresent(processingSspSetToDrain::remove);
+ }
+
+ if (!hasIntermediateStreams) {
Review Comment:
Is this condition strong enough? Seems to me that we want to check whether this drain message is coming from an input stream v.s. intermediate stream? The task might consume both input stream and intermediate stream partitions.
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -805,6 +877,52 @@ private boolean checkEndOfStream() {
return processingSspSet.isEmpty();
}
+ private boolean shouldDrain() {
+ if (endOfStream) {
+ return false;
+ }
+
+ if (!pendingEnvelopeQueue.isEmpty()) {
+ PendingEnvelope pendingEnvelope = pendingEnvelopeQueue.peek();
+ IncomingMessageEnvelope envelope = pendingEnvelope.envelope;
+
+ if (envelope.isDrain()) {
+ final DrainMessage message = (DrainMessage) envelope.getMessage();
+ if (!message.getDeploymentId().equals(deploymentId)) {
+ // Removing the drain message from the pending queue as it doesn't match with the current deploymentId
+ // Removing it will ensure that it is not picked up by process()
+ pendingEnvelopeQueue.remove();
+ } else {
+ if (elasticityFactor <= 1) {
+ SystemStreamPartition ssp = envelope.getSystemStreamPartition();
+ processingSspSetToDrain.remove(ssp);
+ } else {
+ // if envelope contains drain message,
Review Comment:
Maybe a little bit more explaination here about the log to handle elasticity. Seems this feature is making things much complex.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r927158579
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -805,6 +877,52 @@ private boolean checkEndOfStream() {
return processingSspSet.isEmpty();
}
+ private boolean shouldDrain() {
+ if (endOfStream) {
+ return false;
+ }
+
+ if (!pendingEnvelopeQueue.isEmpty()) {
+ PendingEnvelope pendingEnvelope = pendingEnvelopeQueue.peek();
+ IncomingMessageEnvelope envelope = pendingEnvelope.envelope;
+
+ if (envelope.isDrain()) {
+ final DrainMessage message = (DrainMessage) envelope.getMessage();
+ if (!message.getDeploymentId().equals(deploymentId)) {
+ // Removing the drain message from the pending queue as it doesn't match with the current deploymentId
+ // Removing it will ensure that it is not picked up by process()
+ pendingEnvelopeQueue.remove();
+ } else {
+ if (elasticityFactor <= 1) {
+ SystemStreamPartition ssp = envelope.getSystemStreamPartition();
+ processingSspSetToDrain.remove(ssp);
+ } else {
+ // if envelope contains drain message,
Review Comment:
Added extra explanation.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r930409996
##########
samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/NamespaceAwareCoordinatorStreamStore.java:
##########
@@ -35,6 +35,10 @@ public class NamespaceAwareCoordinatorStreamStore implements MetadataStore {
private final MetadataStore metadataStore;
private final String namespace;
+ public MetadataStore getMetadataStore() {
+ return metadataStore;
+ }
+
Review Comment:
I think i added it for some test but not necessary anymore. Removed it.
##########
samza-core/src/main/java/org/apache/samza/config/JobConfig.java:
##########
@@ -169,6 +169,9 @@ public class JobConfig extends MapConfig {
public static final String DRAIN_MONITOR_ENABLED = "samza.drain-monitor.enabled";
public static final boolean DRAIN_MONITOR_ENABLED_DEFAULT = false;
+ public static final String DRAIN_MONITOR_POLL_INTERVAL_MILLIS = "samza.drain-monitor.poll.interval.ms";
+ public static final long DRAIN_MONITOR_POLL_INTERVAL_MILLIS_DEFAULT = 60_000;
Review Comment:
Did not notice that this wasn't consistent. Changed 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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] xinyuiscool commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
xinyuiscool commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r931322948
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -134,17 +138,26 @@ public RunLoop(Map<TaskName, RunLoopTask> runLoopTasks,
this.latch = new Object();
this.workerTimer = Executors.newSingleThreadScheduledExecutor();
this.clock = clock;
- Map<TaskName, AsyncTaskWorker> workers = new HashMap<>();
+ // assign runId before creating workers. As the inner AsyncTaskWorker class is not static, it relies on
+ // the outer class fields to be init first
+ this.runId = runId;
+ Map<TaskName, AsyncTaskWorker> workers = new HashMap<>();
for (RunLoopTask task : runLoopTasks.values()) {
workers.put(task.taskName(), new AsyncTaskWorker(task));
}
// Partions and tasks assigned to the container will not change during the run loop life time
this.sspToTaskWorkerMapping = Collections.unmodifiableMap(getSspToAsyncTaskWorkerMap(runLoopTasks, workers));
+
this.taskWorkers = Collections.unmodifiableList(new ArrayList<>(workers.values()));
this.isAsyncCommitEnabled = isAsyncCommitEnabled;
this.elasticityFactor = elasticityFactor;
}
+ public void drain() {
Review Comment:
As we discussed offline, this method can be removed as we can set these flags when runloop receives the drain message. It will also siimply the logic of invoking runloop and factory interface.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r923695705
##########
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala:
##########
@@ -237,6 +233,13 @@ class SystemConsumers (
}
}
+ def drain(): Unit = {
+ if (!isDraining) {
+ isDraining = true;
+ info("SystemConsumers is set to drain mode.")
+ writeDrainControlMessageToSspQueue()
Review Comment:
New envelopes will be added to SystemConsumers' in-memory queue from the registered consumers whenever refresh is called. When in drain mode, refresh will not add new messages from the consumers.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r931683098
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -134,17 +138,26 @@ public RunLoop(Map<TaskName, RunLoopTask> runLoopTasks,
this.latch = new Object();
this.workerTimer = Executors.newSingleThreadScheduledExecutor();
this.clock = clock;
- Map<TaskName, AsyncTaskWorker> workers = new HashMap<>();
+ // assign runId before creating workers. As the inner AsyncTaskWorker class is not static, it relies on
+ // the outer class fields to be init first
+ this.runId = runId;
+ Map<TaskName, AsyncTaskWorker> workers = new HashMap<>();
for (RunLoopTask task : runLoopTasks.values()) {
workers.put(task.taskName(), new AsyncTaskWorker(task));
}
// Partions and tasks assigned to the container will not change during the run loop life time
this.sspToTaskWorkerMapping = Collections.unmodifiableMap(getSspToAsyncTaskWorkerMap(runLoopTasks, workers));
+
this.taskWorkers = Collections.unmodifiableList(new ArrayList<>(workers.values()));
this.isAsyncCommitEnabled = isAsyncCommitEnabled;
this.elasticityFactor = elasticityFactor;
}
+ public void drain() {
Review Comment:
Made this private. `SamzaContainer` won't call this anymore.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r930410422
##########
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java:
##########
@@ -934,7 +941,9 @@ private void startSideInputs() {
taskConfig.getMaxIdleMs(),
sideInputContainerMetrics,
System::nanoTime,
- false); // commit must be synchronous to ensure integrity of state flush
+ false,
+ DEFAULT_SIDE_INPUT_ELASTICITY_FACTOR,
Review Comment:
Yes, it defaults to 1 currently even in RunLoop.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] ajothomas commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
ajothomas commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r930392479
##########
samza-core/src/main/java/org/apache/samza/container/RunLoopFactory.java:
##########
@@ -34,14 +34,15 @@
public class RunLoopFactory {
private static final Logger log = LoggerFactory.getLogger(RunLoopFactory.class);
- public static Runnable createRunLoop(scala.collection.immutable.Map<TaskName, RunLoopTask> taskInstances,
+ public static RunLoop createRunLoop(scala.collection.immutable.Map<TaskName, RunLoopTask> taskInstances,
Review Comment:
The reason for the change is that currently the `RunLoop` is only a `Runnable` in the`SamzaContainer`. We need to be able to set the `RunLoop` to drain mode by calling it's `drain` method from `SamzaContainer`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [samza] xinyuiscool commented on a diff in pull request #1616: SAMZA-2751: [Pipeline Drain] [WIP] RunLoop and High Level API changes for Drain
Posted by GitBox <gi...@apache.org>.
xinyuiscool commented on code in PR #1616:
URL: https://github.com/apache/samza/pull/1616#discussion_r906439300
##########
samza-core/src/main/java/org/apache/samza/container/RunLoop.java:
##########
@@ -514,18 +548,46 @@ private void run() {
case END_OF_STREAM:
endOfStream();
break;
+ case DRAIN:
+ drain();
+ break;
default:
//no op
break;
}
}
+ /**
+ * Called when a task has drained i.e all SSPs for the task have received a drain message.
+ * */
+ private void drain() {
+ state.complete = true;
+ state.startDrain();
+ try {
+ ReadableCoordinator coordinator = new ReadableCoordinator(task.taskName());
+
+ task.drain(coordinator);
+
+ // issue a shutdown request for the task
+ coordinator.shutdown(TaskCoordinator.RequestScope.CURRENT_TASK);
+ coordinatorRequests.update(coordinator);
+
+ // issue a commit explicitly before we shutdown the task
+ // Adding commit to coordinator will not work as the state is marked complete and NO_OP will always be the
+ // next operation for this task
+ commit();
Review Comment:
Do we make sure this commit will not be async?
##########
samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala:
##########
@@ -237,6 +233,13 @@ class SystemConsumers (
}
}
+ def drain(): Unit = {
+ if (!isDraining) {
+ isDraining = true;
+ info("SystemConsumers is set to drain mode.")
+ writeDrainControlMessageToSspQueue()
Review Comment:
Given Kafka consumer will spawn up a thread, seems it's possible in a small chance to have more envelopes being added after the drain message. For example, the consumer thread is trying to pull the consumer, and we get the drain command at the same time. Is this situation being handled? It will be good to write that in the design doc.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@samza.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org