You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2021/04/14 20:48:33 UTC
[nifi] branch main updated: NIFI-8430: Close Atlas client in order
to free up resources
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 2fb05cd NIFI-8430: Close Atlas client in order to free up resources
2fb05cd is described below
commit 2fb05cd8dc6b71d1808cc7447ce085a8e7ab8236
Author: Peter Turcsanyi <tu...@apache.org>
AuthorDate: Wed Apr 14 21:23:14 2021 +0200
NIFI-8430: Close Atlas client in order to free up resources
This closes #5002
Signed-off-by: David Handermann <ex...@apache.org>
---
.../org/apache/nifi/atlas/NiFiAtlasClient.java | 7 ++-
.../nifi/atlas/reporting/ReportLineageToAtlas.java | 64 +++++++++++-----------
2 files changed, 38 insertions(+), 33 deletions(-)
diff --git a/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/NiFiAtlasClient.java b/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/NiFiAtlasClient.java
index 3436ff3..e5c289a 100644
--- a/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/NiFiAtlasClient.java
+++ b/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/NiFiAtlasClient.java
@@ -74,7 +74,7 @@ import static org.apache.nifi.atlas.NiFiTypes.TYPE_NIFI_INPUT_PORT;
import static org.apache.nifi.atlas.NiFiTypes.TYPE_NIFI_OUTPUT_PORT;
import static org.apache.nifi.atlas.NiFiTypes.TYPE_NIFI_QUEUE;
-public class NiFiAtlasClient {
+public class NiFiAtlasClient implements AutoCloseable {
private static final Logger logger = LoggerFactory.getLogger(NiFiAtlasClient.class);
@@ -84,6 +84,11 @@ public class NiFiAtlasClient {
this.atlasClient = atlasClient;
}
+ @Override
+ public void close() {
+ atlasClient.close();
+ }
+
/**
* This is an utility method to delete unused types.
* Should be used during development or testing only.
diff --git a/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/reporting/ReportLineageToAtlas.java b/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/reporting/ReportLineageToAtlas.java
index 9afacf5..ca7ca36 100644
--- a/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/reporting/ReportLineageToAtlas.java
+++ b/nifi-nar-bundles/nifi-atlas-bundle/nifi-atlas-reporting-task/src/main/java/org/apache/nifi/atlas/reporting/ReportLineageToAtlas.java
@@ -814,46 +814,46 @@ public class ReportLineageToAtlas extends AbstractReportingTask {
// If standalone or being primary node in a NiFi cluster, this node is responsible for doing primary tasks.
final boolean isResponsibleForPrimaryTasks = !isClustered || getNodeTypeProvider().isPrimary();
- final NiFiAtlasClient atlasClient = createNiFiAtlasClient(context);
-
- // Create Entity defs in Atlas if there's none yet.
- if (!isTypeDefCreated) {
- try {
- if (isResponsibleForPrimaryTasks) {
- // Create NiFi type definitions in Atlas type system.
- atlasClient.registerNiFiTypeDefs(false);
- } else {
- // Otherwise, just check existence of NiFi type definitions.
- if (!atlasClient.isNiFiTypeDefsRegistered()) {
- getLogger().debug("NiFi type definitions are not ready in Atlas type system yet.");
- return;
+ try (final NiFiAtlasClient atlasClient = createNiFiAtlasClient(context)) {
+
+ // Create Entity defs in Atlas if there's none yet.
+ if (!isTypeDefCreated) {
+ try {
+ if (isResponsibleForPrimaryTasks) {
+ // Create NiFi type definitions in Atlas type system.
+ atlasClient.registerNiFiTypeDefs(false);
+ } else {
+ // Otherwise, just check existence of NiFi type definitions.
+ if (!atlasClient.isNiFiTypeDefsRegistered()) {
+ getLogger().debug("NiFi type definitions are not ready in Atlas type system yet.");
+ return;
+ }
}
+ isTypeDefCreated = true;
+ } catch (AtlasServiceException e) {
+ throw new RuntimeException("Failed to check and create NiFi flow type definitions in Atlas due to " + e, e);
}
- isTypeDefCreated = true;
- } catch (AtlasServiceException e) {
- throw new RuntimeException("Failed to check and create NiFi flow type definitions in Atlas due to " + e, e);
}
- }
- // Regardless of whether being a primary task node, each node has to analyse NiFiFlow.
- // Assuming each node has the same flow definition, that is guaranteed by NiFi cluster management mechanism.
- final NiFiFlow nifiFlow = createNiFiFlow(context, atlasClient);
+ // Regardless of whether being a primary task node, each node has to analyse NiFiFlow.
+ // Assuming each node has the same flow definition, that is guaranteed by NiFi cluster management mechanism.
+ final NiFiFlow nifiFlow = createNiFiFlow(context, atlasClient);
- if (isResponsibleForPrimaryTasks) {
- try {
- atlasClient.registerNiFiFlow(nifiFlow);
- } catch (AtlasServiceException e) {
- throw new RuntimeException("Failed to register NiFI flow. " + e, e);
+ if (isResponsibleForPrimaryTasks) {
+ try {
+ atlasClient.registerNiFiFlow(nifiFlow);
+ } catch (AtlasServiceException e) {
+ throw new RuntimeException("Failed to register NiFI flow. " + e, e);
+ }
}
- }
-
- // NOTE: There is a race condition between the primary node and other nodes.
- // If a node notifies an event related to a NiFi component which is not yet created by NiFi primary node,
- // then the notification message will fail due to having a reference to a non-existing entity.
- nifiAtlasHook.setAtlasClient(atlasClient);
- consumeNiFiProvenanceEvents(context, nifiFlow);
+ // NOTE: There is a race condition between the primary node and other nodes.
+ // If a node notifies an event related to a NiFi component which is not yet created by NiFi primary node,
+ // then the notification message will fail due to having a reference to a non-existing entity.
+ nifiAtlasHook.setAtlasClient(atlasClient);
+ consumeNiFiProvenanceEvents(context, nifiFlow);
+ }
}
private NiFiFlow createNiFiFlow(ReportingContext context, NiFiAtlasClient atlasClient) {