You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2020/06/15 20:48:03 UTC
[GitHub] [nifi] markap14 opened a new pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
markap14 opened a new pull request #4337:
URL: https://github.com/apache/nifi/pull/4337
Thank you for submitting a contribution to Apache NiFi.
Please provide a short description of the PR here:
#### Description of PR
_Enables X functionality; fixes bug NIFI-YYYY._
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [ ] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [ ] Does your PR title start with **NIFI-XXXX** where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [ ] Has your PR been rebased against the latest commit within the target branch (typically `master`)?
- [ ] Is your initial contribution a single, squashed commit? _Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not `squash` or use `--force` when pushing to allow for clean monitoring of changes._
### For code changes:
- [ ] Have you ensured that the full suite of tests is executed via `mvn -Pcontrib-check clean install` at the root `nifi` folder?
- [ ] Have you written or updated unit tests to verify your changes?
- [ ] Have you verified that the full build is successful on JDK 8?
- [ ] Have you verified that the full build is successful on JDK 11?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the `LICENSE` file, including the main `LICENSE` file under `nifi-assembly`?
- [ ] If applicable, have you updated the `NOTICE` file, including the main `NOTICE` file found under `nifi-assembly`?
- [ ] If adding new Properties, have you added `.displayName` in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI for build issues and submit an update to your PR as soon as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tpalfy commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
tpalfy commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442439764
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
+ .collect(Collectors.toMap(entity -> entity.getScheduleSummary().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorScheduleSummariesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ for (final ProcessorScheduleSummaryEntity processorEntity : nodeResponseEntity.getScheduleSummaries()) {
+ final String processorId = processorEntity.getScheduleSummary().getId();
+
+ final ProcessorScheduleSummaryEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorScheduleSummaryEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity();
+ mergedEntity.setScheduleSummaries(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorScheduleSummaryEntity target, final ProcessorScheduleSummaryEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorScheduleSummaryDTO targetSummaryDto = target.getScheduleSummary();
+ final ProcessorScheduleSummaryDTO additionalSummaryDto = additional.getScheduleSummary();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
Review comment:
My issue is that the outcome depends on which order we process the nodes.
So considering `nodeRunStatuses -> mergedRunStatus`, then we have these outcomes:
```source
(running, invalid) -> invalid
(invalid, running) -> running
```
etc.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#issuecomment-646681043
Thanks for the review @tpalfy! I've merged to master.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442399466
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
+ if (RunStatus.Running.name().equals(additionalRunStatus)) {
+ targetSummaryDto.setRunStatus(RunStatus.Running.name());
+ } else if (RunStatus.Validating.name().equals(additionalRunStatus)) {
+ targetSummaryDto.setRunStatus(RunStatus.Validating.name());
+ } else if (RunStatus.Invalid.name().equals(additionalRunStatus)) {
+ targetSummaryDto.setRunStatus(RunStatus.Invalid.name());
+ }
+
+ // If validation errors is null, it's due to eprmissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getValidationErrors() == null) {
Review comment:
Yes - if the Processor is Validating, its validation errors should be null.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442335960
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
Review comment:
It would be more descriptive in terms of the reasoning behind why it would be null. However, the convention is more to merge field-by-field. If for some reason the name were ever to become null for another reason, this would probably be safer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tpalfy commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
tpalfy commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r440965363
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
+ .collect(Collectors.toMap(entity -> entity.getScheduleSummary().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorScheduleSummariesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ for (final ProcessorScheduleSummaryEntity processorEntity : nodeResponseEntity.getScheduleSummaries()) {
+ final String processorId = processorEntity.getScheduleSummary().getId();
+
+ final ProcessorScheduleSummaryEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorScheduleSummaryEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity();
+ mergedEntity.setScheduleSummaries(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorScheduleSummaryEntity target, final ProcessorScheduleSummaryEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorScheduleSummaryDTO targetSummaryDto = target.getScheduleSummary();
+ final ProcessorScheduleSummaryDTO additionalSummaryDto = additional.getScheduleSummary();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
Review comment:
To me it seems the `runStatus` is simply overwritten with the one coming from the last node we are processing (in case of `running`, `validating` and `invalid`).
Is that intentional?
What if the 1st node returns an `invalid` and the 2nd a `validating`?
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
##########
@@ -3332,6 +3336,38 @@ private ProcessorEntity createProcessorEntity(final ProcessorNode processor, fin
.collect(Collectors.toSet());
}
+ @Override
+ public ProcessorsRunStatusDetailsEntity getProcessorsRunStatusDetails(final Set<String> processorIds, final NiFiUser user) {
+ final List<ProcessorRunStatusDetailsEntity> summaryEntities = processorIds.stream()
Review comment:
Have some old `summary...` names.
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
Review comment:
Minor: This map initialization suggests that we want to make sure the values in the map are the same objects that are coming with the clientResponse (maybe because of the `revision` which we do not merge). Is that true?
Otherwise I could imagine something like this:
```java
if (!canHandle(uri, method)) {
throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
}
Collection<ProcessorScheduleSummaryEntity> mergedEntities = successfulResponses.stream()
.map(nodeResponse -> nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class))
.flatMap(nodeResponseEntity -> nodeResponseEntity.getScheduleSummaries().stream())
.collect(Collectors.toMap(processorEntity -> processorEntity.getScheduleSummary().getId(), processorEntity -> processorEntity, this::merge))
.values();
final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity();
mergedEntity.setScheduleSummaries(new ArrayList<>(mergedEntities));
return new NodeResponse(clientResponse, mergedEntity);
```
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ClusterReplicationComponentLifecycle.java
##########
@@ -171,34 +172,37 @@ private boolean waitForProcessorValidation(final NiFiUser user, final URI origin
URI groupUri;
try {
groupUri = new URI(originalUri.getScheme(), originalUri.getUserInfo(), originalUri.getHost(),
- originalUri.getPort(), "/nifi-api/process-groups/" + groupId + "/processors", "includeDescendantGroups=true", originalUri.getFragment());
+ originalUri.getPort(), "/nifi-api/processors/run-status-details/queries", null, originalUri.getFragment());
} catch (URISyntaxException e) {
throw new RuntimeException(e);
}
final Map<String, String> headers = new HashMap<>();
- final MultivaluedMap<String, String> requestEntity = new MultivaluedHashMap<>();
+ final RunStatusDetailsRequestEntity requestEntity = new RunStatusDetailsRequestEntity();
+ final Set<String> processorIds = processors.values().stream()
+ .map(AffectedComponentEntity::getId)
+ .collect(Collectors.toSet());
+ requestEntity.setProcessorIds(processorIds);
boolean continuePolling = true;
while (continuePolling) {
// Determine whether we should replicate only to the cluster coordinator, or if we should replicate directly to the cluster nodes themselves.
final NodeResponse clusterResponse;
if (getReplicationTarget() == ReplicationTarget.CLUSTER_NODES) {
- clusterResponse = getRequestReplicator().replicate(user, HttpMethod.GET, groupUri, requestEntity, headers).awaitMergedResponse();
+ clusterResponse = getRequestReplicator().replicate(user, HttpMethod.POST, groupUri, requestEntity, headers).awaitMergedResponse();
} else {
clusterResponse = getRequestReplicator().forwardToCoordinator(
- getClusterCoordinatorNode(), user, HttpMethod.GET, groupUri, requestEntity, headers).awaitMergedResponse();
+ getClusterCoordinatorNode(), user, HttpMethod.POST, groupUri, requestEntity, headers).awaitMergedResponse();
}
if (clusterResponse.getStatus() != Status.OK.getStatusCode()) {
return false;
}
- final ProcessorsEntity processorsEntity = getResponseEntity(clusterResponse, ProcessorsEntity.class);
- final Set<ProcessorEntity> processorEntities = processorsEntity.getProcessors();
+ final ProcessorsRunStatusDetailsEntity summariesEntity = getResponseEntity(clusterResponse, ProcessorsRunStatusDetailsEntity.class);
Review comment:
There are some `summariesEntity` old names.
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
##########
@@ -3332,6 +3336,38 @@ private ProcessorEntity createProcessorEntity(final ProcessorNode processor, fin
.collect(Collectors.toSet());
}
+ @Override
+ public ProcessorsRunStatusDetailsEntity getProcessorsRunStatusDetails(final Set<String> processorIds, final NiFiUser user) {
+ final List<ProcessorRunStatusDetailsEntity> summaryEntities = processorIds.stream()
+ .map(processorDAO::getProcessor)
+ .map(processor -> createRunStatusDetailsEntity(processor, user))
+ .collect(Collectors.toList());
+
+ final ProcessorsRunStatusDetailsEntity summariesEntity = new ProcessorsRunStatusDetailsEntity();
+ summariesEntity.setRunStatusDetails(summaryEntities);
+ return summariesEntity;
+ }
+
+ private ProcessorRunStatusDetailsEntity createRunStatusDetailsEntity(final ProcessorNode processor, final NiFiUser user) {
+ final RevisionDTO revision = dtoFactory.createRevisionDTO(revisionManager.getRevision(processor.getIdentifier()));
+ final PermissionsDTO permissions = dtoFactory.createPermissionsDto(processor, user);
+ final ProcessorStatus processorStatus = controllerFacade.getProcessorStatus(processor.getIdentifier());
+ final ProcessorRunStatusDetailsDTO runStatusDetailsDto = dtoFactory.createProcessorRunStatusDetailsDto(processor, processorStatus);
+
+ if (!Boolean.TRUE.equals(permissions.getCanRead())) {
Review comment:
Minor: Not sure why `getCanRead()` has the return type `Boolean` when it's actually a getter for a `boolean` property.
Simply doing
```suggestion
if (permissions.getCanRead()) {
```
should be safe.
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
Review comment:
Wouldn't `!additional.getPermissions().getCanRead()` be a bit more descriptive?
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
+ if (RunStatus.Running.name().equals(additionalRunStatus)) {
+ targetSummaryDto.setRunStatus(RunStatus.Running.name());
+ } else if (RunStatus.Validating.name().equals(additionalRunStatus)) {
+ targetSummaryDto.setRunStatus(RunStatus.Validating.name());
+ } else if (RunStatus.Invalid.name().equals(additionalRunStatus)) {
+ targetSummaryDto.setRunStatus(RunStatus.Invalid.name());
+ }
+
+ // If validation errors is null, it's due to eprmissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getValidationErrors() == null) {
Review comment:
Typo: eprmissions
Again, could use `!additional.getPermissions().getCanRead()` instead.
We may be relying on a "secondary" truth when we have access to the "primary".
Also to me it looks like the check could be incorrect in this way. Wouldn't the `validationErrors` be set to `null` if the processor is in `VALIDATING` according to this logic:
```java
dto.setValidationErrors(convertValidationErrors(processor.getValidationErrors()));
...
protected Collection<ValidationResult> getValidationErrors(final Set<ControllerServiceNode> servicesToIgnore) {
final ValidationState validationState = this.validationState.get();
if (validationState.getStatus() == ValidationStatus.VALIDATING) {
return null;
}
...
private Set<String> convertValidationErrors(final Collection<ValidationResult> validationErrors) {
if (validationErrors == null) {
return null;
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r440956453
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ClusterReplicationComponentLifecycle.java
##########
@@ -171,34 +172,37 @@ private boolean waitForProcessorValidation(final NiFiUser user, final URI origin
URI groupUri;
try {
groupUri = new URI(originalUri.getScheme(), originalUri.getUserInfo(), originalUri.getHost(),
- originalUri.getPort(), "/nifi-api/process-groups/" + groupId + "/processors", "includeDescendantGroups=true", originalUri.getFragment());
+ originalUri.getPort(), "/nifi-api/processors/schedule-summaries/queries", null, originalUri.getFragment());
Review comment:
Good catch! Will fix.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tpalfy commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
tpalfy commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442813639
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> runStatusDetailMap = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = runStatusDetailMap.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(runStatusDetailMap.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If any node indicates that we do not have read permissions, null out both the name and validation errors.
+ if (!additional.getPermissions().getCanRead()) {
+ targetSummaryDto.setName(null);
+ targetSummaryDto.setValidationErrors(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ // if the status to merge is validating/invalid allow it to take precedence. whether the
+ // processor run status is disabled/stopped/running is part of the flow configuration
+ // and should not differ amongst nodes. however, whether a processor is validating/invalid
+ // can be driven by environmental conditions. this check allows any of those to
+ // take precedence over the configured run status.
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
+ if (RunStatus.Invalid.name().equals(additionalRunStatus)) {
Review comment:
```suggestion
if (RunStatus.Invalid.name().equals(additionalRunStatus) || RunStatus.Validating.name().equals(additionalRunStatus)) {
targetSummaryDto.setRunStatus(additionalRunStatus);
}
```
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> runStatusDetailMap = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = runStatusDetailMap.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(runStatusDetailMap.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If any node indicates that we do not have read permissions, null out both the name and validation errors.
+ if (!additional.getPermissions().getCanRead()) {
+ targetSummaryDto.setName(null);
Review comment:
It's still called `summary` instead of `details` :)
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
Review comment:
`summary`
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
Review comment:
`summary` (in comment)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tpalfy commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
tpalfy commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r440866568
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ClusterReplicationComponentLifecycle.java
##########
@@ -171,34 +172,37 @@ private boolean waitForProcessorValidation(final NiFiUser user, final URI origin
URI groupUri;
try {
groupUri = new URI(originalUri.getScheme(), originalUri.getUserInfo(), originalUri.getHost(),
- originalUri.getPort(), "/nifi-api/process-groups/" + groupId + "/processors", "includeDescendantGroups=true", originalUri.getFragment());
+ originalUri.getPort(), "/nifi-api/processors/schedule-summaries/queries", null, originalUri.getFragment());
Review comment:
Shouldn't the endpoint String here be the same as in
```java
public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
```
`.../query` vs `.../queries`
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
Review comment:
Maybe we could use a name like `ProcessorRunStatusDetails...` for this use-case as scheduling related information is not really a part of 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 closed pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 closed pull request #4337:
URL: https://github.com/apache/nifi/pull/4337
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442398673
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
Review comment:
Yeah I think it's probably reasonable to use that as the conditional.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r440956698
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
Review comment:
I agree. Will 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442395701
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
+ .collect(Collectors.toMap(entity -> entity.getScheduleSummary().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorScheduleSummariesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ for (final ProcessorScheduleSummaryEntity processorEntity : nodeResponseEntity.getScheduleSummaries()) {
+ final String processorId = processorEntity.getScheduleSummary().getId();
+
+ final ProcessorScheduleSummaryEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorScheduleSummaryEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity();
+ mergedEntity.setScheduleSummaries(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorScheduleSummaryEntity target, final ProcessorScheduleSummaryEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorScheduleSummaryDTO targetSummaryDto = target.getScheduleSummary();
+ final ProcessorScheduleSummaryDTO additionalSummaryDto = additional.getScheduleSummary();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
Review comment:
Yes. If any node indicates that the Processor is running, it is to be considered Running. Otherwise, if any node considers it invalid, it should be invalid. Otherwise, if any node indicates it is Validating, it should be Validating because its validation is not yet known. Invalid vs. Validating I would say is debatable. If one node says Validating and the other Invalid, we could say it is Invalid or Validating because the validation is not actually known. On the other hand, regardless of whether Validating becomes Valid or Invalid, the overall status will be Invalid if any is Invalid. So the precedence there is kind of irrelevant.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442334635
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
##########
@@ -3332,6 +3336,38 @@ private ProcessorEntity createProcessorEntity(final ProcessorNode processor, fin
.collect(Collectors.toSet());
}
+ @Override
+ public ProcessorsRunStatusDetailsEntity getProcessorsRunStatusDetails(final Set<String> processorIds, final NiFiUser user) {
+ final List<ProcessorRunStatusDetailsEntity> summaryEntities = processorIds.stream()
+ .map(processorDAO::getProcessor)
+ .map(processor -> createRunStatusDetailsEntity(processor, user))
+ .collect(Collectors.toList());
+
+ final ProcessorsRunStatusDetailsEntity summariesEntity = new ProcessorsRunStatusDetailsEntity();
+ summariesEntity.setRunStatusDetails(summaryEntities);
+ return summariesEntity;
+ }
+
+ private ProcessorRunStatusDetailsEntity createRunStatusDetailsEntity(final ProcessorNode processor, final NiFiUser user) {
+ final RevisionDTO revision = dtoFactory.createRevisionDTO(revisionManager.getRevision(processor.getIdentifier()));
+ final PermissionsDTO permissions = dtoFactory.createPermissionsDto(processor, user);
+ final ProcessorStatus processorStatus = controllerFacade.getProcessorStatus(processor.getIdentifier());
+ final ProcessorRunStatusDetailsDTO runStatusDetailsDto = dtoFactory.createProcessorRunStatusDetailsDto(processor, processorStatus);
+
+ if (!Boolean.TRUE.equals(permissions.getCanRead())) {
Review comment:
@tpalfy DTO objects in nifi always use Objects rather than primitives because a primitive would default to `false` and we don't want that - we want the exact value that was set, including `null` potentially. Because it can be `null`, we cannot use auto-boxing, as it could throw NPE
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#issuecomment-646655835
Good catches @tpalfy. I had tried searching the code base for things like "summary" but got so many hits that were unrelated that it didn't help. It just occurred to me that I could run a "git diff" to see what was changing and search that for summary. The latest commit has no mention of summary :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442335960
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO;
+import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity;
+import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream()
+ .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class);
+
+ for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) {
+ final String processorId = processorEntity.getRunStatusDetails().getId();
+
+ final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity();
+ mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails();
+ final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
Review comment:
It would be more descriptive in terms of the reasoning behind why it would be null. However, the convention is more to merge field-by-field. If for some reason the name were ever to become null for another reason, this would probably be safer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442458341
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
+ .collect(Collectors.toMap(entity -> entity.getScheduleSummary().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorScheduleSummariesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ for (final ProcessorScheduleSummaryEntity processorEntity : nodeResponseEntity.getScheduleSummaries()) {
+ final String processorId = processorEntity.getScheduleSummary().getId();
+
+ final ProcessorScheduleSummaryEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorScheduleSummaryEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity();
+ mergedEntity.setScheduleSummaries(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorScheduleSummaryEntity target, final ProcessorScheduleSummaryEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorScheduleSummaryDTO targetSummaryDto = target.getScheduleSummary();
+ final ProcessorScheduleSummaryDTO additionalSummaryDto = additional.getScheduleSummary();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
Review comment:
Looking at how we merge the RunStatus in StatusMerger, I see the confusion/issue. We should only be checking for states of Invalid / Validating. If the additional run status is Validating or Invalid, then set the target's RunStatus to that. Otherwise, leave it as is. That way, if any node has a status of Validating/Invalid, then that will take precedence. Which one of those wins doesn't really matter, as mentioned above. This is the logic used elsewhere, so will update this PR to use that logic as well.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442392977
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
##########
@@ -3332,6 +3336,38 @@ private ProcessorEntity createProcessorEntity(final ProcessorNode processor, fin
.collect(Collectors.toSet());
}
+ @Override
+ public ProcessorsRunStatusDetailsEntity getProcessorsRunStatusDetails(final Set<String> processorIds, final NiFiUser user) {
+ final List<ProcessorRunStatusDetailsEntity> summaryEntities = processorIds.stream()
+ .map(processorDAO::getProcessor)
+ .map(processor -> createRunStatusDetailsEntity(processor, user))
+ .collect(Collectors.toList());
+
+ final ProcessorsRunStatusDetailsEntity summariesEntity = new ProcessorsRunStatusDetailsEntity();
+ summariesEntity.setRunStatusDetails(summaryEntities);
+ return summariesEntity;
+ }
+
+ private ProcessorRunStatusDetailsEntity createRunStatusDetailsEntity(final ProcessorNode processor, final NiFiUser user) {
+ final RevisionDTO revision = dtoFactory.createRevisionDTO(revisionManager.getRevision(processor.getIdentifier()));
+ final PermissionsDTO permissions = dtoFactory.createPermissionsDto(processor, user);
+ final ProcessorStatus processorStatus = controllerFacade.getProcessorStatus(processor.getIdentifier());
+ final ProcessorRunStatusDetailsDTO runStatusDetailsDto = dtoFactory.createProcessorRunStatusDetailsDto(processor, processorStatus);
+
+ if (!Boolean.TRUE.equals(permissions.getCanRead())) {
Review comment:
Actually, in this case I guess it doesn't matter - we know that the permissions always will be populated. So even though it's a Boolean, it's safe to assume that the value will be set.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442450270
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
+ .collect(Collectors.toMap(entity -> entity.getScheduleSummary().getId(), entity -> entity));
+
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ final ProcessorScheduleSummariesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity :
+ nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ for (final ProcessorScheduleSummaryEntity processorEntity : nodeResponseEntity.getScheduleSummaries()) {
+ final String processorId = processorEntity.getScheduleSummary().getId();
+
+ final ProcessorScheduleSummaryEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorScheduleSummaryEntity());
+ merge(mergedEntity, processorEntity);
+ }
+ }
+
+ final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity();
+ mergedEntity.setScheduleSummaries(new ArrayList<>(scheduleSummaries.values()));
+ return new NodeResponse(clientResponse, mergedEntity);
+ }
+
+ private void merge(final ProcessorScheduleSummaryEntity target, final ProcessorScheduleSummaryEntity additional) {
+ PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions());
+
+ final ProcessorScheduleSummaryDTO targetSummaryDto = target.getScheduleSummary();
+ final ProcessorScheduleSummaryDTO additionalSummaryDto = additional.getScheduleSummary();
+
+ // If name is null, it's because of permissions, so we want to nullify it in the target.
+ if (additionalSummaryDto.getName() == null) {
+ targetSummaryDto.setName(null);
+ }
+
+ targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount());
+
+ final String additionalRunStatus = additionalSummaryDto.getRunStatus();
Review comment:
Wow, yes. Thanks. Now I understand. I think we have this same bug elsewhere then :) Will address.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] markap14 commented on a change in pull request #4337: NIFI-7536: Fix to improve performance of updating parameters
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4337:
URL: https://github.com/apache/nifi/pull/4337#discussion_r442398328
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.PermissionsDtoMerger;
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity;
+import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger {
+ public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query";
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath());
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method);
+ }
+
+ final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class);
+
+ // Create mapping of Processor ID to its schedule Summary.
+ final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream()
Review comment:
This suggestion is definitely more succinct. But I would argue less clear. I would also be hesitant to change it because this is a pretty common pattern repeated in most of the Endpoint Mergers, so I'd rather stick with the pattern that is common and heavily tested/utilized.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org