markap14 commented on code in PR #11570: URL: https://github.com/apache/nifi/pull/11570#discussion_r3847336449
########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovedConnectionDrainCoordinator.java: ########## @@ -0,0 +1,539 @@ +/* + * 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.web; + +import org.apache.nifi.connectable.ConnectableType; +import org.apache.nifi.controller.ScheduledState; +import org.apache.nifi.web.api.dto.AffectedComponentDTO; +import org.apache.nifi.web.api.entity.AffectedComponentEntity; +import org.apache.nifi.web.util.CancellableTimedPause; +import org.apache.nifi.web.util.ComponentLifecycle; +import org.apache.nifi.web.util.InvalidComponentAction; +import org.apache.nifi.web.util.LifecycleManagementException; +import org.apache.nifi.web.util.Pause; + +import java.net.URI; +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; +import java.util.stream.Collectors; + +public final class RemovedConnectionDrainCoordinator { + static final Duration DEFAULT_DRAIN_TIMEOUT = Duration.ofSeconds(30); + private static final Duration DEFAULT_POLL_INTERVAL = Duration.ofMillis(250); + + private final RemovedConnectionDrainClassifier classifier; + private final PauseFactory pauseFactory; + private final Duration drainTimeout; + + public RemovedConnectionDrainCoordinator() { + this(new RemovedConnectionDrainClassifier(), new MonotonicPauseFactory(DEFAULT_POLL_INTERVAL, System::nanoTime), DEFAULT_DRAIN_TIMEOUT); + } + + RemovedConnectionDrainCoordinator(final RemovedConnectionDrainClassifier classifier, final PauseFactory pauseFactory, final Duration drainTimeout) { + this.classifier = Objects.requireNonNull(classifier, "Removed Connection Drain Classifier required"); + this.pauseFactory = Objects.requireNonNull(pauseFactory, "Pause Factory required"); + this.drainTimeout = Objects.requireNonNull(drainTimeout, "Drain Timeout required"); + } + + public DrainResult coordinateDrain(final FlowUpdateImpact flowUpdateImpact, final RemovedConnectionDrainClassifier.Context context, + final ComponentLifecycle componentLifecycle, final URI requestUri, final String groupId, + final CancellationHandle cancellationHandle) throws LifecycleManagementException { + Objects.requireNonNull(flowUpdateImpact, "Flow Update Impact required"); + Objects.requireNonNull(context, "Removed Connection Drain Context required"); + Objects.requireNonNull(componentLifecycle, "Component Lifecycle required"); + Objects.requireNonNull(requestUri, "Request URI required"); + Objects.requireNonNull(groupId, "Group ID required"); + Objects.requireNonNull(cancellationHandle, "Cancellation Handle required"); + + final RemovedConnectionDrainClassifier.Context queueAwareContext = createQueueAwareContext(flowUpdateImpact, context, componentLifecycle, requestUri); + final RemovedConnectionDrainClassifier.BatchResult batchResult = classifier.classify(flowUpdateImpact, queueAwareContext); + if (!batchResult.isSupported()) { + throw new LifecycleManagementException(buildClassificationFailureMessage(batchResult)); + } + + final Set<String> candidateConnectionIds = batchResult.connectionResults().stream() + .filter(result -> result.classification() == RemovedConnectionDrainClassifier.Classification.CANDIDATE) + .map(result -> result.connection().getConnectionInstanceId()) + .collect(Collectors.toCollection(LinkedHashSet::new)); + + if (candidateConnectionIds.isEmpty()) { + return DrainResult.success(Collections.emptySet(), Collections.emptySet()); + } + + final Map<String, AffectedComponentEntity> affectedComponentsById = flowUpdateImpact.getAffectedComponents().stream() + .collect(Collectors.toMap(AffectedComponentEntity::getId, entity -> entity, (left, right) -> left, LinkedHashMap::new)); + final Set<AffectedComponentEntity> componentsToStop = new LinkedHashSet<>(); + for (final String producerBarrierComponentId : batchResult.producerBarrierComponentIds()) { + final AffectedComponentEntity entity = getProducerBarrierEntity(affectedComponentsById, queueAwareContext, producerBarrierComponentId); + if (entity == null || entity.getComponent() == null) { + continue; + } + + if (isActive(entity.getComponent())) { + componentsToStop.add(entity); + } + } + + final DeadlinePause drainPause = pauseFactory.createDrainPause(drainTimeout); + cancellationHandle.setCancelCallback(drainPause::cancel); + + final Set<AffectedComponentEntity> drainStoppedComponents = new LinkedHashSet<>(); + try { + if (!componentsToStop.isEmpty()) { + final Set<AffectedComponentEntity> updatedStoppedComponents = componentLifecycle.scheduleComponents( + requestUri, groupId, componentsToStop, ScheduledState.STOPPED, drainPause, InvalidComponentAction.SKIP); + drainStoppedComponents.addAll(getStoppedComponents(componentsToStop, updatedStoppedComponents)); + + if (!allComponentsStopped(componentsToStop, updatedStoppedComponents)) { + if (cancellationHandle.isCancelled()) { + return restoreAfterCancellation(componentLifecycle, requestUri, groupId, candidateConnectionIds, drainStoppedComponents); + } + + final Set<String> producerBarrierIds = componentsToStop.stream() + .map(AffectedComponentEntity::getId) + .collect(Collectors.toCollection(LinkedHashSet::new)); + throw new LifecycleManagementException(buildStopTimeoutMessage(producerBarrierIds)); + } + } + + if (cancellationHandle.isCancelled()) { + return restoreAfterCancellation(componentLifecycle, requestUri, groupId, candidateConnectionIds, drainStoppedComponents); + } + + final boolean queuesDrained = componentLifecycle.waitForConnectionQueuesEmpty(requestUri, candidateConnectionIds, drainPause); + if (queuesDrained) { Review Comment: [claude-opus-4.8] **Cancellation is ignored when the queues drain successfully, and the drain-stopped producers are not restored.** The `isCancelled()` check at line 131 only runs when the wait returns `false`. `ClusterReplicationComponentLifecycle.waitForConnectionQueuesEmpty` returns `true` at line 672 as soon as it observes all queues empty, without consulting the pause. So if the user cancels the request in the same moment the queues drain, this returns success. What follows in `FlowUpdateResource`: 1. Line 368 merges the drain-stopped producers into `runningComponents`. 2. Line 375 stops the full affected set, including the destinations the coordinator had deliberately left running. 3. Line 377 returns because the request is cancelled, which happens before the `try`/`finally` at lines 429/489 that would restart anything. Both the producers and the destinations end up stopped. NiFi already leaves components stopped when a user cancels mid-stop, so that part is not new, but these particular producers were stopped by the framework rather than by the user's own step, and the documented contract of this class is to restore them on cancellation. Checking `cancellationHandle.isCancelled()` before returning success, and routing to `restoreAfterCancellation` when it is set, would make the contract hold. Worth a coordinator unit test for cancel-during-a-successful-wait; the existing cancellation test only covers the case where the wait returns `false`, which is exactly why this path slipped through. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovedConnectionDrainCoordinator.java: ########## @@ -0,0 +1,539 @@ +/* + * 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.web; + +import org.apache.nifi.connectable.ConnectableType; +import org.apache.nifi.controller.ScheduledState; +import org.apache.nifi.web.api.dto.AffectedComponentDTO; +import org.apache.nifi.web.api.entity.AffectedComponentEntity; +import org.apache.nifi.web.util.CancellableTimedPause; +import org.apache.nifi.web.util.ComponentLifecycle; +import org.apache.nifi.web.util.InvalidComponentAction; +import org.apache.nifi.web.util.LifecycleManagementException; +import org.apache.nifi.web.util.Pause; + +import java.net.URI; +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; +import java.util.stream.Collectors; + +public final class RemovedConnectionDrainCoordinator { Review Comment: [claude-opus-4.8] **No logging anywhere in the drain path.** Neither this class nor `RemovedConnectionDrainClassifier` declares a logger. From an operator's point of view, the framework stops a set of processors that the user did not ask to stop, waits up to 30 seconds, and then either proceeds or restores them, and `nifi-app.log` contains nothing about any of it. When an update takes an unexplained 30 extra seconds or fails with a drain timeout, there is no record of which connections were being drained, which producers were stopped, how much data remained, or whether restoration succeeded. Suggested minimum: * INFO when the drain begins, naming the connections to drain and the producers being stopped. * INFO or WARN on timeout with the remaining queue sizes, since that is the case an operator will actually investigate. * WARN when restoration fails, from this class. Today the only such log is `FlowUpdateResource` line 363, which covers the cancellation path only; the restoration failure in the catch block at line 136 is attached to the thrown exception and never logged here. * DEBUG per poll iteration in `waitForConnectionQueuesEmpty` showing which nodes reported and which queues are still non-empty. Without this, diagnosing an incomplete-coverage timeout in a cluster means attaching a debugger. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowUpdateResource.java: ########## @@ -208,12 +210,12 @@ protected Response initiateFlowUpdate(final String groupId, final T requestEntit final Set<String> unresolvedParameterProviders = serviceFacade.resolveParameterProviders(flowSnapshot, user); // Step 1: Determine which components will be affected by updating the flow Review Comment: [claude-opus-4.8] The numbered workflow description in the comment block starting at line 175 still describes the old sequence and does not mention the new pre-drain step, even though the step was inserted into the middle of the flow it documents. That comment is the first thing a maintainer reads before touching this method, and the step numbering it establishes (`// Step 1:` here, `// Steps 5-6:` further down) is now out of sync with what the code actually does. Worth updating it to include the drain step. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovedConnectionDrainCoordinator.java: ########## @@ -0,0 +1,539 @@ +/* + * 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.web; + +import org.apache.nifi.connectable.ConnectableType; +import org.apache.nifi.controller.ScheduledState; +import org.apache.nifi.web.api.dto.AffectedComponentDTO; +import org.apache.nifi.web.api.entity.AffectedComponentEntity; +import org.apache.nifi.web.util.CancellableTimedPause; +import org.apache.nifi.web.util.ComponentLifecycle; +import org.apache.nifi.web.util.InvalidComponentAction; +import org.apache.nifi.web.util.LifecycleManagementException; +import org.apache.nifi.web.util.Pause; + +import java.net.URI; +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; +import java.util.stream.Collectors; + +public final class RemovedConnectionDrainCoordinator { + static final Duration DEFAULT_DRAIN_TIMEOUT = Duration.ofSeconds(30); + private static final Duration DEFAULT_POLL_INTERVAL = Duration.ofMillis(250); + + private final RemovedConnectionDrainClassifier classifier; + private final PauseFactory pauseFactory; + private final Duration drainTimeout; + + public RemovedConnectionDrainCoordinator() { + this(new RemovedConnectionDrainClassifier(), new MonotonicPauseFactory(DEFAULT_POLL_INTERVAL, System::nanoTime), DEFAULT_DRAIN_TIMEOUT); + } + + RemovedConnectionDrainCoordinator(final RemovedConnectionDrainClassifier classifier, final PauseFactory pauseFactory, final Duration drainTimeout) { + this.classifier = Objects.requireNonNull(classifier, "Removed Connection Drain Classifier required"); + this.pauseFactory = Objects.requireNonNull(pauseFactory, "Pause Factory required"); + this.drainTimeout = Objects.requireNonNull(drainTimeout, "Drain Timeout required"); + } + + public DrainResult coordinateDrain(final FlowUpdateImpact flowUpdateImpact, final RemovedConnectionDrainClassifier.Context context, + final ComponentLifecycle componentLifecycle, final URI requestUri, final String groupId, + final CancellationHandle cancellationHandle) throws LifecycleManagementException { + Objects.requireNonNull(flowUpdateImpact, "Flow Update Impact required"); + Objects.requireNonNull(context, "Removed Connection Drain Context required"); + Objects.requireNonNull(componentLifecycle, "Component Lifecycle required"); + Objects.requireNonNull(requestUri, "Request URI required"); + Objects.requireNonNull(groupId, "Group ID required"); + Objects.requireNonNull(cancellationHandle, "Cancellation Handle required"); + + final RemovedConnectionDrainClassifier.Context queueAwareContext = createQueueAwareContext(flowUpdateImpact, context, componentLifecycle, requestUri); + final RemovedConnectionDrainClassifier.BatchResult batchResult = classifier.classify(flowUpdateImpact, queueAwareContext); + if (!batchResult.isSupported()) { + throw new LifecycleManagementException(buildClassificationFailureMessage(batchResult)); + } + + final Set<String> candidateConnectionIds = batchResult.connectionResults().stream() + .filter(result -> result.classification() == RemovedConnectionDrainClassifier.Classification.CANDIDATE) + .map(result -> result.connection().getConnectionInstanceId()) + .collect(Collectors.toCollection(LinkedHashSet::new)); + + if (candidateConnectionIds.isEmpty()) { + return DrainResult.success(Collections.emptySet(), Collections.emptySet()); + } + + final Map<String, AffectedComponentEntity> affectedComponentsById = flowUpdateImpact.getAffectedComponents().stream() + .collect(Collectors.toMap(AffectedComponentEntity::getId, entity -> entity, (left, right) -> left, LinkedHashMap::new)); + final Set<AffectedComponentEntity> componentsToStop = new LinkedHashSet<>(); + for (final String producerBarrierComponentId : batchResult.producerBarrierComponentIds()) { + final AffectedComponentEntity entity = getProducerBarrierEntity(affectedComponentsById, queueAwareContext, producerBarrierComponentId); + if (entity == null || entity.getComponent() == null) { + continue; + } + + if (isActive(entity.getComponent())) { + componentsToStop.add(entity); + } + } + + final DeadlinePause drainPause = pauseFactory.createDrainPause(drainTimeout); + cancellationHandle.setCancelCallback(drainPause::cancel); + + final Set<AffectedComponentEntity> drainStoppedComponents = new LinkedHashSet<>(); + try { + if (!componentsToStop.isEmpty()) { + final Set<AffectedComponentEntity> updatedStoppedComponents = componentLifecycle.scheduleComponents( + requestUri, groupId, componentsToStop, ScheduledState.STOPPED, drainPause, InvalidComponentAction.SKIP); + drainStoppedComponents.addAll(getStoppedComponents(componentsToStop, updatedStoppedComponents)); + + if (!allComponentsStopped(componentsToStop, updatedStoppedComponents)) { + if (cancellationHandle.isCancelled()) { + return restoreAfterCancellation(componentLifecycle, requestUri, groupId, candidateConnectionIds, drainStoppedComponents); + } + + final Set<String> producerBarrierIds = componentsToStop.stream() + .map(AffectedComponentEntity::getId) + .collect(Collectors.toCollection(LinkedHashSet::new)); + throw new LifecycleManagementException(buildStopTimeoutMessage(producerBarrierIds)); + } + } + + if (cancellationHandle.isCancelled()) { + return restoreAfterCancellation(componentLifecycle, requestUri, groupId, candidateConnectionIds, drainStoppedComponents); + } + + final boolean queuesDrained = componentLifecycle.waitForConnectionQueuesEmpty(requestUri, candidateConnectionIds, drainPause); + if (queuesDrained) { + return DrainResult.success(candidateConnectionIds, drainStoppedComponents); + } + + if (cancellationHandle.isCancelled()) { + return restoreAfterCancellation(componentLifecycle, requestUri, groupId, candidateConnectionIds, drainStoppedComponents); + } + + throw new LifecycleManagementException(buildQueueTimeoutMessage(candidateConnectionIds)); + } catch (final LifecycleManagementException e) { Review Comment: [claude-opus-4.8] **Producers can be left stopped with no restore when the drain wait throws an unchecked exception.** This catch handles only `LifecycleManagementException`, and the `finally` below only resets the cancel callback. But in a cluster the wait can throw unchecked exceptions that come straight from request replication. `ClusterReplicationComponentLifecycle.waitForConnectionQueuesEmpty` calls `createFlowFileListingRequest`, which replicates with `performVerification = true`: ```java // ClusterReplicationComponentLifecycle line 789 private AsyncClusterResponse replicateFlowFileListingRequest(final Set<NodeIdentifier> expectedNodes, final NiFiUser user, final String method, final URI requestUri) { return getRequestReplicator().replicate(expectedNodes, user, method, requestUri, Collections.emptyMap(), Collections.emptyMap(), true, true); } ``` `ThreadPoolRequestReplicator` throws two unchecked exceptions out of that call: * `IllegalClusterStateException` (line 365) when a node in `expectedNodes` is no longer `CONNECTED`. * `ConnectingNodeMutableRequestException` (line 708, via `verifyClusterState`) when any node is `CONNECTING`. This one applies because creating a listing request is a `POST`, so it is classified as a mutable request. Both extend `ClusterException` and are unchecked. `createFlowFileListingRequest` only catches `InterruptedException`, so they propagate out of the wait and past this catch block. The target node set is a snapshot taken once per poll iteration at line 646, so an ordinary membership change during the 30-second drain is enough to trigger this. `FlowUpdateResource` line 307 then catches `Exception` and fails the request, but nothing restores the producers. The user is left with a failed update and a set of processors the framework stopped on its own and never restarted, with nothing in `nifi-app.log` saying so. Suggest catching `Exception` (or at minimum `ClusterException`) here, or moving the restore into the `finally` so it runs regardless of how the try block exits. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowUpdateResource.java: ########## @@ -349,7 +353,21 @@ private void updateFlow(final String groupId, final ComponentLifecycle component final Set<AffectedComponentEntity> runningComponents = affectedComponents.stream() .filter(entity -> stoppableReferenceTypes.contains(entity.getComponent().getReferenceType())) .filter(entity -> isActive(entity.getComponent())) - .collect(Collectors.toSet()); + .collect(Collectors.toCollection(LinkedHashSet::new)); + + if ("update-requests".equals(requestType)) { Review Comment: [claude-opus-4.8] `"update-requests"` is a bare string literal here, and again at line 644 in `getUpdateFlowSteps`, while `VersionsResource` already defines constants for request types and passes this same value at line 1426. Since both the drain step itself and the progress-step list are gated on this exact string, a typo in a future caller silently disables draining rather than failing, and the two gates can drift apart from each other. Worth promoting to a shared constant alongside the existing request-type constants. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovedConnectionDrainClassifier.java: ########## @@ -0,0 +1,523 @@ +/* + * 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.web; + +import org.apache.nifi.components.validation.ValidationStatus; +import org.apache.nifi.connectable.Connectable; +import org.apache.nifi.connectable.ConnectableType; +import org.apache.nifi.connectable.Connection; +import org.apache.nifi.connectable.Port; +import org.apache.nifi.controller.ProcessorNode; +import org.apache.nifi.controller.ScheduledState; +import org.apache.nifi.controller.flow.FlowManager; +import org.apache.nifi.groups.ProcessGroup; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collector; +import java.util.stream.Collectors; + +public final class RemovedConnectionDrainClassifier { + private static final Set<ConnectableType> SUPPORTED_SOURCE_TYPES = Set.of( + ConnectableType.PROCESSOR, + ConnectableType.INPUT_PORT, + ConnectableType.OUTPUT_PORT, + ConnectableType.FUNNEL); + private static final Set<ConnectableType> SUPPORTED_PRODUCER_BARRIER_TYPES = Set.of( + ConnectableType.PROCESSOR, + ConnectableType.INPUT_PORT, + ConnectableType.OUTPUT_PORT); + private static final Set<ConnectableType> SUPPORTED_DESTINATION_TYPES = Set.of( + ConnectableType.PROCESSOR, + ConnectableType.INPUT_PORT, + ConnectableType.OUTPUT_PORT); + private static final Comparator<RemovedConnectionDescriptor> CONNECTION_ORDER = + Comparator.comparing(RemovedConnectionDescriptor::getConnectionInstanceId, Comparator.nullsLast(String::compareTo)) + .thenComparing(RemovedConnectionDescriptor::getConnectionVersionedId, Comparator.nullsLast(String::compareTo)); + + BatchResult classify(final FlowUpdateImpact flowUpdateImpact, final Context context) { + Objects.requireNonNull(flowUpdateImpact, "Flow Update Impact required"); + Objects.requireNonNull(context, "Removed Connection Drain Context required"); + + final List<RemovedConnectionDescriptor> orderedConnections = flowUpdateImpact.getRemovedConnections().stream() + .sorted(CONNECTION_ORDER) + .toList(); + + final Set<String> removedConnectionIds = orderedConnections.stream() + .map(RemovedConnectionDescriptor::getConnectionInstanceId) + .filter(Objects::nonNull) + .collect(toOrderedSet()); + + final Set<String> nonEmptyRemovedDestinationIds = new LinkedHashSet<>(); + final Map<String, ConnectionResult> initialResults = new LinkedHashMap<>(); + + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + final LiveConnection liveConnection = context.getConnection(descriptor.getConnectionInstanceId()); + if (liveConnection == null) { + initialResults.put(descriptor.getConnectionInstanceId(), ConnectionResult.unsupported(descriptor, UnsupportedReason.CONNECTION_NOT_FOUND)); + continue; + } + + if (liveConnection.knownQueueEmpty()) { + initialResults.put(descriptor.getConnectionInstanceId(), ConnectionResult.noDrain(descriptor)); + continue; + } + + if (descriptor.getDestinationInstanceId() != null) { + nonEmptyRemovedDestinationIds.add(descriptor.getDestinationInstanceId()); + } + } + + final Map<String, ConnectionResult> classifiedResults = new LinkedHashMap<>(initialResults); + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + if (classifiedResults.containsKey(descriptor.getConnectionInstanceId())) { + continue; + } + + classifiedResults.put(descriptor.getConnectionInstanceId(), classifyNonEmpty(descriptor, flowUpdateImpact, context, + nonEmptyRemovedDestinationIds, removedConnectionIds)); + } + + final Set<String> candidateProducerBarrierIds = getCandidateProducerBarrierIds(orderedConnections, classifiedResults); + + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + final ConnectionResult connectionResult = classifiedResults.get(descriptor.getConnectionInstanceId()); + if (connectionResult.classification() != Classification.CANDIDATE) { + continue; + } + + if (hasRetainedFeedbackPath(descriptor.getDestinationInstanceId(), candidateProducerBarrierIds, removedConnectionIds, context)) { + classifiedResults.put(descriptor.getConnectionInstanceId(), ConnectionResult.unsupported(descriptor, UnsupportedReason.RETAINED_FEEDBACK_PATH)); + } + } + + final List<ConnectionResult> connectionResults = new ArrayList<>(orderedConnections.size()); + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + connectionResults.add(classifiedResults.get(descriptor.getConnectionInstanceId())); + } + + return new BatchResult(connectionResults, getCandidateProducerBarrierIds(orderedConnections, classifiedResults)); + } + + private ConnectionResult classifyNonEmpty(final RemovedConnectionDescriptor descriptor, final FlowUpdateImpact flowUpdateImpact, + final Context context, final Set<String> nonEmptyRemovedDestinationIds, + final Set<String> removedConnectionIds) { + if (descriptor.getRemovalReason() == RemovalReason.SOURCE_CHANGED) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_CHANGED_REMOVAL); + } + + if (!isRetainedGroupHierarchy(descriptor.getContainingProcessGroupId(), flowUpdateImpact.getRemovedProcessGroupIds(), context)) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.CONNECTION_IN_REMOVED_GROUP); + } + + if (isRemovedEndpoint(flowUpdateImpact, descriptor.getSourceInstanceId())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_COMPONENT_REMOVED); + } + + if (isRemovedEndpoint(flowUpdateImpact, descriptor.getDestinationInstanceId())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.DESTINATION_COMPONENT_REMOVED); + } + + if (!SUPPORTED_SOURCE_TYPES.contains(descriptor.getSourceType())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.UNSUPPORTED_SOURCE_TYPE); + } + + if (!SUPPORTED_DESTINATION_TYPES.contains(descriptor.getDestinationType())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.UNSUPPORTED_DESTINATION_TYPE); + } + + final LiveConnectable destination = context.getConnectable(descriptor.getDestinationInstanceId()); + if (destination == null) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.DESTINATION_COMPONENT_NOT_FOUND); + } + + final Optional<UnsupportedReason> destinationValidationFailure = validateDestination(destination); + if (destinationValidationFailure.isPresent()) { + return ConnectionResult.unsupported(descriptor, destinationValidationFailure.get()); + } + + final Set<String> producerBarrierIds = resolveProducerBarrierIds(descriptor, flowUpdateImpact, context, nonEmptyRemovedDestinationIds); + if (producerBarrierIds.isEmpty()) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.FUNNEL_SOURCE_WITHOUT_SUPPORTED_PRODUCER); Review Comment: [claude-opus-4.8] `FUNNEL_SOURCE_WITHOUT_SUPPORTED_PRODUCER` is reported whenever no producer barrier can be resolved, not only for funnel sources. A connection whose source is a remote process group port, or a processor whose upstream traversal terminates unexpectedly, is reported to the user as a funnel problem. This matters because the reason name reaches the user: `buildClassificationFailureMessage` puts it into the request's failure reason, which surfaces in the UI. Either rename the constant to something accurate such as `NO_SUPPORTED_PRODUCER_FOUND`, or split the funnel case out from the general one so the funnel-specific reason is only used when the source really is a funnel. ########## nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/CreateConnectorIT.java: ########## @@ -63,7 +63,7 @@ public void testCreateStartAndStopGenerateAndUpdateConnector() throws IOExceptio assertEquals("org.apache.nifi.lookup.SimpleKeyValueLookupService", controllerServices.iterator().next().getType()); testRunner.startConnector(); - testRunner.stopConnector(); + assertDoesNotThrow(() -> testRunner.stopConnector(Duration.ofSeconds(120))); Review Comment: [claude-opus-4.8] If `stopConnector` throws, the test fails either way, so the `assertDoesNotThrow` wrapper adds a lambda and an import without adding any signal. The same applies at line 85. Just calling `testRunner.stopConnector(Duration.ofSeconds(120))` directly reads better and behaves identically. Separately, this change appears unrelated to the removed-connection drain work. It would be easier to review, and easier to revert independently, as its own commit. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/RemovedConnectionDrainClassifier.java: ########## @@ -0,0 +1,523 @@ +/* + * 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.web; + +import org.apache.nifi.components.validation.ValidationStatus; +import org.apache.nifi.connectable.Connectable; +import org.apache.nifi.connectable.ConnectableType; +import org.apache.nifi.connectable.Connection; +import org.apache.nifi.connectable.Port; +import org.apache.nifi.controller.ProcessorNode; +import org.apache.nifi.controller.ScheduledState; +import org.apache.nifi.controller.flow.FlowManager; +import org.apache.nifi.groups.ProcessGroup; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collector; +import java.util.stream.Collectors; + +public final class RemovedConnectionDrainClassifier { + private static final Set<ConnectableType> SUPPORTED_SOURCE_TYPES = Set.of( + ConnectableType.PROCESSOR, + ConnectableType.INPUT_PORT, + ConnectableType.OUTPUT_PORT, + ConnectableType.FUNNEL); + private static final Set<ConnectableType> SUPPORTED_PRODUCER_BARRIER_TYPES = Set.of( + ConnectableType.PROCESSOR, + ConnectableType.INPUT_PORT, + ConnectableType.OUTPUT_PORT); + private static final Set<ConnectableType> SUPPORTED_DESTINATION_TYPES = Set.of( + ConnectableType.PROCESSOR, + ConnectableType.INPUT_PORT, + ConnectableType.OUTPUT_PORT); + private static final Comparator<RemovedConnectionDescriptor> CONNECTION_ORDER = + Comparator.comparing(RemovedConnectionDescriptor::getConnectionInstanceId, Comparator.nullsLast(String::compareTo)) + .thenComparing(RemovedConnectionDescriptor::getConnectionVersionedId, Comparator.nullsLast(String::compareTo)); + + BatchResult classify(final FlowUpdateImpact flowUpdateImpact, final Context context) { + Objects.requireNonNull(flowUpdateImpact, "Flow Update Impact required"); + Objects.requireNonNull(context, "Removed Connection Drain Context required"); + + final List<RemovedConnectionDescriptor> orderedConnections = flowUpdateImpact.getRemovedConnections().stream() + .sorted(CONNECTION_ORDER) + .toList(); + + final Set<String> removedConnectionIds = orderedConnections.stream() + .map(RemovedConnectionDescriptor::getConnectionInstanceId) + .filter(Objects::nonNull) + .collect(toOrderedSet()); + + final Set<String> nonEmptyRemovedDestinationIds = new LinkedHashSet<>(); + final Map<String, ConnectionResult> initialResults = new LinkedHashMap<>(); + + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + final LiveConnection liveConnection = context.getConnection(descriptor.getConnectionInstanceId()); + if (liveConnection == null) { + initialResults.put(descriptor.getConnectionInstanceId(), ConnectionResult.unsupported(descriptor, UnsupportedReason.CONNECTION_NOT_FOUND)); + continue; + } + + if (liveConnection.knownQueueEmpty()) { + initialResults.put(descriptor.getConnectionInstanceId(), ConnectionResult.noDrain(descriptor)); + continue; + } + + if (descriptor.getDestinationInstanceId() != null) { + nonEmptyRemovedDestinationIds.add(descriptor.getDestinationInstanceId()); + } + } + + final Map<String, ConnectionResult> classifiedResults = new LinkedHashMap<>(initialResults); + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + if (classifiedResults.containsKey(descriptor.getConnectionInstanceId())) { + continue; + } + + classifiedResults.put(descriptor.getConnectionInstanceId(), classifyNonEmpty(descriptor, flowUpdateImpact, context, + nonEmptyRemovedDestinationIds, removedConnectionIds)); + } + + final Set<String> candidateProducerBarrierIds = getCandidateProducerBarrierIds(orderedConnections, classifiedResults); + + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + final ConnectionResult connectionResult = classifiedResults.get(descriptor.getConnectionInstanceId()); + if (connectionResult.classification() != Classification.CANDIDATE) { + continue; + } + + if (hasRetainedFeedbackPath(descriptor.getDestinationInstanceId(), candidateProducerBarrierIds, removedConnectionIds, context)) { + classifiedResults.put(descriptor.getConnectionInstanceId(), ConnectionResult.unsupported(descriptor, UnsupportedReason.RETAINED_FEEDBACK_PATH)); + } + } + + final List<ConnectionResult> connectionResults = new ArrayList<>(orderedConnections.size()); + for (final RemovedConnectionDescriptor descriptor : orderedConnections) { + connectionResults.add(classifiedResults.get(descriptor.getConnectionInstanceId())); + } + + return new BatchResult(connectionResults, getCandidateProducerBarrierIds(orderedConnections, classifiedResults)); + } + + private ConnectionResult classifyNonEmpty(final RemovedConnectionDescriptor descriptor, final FlowUpdateImpact flowUpdateImpact, + final Context context, final Set<String> nonEmptyRemovedDestinationIds, + final Set<String> removedConnectionIds) { + if (descriptor.getRemovalReason() == RemovalReason.SOURCE_CHANGED) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_CHANGED_REMOVAL); + } + + if (!isRetainedGroupHierarchy(descriptor.getContainingProcessGroupId(), flowUpdateImpact.getRemovedProcessGroupIds(), context)) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.CONNECTION_IN_REMOVED_GROUP); + } + + if (isRemovedEndpoint(flowUpdateImpact, descriptor.getSourceInstanceId())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_COMPONENT_REMOVED); + } + + if (isRemovedEndpoint(flowUpdateImpact, descriptor.getDestinationInstanceId())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.DESTINATION_COMPONENT_REMOVED); + } + + if (!SUPPORTED_SOURCE_TYPES.contains(descriptor.getSourceType())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.UNSUPPORTED_SOURCE_TYPE); + } + + if (!SUPPORTED_DESTINATION_TYPES.contains(descriptor.getDestinationType())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.UNSUPPORTED_DESTINATION_TYPE); + } + + final LiveConnectable destination = context.getConnectable(descriptor.getDestinationInstanceId()); + if (destination == null) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.DESTINATION_COMPONENT_NOT_FOUND); + } + + final Optional<UnsupportedReason> destinationValidationFailure = validateDestination(destination); + if (destinationValidationFailure.isPresent()) { + return ConnectionResult.unsupported(descriptor, destinationValidationFailure.get()); + } + + final Set<String> producerBarrierIds = resolveProducerBarrierIds(descriptor, flowUpdateImpact, context, nonEmptyRemovedDestinationIds); + if (producerBarrierIds.isEmpty()) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.FUNNEL_SOURCE_WITHOUT_SUPPORTED_PRODUCER); + } + + for (final String producerBarrierId : producerBarrierIds) { + if (Objects.equals(producerBarrierId, descriptor.getDestinationInstanceId())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SELF_LOOP); + } + + if (nonEmptyRemovedDestinationIds.contains(producerBarrierId)) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.PRODUCER_BARRIER_IS_REMOVED_DESTINATION); + } + + final LiveConnectable producerBarrier = context.getConnectable(producerBarrierId); + if (producerBarrier == null) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_COMPONENT_NOT_FOUND); + } + + if (!SUPPORTED_PRODUCER_BARRIER_TYPES.contains(producerBarrier.type())) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.UNSUPPORTED_SOURCE_TYPE); + } + + if (!isRetainedGroupHierarchy(producerBarrier.processGroupId(), flowUpdateImpact.getRemovedProcessGroupIds(), context)) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_COMPONENT_REMOVED); + } + + if (isRemovedEndpoint(flowUpdateImpact, producerBarrierId)) { + return ConnectionResult.unsupported(descriptor, UnsupportedReason.SOURCE_COMPONENT_REMOVED); + } + } + + return ConnectionResult.candidate(descriptor, producerBarrierIds); + } + + private Set<String> resolveProducerBarrierIds(final RemovedConnectionDescriptor descriptor, final FlowUpdateImpact flowUpdateImpact, + final Context context, final Set<String> nonEmptyRemovedDestinationIds) { + if (descriptor.getSourceType() != ConnectableType.FUNNEL) { + if (descriptor.getSourceInstanceId() == null) { + return Collections.emptySet(); + } + + final LiveConnectable source = context.getConnectable(descriptor.getSourceInstanceId()); + if (source == null) { + return Collections.emptySet(); + } + + if (!isRetainedGroupHierarchy(source.processGroupId(), flowUpdateImpact.getRemovedProcessGroupIds(), context)) { + return Collections.emptySet(); + } + + return Set.of(source.id()); + } + + final LiveConnectable funnel = context.getConnectable(descriptor.getSourceInstanceId()); + if (funnel == null) { + return Collections.emptySet(); + } + + final Set<String> producerBarrierIds = new LinkedHashSet<>(); + collectUpstreamProducerBarrierIds(funnel, context, producerBarrierIds, new HashSet<>()); + return producerBarrierIds; + } + + private void collectUpstreamProducerBarrierIds(final LiveConnectable connectable, final Context context, + final Set<String> producerBarrierIds, final Set<String> visitedConnectionIds) { + if (connectable == null) { + return; + } + + if (connectable.type() != ConnectableType.FUNNEL) { + producerBarrierIds.add(connectable.id()); + return; + } + + for (final String incomingConnectionId : connectable.incomingConnectionIds()) { + if (!visitedConnectionIds.add(incomingConnectionId)) { + continue; + } + + final LiveConnection incomingConnection = context.getConnection(incomingConnectionId); + if (incomingConnection == null) { + continue; + } + + collectUpstreamProducerBarrierIds(context.getConnectable(incomingConnection.sourceId()), context, producerBarrierIds, visitedConnectionIds); + } + } + + private Optional<UnsupportedReason> validateDestination(final LiveConnectable destination) { + if (destination.type() == ConnectableType.PROCESSOR) { + if (destination.physicalScheduledState() != ScheduledState.RUNNING) { + return Optional.of(UnsupportedReason.DESTINATION_NOT_RUNNING); + } + + if (destination.validationStatus() != ValidationStatus.VALID) { + return Optional.of(UnsupportedReason.DESTINATION_NOT_VALID); + } + } else if ((destination.type() == ConnectableType.INPUT_PORT || destination.type() == ConnectableType.OUTPUT_PORT) && !destination.running()) { + return Optional.of(UnsupportedReason.DESTINATION_NOT_RUNNING); + } + + return Optional.empty(); + } + + private boolean hasRetainedFeedbackPath(final String destinationId, final Set<String> producerBarrierIds, + final Set<String> removedConnectionIds, final Context context) { + if (destinationId == null || producerBarrierIds.isEmpty()) { + return false; + } + + final Deque<String> pendingConnectables = new ArrayDeque<>(); + final Set<String> visitedConnectables = new HashSet<>(); + pendingConnectables.add(destinationId); + + while (!pendingConnectables.isEmpty()) { + final String connectableId = pendingConnectables.removeFirst(); + if (!visitedConnectables.add(connectableId)) { + continue; + } + + final LiveConnectable connectable = context.getConnectable(connectableId); + if (connectable == null) { + continue; + } + + for (final String outgoingConnectionId : connectable.outgoingConnectionIds()) { + if (removedConnectionIds.contains(outgoingConnectionId)) { + continue; + } + + final LiveConnection outgoingConnection = context.getConnection(outgoingConnectionId); + if (outgoingConnection == null) { + continue; + } + + final String downstreamConnectableId = outgoingConnection.destinationId(); + if (producerBarrierIds.contains(downstreamConnectableId)) { + return true; + } + + pendingConnectables.addLast(downstreamConnectableId); + } + } + + return false; + } + + private boolean isRemovedEndpoint(final FlowUpdateImpact flowUpdateImpact, final String componentInstanceId) { + return componentInstanceId != null && flowUpdateImpact.getRemovedEndpointIds().contains(componentInstanceId); + } + + private boolean isRetainedGroupHierarchy(final String processGroupId, final Set<String> removedProcessGroupIds, final Context context) { + String currentGroupId = processGroupId; + final Set<String> visitedGroupIds = new HashSet<>(); + while (currentGroupId != null) { + if (!visitedGroupIds.add(currentGroupId)) { + return false; + } + + if (removedProcessGroupIds.contains(currentGroupId)) { + return false; + } + + final LiveProcessGroup processGroup = context.getProcessGroup(currentGroupId); + if (processGroup == null) { + return false; + } + + currentGroupId = processGroup.parentProcessGroupId(); + } + + return true; Review Comment: [claude-opus-4.8] **A connection whose group cannot be resolved is treated as retained.** When `processGroupId` is `null` the `while` loop at line 320 never executes and this returns `true`, meaning "this group survives the update", which makes the connection a drain candidate. This is reachable rather than defensive: `StandardNiFiServiceFacade.getComponentGroupRuntimeId` can return `null`. Every other ambiguous case in this class resolves toward `unsupported`, which is the safe direction because it fails the request before anything is stopped. Note that the not-found case a few lines up at line 330 already returns `false`. Here an unresolvable group instead leads to stopping producers for a connection whose containing group may itself be going away, in which case the queue never drains and the request burns the full 30 seconds before timing out. Returning `false` for a null group id, to match the not-found case, would keep this consistent with the rest of the class. ########## nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/concurrent/StandardAsynchronousWebRequest.java: ########## @@ -156,6 +156,12 @@ public synchronized void fail(final String explanation) { applyFailure(explanation); } + @Override + public synchronized void appendFailureDetail(final String detail) { + final String currentFailureReason = Objects.requireNonNull(failureReason, "Request must already have a failure reason"); Review Comment: [claude-opus-4.8] **This throws `NullPointerException` when the request has not already failed.** `Objects.requireNonNull(failureReason, ...)` makes "a failure reason already exists" a hard precondition, but nothing in the `AsynchronousWebRequest.appendFailureDetail` signature communicates that, and the interface method has no Javadoc stating it. The single caller today, `FlowUpdateResource` line 364, happens to satisfy the precondition because cancellation always sets a failure reason first. But the method is on a public framework interface, so the next caller has no way to know the ordering requirement short of reading this implementation, and gets an NPE if they get it wrong. An NPE is a bug, not an API contract. Suggest one of: append the detail to an empty base when no failure reason is set, or throw `IllegalStateException` with a message that explains the ordering requirement. Either way the interface method should say what it requires. ########## nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/ClusteredRegistryClientIT.java: ########## @@ -26,4 +28,10 @@ public NiFiInstanceFactory getInstanceFactory() { return createTwoNodeInstanceFactory(); } + @Override + @Test + @Disabled("Standalone-only AC19 proof: the inherited fixture waits for a single port-output FlowFile, but clustered GenerateFlowFile execution produces per-node queueing") + public void testRemovedConnectionToInputPortStopsPortBeforeRemoval() throws Exception { Review Comment: [claude-opus-4.8] Could you say more about why this fails in a cluster? The disable reason describes the fixture mismatch (the inherited test waits for a single port-output FlowFile while clustered `GenerateFlowFile` queues per node), but that reads like a reason the test needs a cluster-aware fixture rather than a reason the scenario cannot be covered. This matters because the cluster path holds the most complex new code in the PR: request replication, the node-coverage check, and the mutable-request verification I flagged in `RemovedConnectionDrainCoordinator`. Disabling this leaves that path with unit coverage only, at the same time as the PR removes the one clustered end-to-end exercise of it. If the underlying issue is only the FlowFile-count assertion, adapting the fixture to expect per-node counts seems worth doing here. If it is something deeper about how the drain behaves in a cluster, that is worth understanding before merge, since it may be a finding rather than a test-infrastructure problem. -- 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]
