exceptionfactory commented on code in PR #115:
URL: https://github.com/apache/nifi-api/pull/115#discussion_r4564878765
##########
src/main/java/org/apache/nifi/components/connector/components/ProcessGroupLifecycle.java:
##########
@@ -51,4 +51,20 @@ public interface ProcessGroupLifecycle {
CompletableFuture<Void> stopStatelessGroups(ComponentHierarchyScope scope);
int getActiveThreadCount();
+
+ /**
+ * Begins draining the Process Group, so that no new data is brought into
the group while the data already queued in it is processed.
+ * The drain stays active until {@link #endDrain()} is called.
+ *
+ * @return a future that completes once the drain has begun, or completes
exceptionally if the drain could not begin
+ */
+ default CompletableFuture<Void> beginDrain() {
+ throw new UnsupportedOperationException("Draining is not supported by
" + getClass().getName());
Review Comment:
Is throwing an `UnsupportedOperationException` the best default approach for
this method? It looks like it would cause existing Connectors to throw the
exception in `drainFlowFiles()`. What about returning a completed Future
instead?
##########
src/main/java/org/apache/nifi/components/connector/AbstractConnector.java:
##########
@@ -202,9 +206,9 @@ public CompletableFuture<Void> drainFlowFiles(final
FlowContext flowContext) {
initialQueueSize.getObjectCount(),
NumberFormat.getNumberInstance().format(initialQueueSize.getByteCount()));
Thread.startVirtualThread(() -> {
+ final ProcessGroupLifecycle lifecycle =
flowContext.getRootGroup().getLifecycle();
try {
- stopSourceComponents(flowContext).get();
- startNonSourceComponents(result, flowContext);
+ lifecycle.beginDrain().get();
Review Comment:
It looks like this `get()` could continue on indefinitely under worst-case
circumstances. What do you think about adding some eventual timeout, even if
long, like an hour or more?
##########
src/main/java/org/apache/nifi/components/connector/AbstractConnector.java:
##########
@@ -202,9 +206,9 @@ public CompletableFuture<Void> drainFlowFiles(final
FlowContext flowContext) {
initialQueueSize.getObjectCount(),
NumberFormat.getNumberInstance().format(initialQueueSize.getByteCount()));
Thread.startVirtualThread(() -> {
+ final ProcessGroupLifecycle lifecycle =
flowContext.getRootGroup().getLifecycle();
try {
- stopSourceComponents(flowContext).get();
Review Comment:
Since `stopSourceComponents()` is no longer called, it seems like the method
should be marked as Deprecated so that implementations can remove overrides,
providing a path for eventual removal.
##########
src/main/java/org/apache/nifi/annotation/lifecycle/OnDrainInitiated.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.nifi.annotation.lifecycle;
+
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.ProcessContext;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * <p>
+ * Marker annotation a {@link org.apache.nifi.processor.Processor Processor} or
+ * {@link org.apache.nifi.controller.ControllerService ControllerService}
implementation
+ * can use to indicate that a method should be called when a drain of its
Process Group begins.
+ * While a drain is in progress, no new data is brought into the group, so the
component should
+ * process the data that it already has rather than wait for more. For
example, a Processor that
+ * groups FlowFiles into bins might send its bins on sooner.
+ * </p>
+ *
+ * <p>
+ * The component may be in any lifecycle state when this method is called, and
the method may be
Review Comment:
It would be helpful to indicate when the framework expects to call these
methods, particularly in relation to `OnScheduled`
##########
src/main/java/org/apache/nifi/components/connector/components/ProcessGroupLifecycle.java:
##########
@@ -51,4 +51,20 @@ public interface ProcessGroupLifecycle {
CompletableFuture<Void> stopStatelessGroups(ComponentHierarchyScope scope);
int getActiveThreadCount();
+
+ /**
+ * Begins draining the Process Group, so that no new data is brought into
the group while the data already queued in it is processed.
+ * The drain stays active until {@link #endDrain()} is called.
+ *
+ * @return a future that completes once the drain has begun, or completes
exceptionally if the drain could not begin
+ */
+ default CompletableFuture<Void> beginDrain() {
+ throw new UnsupportedOperationException("Draining is not supported by
" + getClass().getName());
+ }
+
+ /**
+ * Ends the active drain of the Process Group, if there is one. Ending a
drain does not stop any components.
+ */
+ default void endDrain() {
Review Comment:
The caller in AbstractConnector does not necessarily know or check that the
drain has completed based on this interface contract. Does it make sense to
introduce some kind of DrainFuture extension of `CompletableFuture` to provide
a more concrete relationship to an active drain? Or is the intent that there is
only ever one active drain, if any?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
