exceptionfactory commented on code in PR #7092:
URL: https://github.com/apache/nifi/pull/7092#discussion_r1155979058
##########
nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java:
##########
@@ -295,6 +295,48 @@ public Response createFlowVersion(
return
Response.status(Response.Status.OK).entity(createdSnapshot).build();
}
+ @POST
+ @Path("{flowId}/versions/migrate")
Review Comment:
Having a verb `migrate` in the path does not follow general REST API
conventions, where the path should be a noun and the HTTP method is the verb.
One option could be to make `migrate` and `Boolean` query parameter on the
`/versions` resource method. However, the word `migrate` itself does not
clearly convey what is happening, so perhaps another parameter name would be
better. As the method name mentions keeping the author, it seems like a better
approach would be to adding a query parameter named something like
`preserveAuthor`. If there are other properties to be preserved, perhaps it
could be named `preserveSourceProperties`.
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/CommandOption.java:
##########
@@ -164,7 +164,9 @@ public enum CommandOption {
OUTPUT_TYPE("ot", "outputType", "The type of output to produce (json or
simple)", true),
VERBOSE("verbose", "verbose", "Indicates that verbose output should be
provided", false),
RECURSIVE("r", "recursive", "Indicates the command should perform the
action recursively", false),
- HELP("h", "help", "Help", false)
+ HELP("h", "help", "Help", false),
+ SKIP("skip", "skip", "Indicates to skip an operation if target object
exists", true),
Review Comment:
The word `skip` does not clearly convey the intent of this parameter.
Perhaps `skip-existing` could work, or some other combination of words that
makes it clear what should be skipped.
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/CommandOption.java:
##########
@@ -164,7 +164,9 @@ public enum CommandOption {
OUTPUT_TYPE("ot", "outputType", "The type of output to produce (json or
simple)", true),
VERBOSE("verbose", "verbose", "Indicates that verbose output should be
provided", false),
RECURSIVE("r", "recursive", "Indicates the command should perform the
action recursively", false),
- HELP("h", "help", "Help", false)
+ HELP("h", "help", "Help", false),
+ SKIP("skip", "skip", "Indicates to skip an operation if target object
exists", true),
+ KEEP("keep", "keep", "Indicates to keep the original format or type of an
object", true)
Review Comment:
Similar to `skip`, the word `keep` by itself does not indicate the meaning.
Perhaps `preserve-format` or `preserve-type` could be better.
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ExportAllFlows.java:
##########
@@ -0,0 +1,182 @@
+/*
+ * 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.toolkit.cli.impl.command.registry.flow;
+
+import org.apache.commons.cli.ParseException;
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.client.NiFiRegistryClient;
+import org.apache.nifi.registry.client.NiFiRegistryException;
+import org.apache.nifi.registry.flow.VersionedFlow;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
+import org.apache.nifi.toolkit.cli.api.CommandException;
+import org.apache.nifi.toolkit.cli.api.Context;
+import org.apache.nifi.toolkit.cli.impl.command.CommandOption;
+import
org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand;
+import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets;
+import
org.apache.nifi.toolkit.cli.impl.result.registry.VersionedFlowSnapshotsResult;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class ExportAllFlows extends
AbstractNiFiRegistryCommand<VersionedFlowSnapshotsResult> {
+ private static final String ALL_BUCKETS_COLLECTED = "All buckets
collected...";
+ private static final String ALL_FLOWS_COLLECTED = "All flows collected...";
+ private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow
versions collected...";
+ private static final String EXPORTING_FLOW_VERSIONS = "Exporting flow
versions...";
+ private final ListBuckets listBuckets;
+ private final ListFlows listFlows;
+ private final ListFlowVersions listFlowVersions;
+ private final ExportFlowVersion exportFlowVersion;
+
+ public ExportAllFlows() {
+ super("export-all-flows", VersionedFlowSnapshotsResult.class);
+ this.listBuckets = new ListBuckets();
+ this.listFlows = new ListFlows();
+ this.listFlowVersions = new ListFlowVersions();
+ this.exportFlowVersion = new ExportFlowVersion();
+ }
+
+ @Override
+ public void doInitialize(final Context context) {
+ addOption(CommandOption.OUTPUT_DIR.createOption());
+
+ listBuckets.initialize(context);
+ listFlows.initialize(context);
+ listFlowVersions.initialize(context);
+ exportFlowVersion.initialize(context);
+ }
+
+ @Override
+ public String getDescription() {
+ return "List all the buckets, for each bucket, list all the flows, for
each flow, list all versions and export each version." +
+ "Versions will be saved in the provided target directory.";
+ }
+
+ @Override
+ public VersionedFlowSnapshotsResult doExecute(NiFiRegistryClient client,
Properties properties) throws IOException, NiFiRegistryException,
ParseException, CommandException {
+ final String outputDirectory = getRequiredArg(properties,
CommandOption.OUTPUT_DIR);
+ final boolean isInteractive = getContext().isInteractive();
+
+ // Gather all buckets and create a map for quick access by bucket id
+ final Map<String, Bucket> bucketMap = getBucketMap(client,
isInteractive);
+
+ // Gather all flows and create a map for quick access by flow id
+ final Map<String, VersionedFlow> flowMap = getFlowMap(client,
bucketMap, isInteractive);
+
+ // Gather all versions for all the flows
+ final List<VersionedFlowSnapshotMetadata>
versionedFlowSnapshotMetadataList =
getVersionedFlowSnapshotMetadataList(client, flowMap, isInteractive);
+
+ // Prepare flow version exports
+ final List<VersionedFlowSnapshot> versionedFlowSnapshotList =
getVersionedFlowSnapshotResults(client, outputDirectory, bucketMap, flowMap,
versionedFlowSnapshotMetadataList, isInteractive);
Review Comment:
Are there potential memory concerns with this approach? What happens if
there are hundreds of large flows?
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.toolkit.cli.impl.command.registry.flow;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.cli.MissingOptionException;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.curator.shaded.com.google.common.collect.ComparisonChain;
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.client.BucketClient;
+import org.apache.nifi.registry.client.FlowClient;
+import org.apache.nifi.registry.client.NiFiRegistryClient;
+import org.apache.nifi.registry.client.NiFiRegistryException;
+import org.apache.nifi.registry.flow.VersionedFlow;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
+import org.apache.nifi.toolkit.cli.api.CommandException;
+import org.apache.nifi.toolkit.cli.api.Context;
+import org.apache.nifi.toolkit.cli.impl.command.CommandOption;
+import
org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand;
+import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets;
+import org.apache.nifi.toolkit.cli.impl.result.StringResult;
+import org.apache.nifi.toolkit.cli.impl.util.JacksonUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ImportAllFlows extends AbstractNiFiRegistryCommand<StringResult> {
+ private static final String FILE_NAME_PREFIX =
"toolkit_registry_export_all_";
+ private static final String SKIPPING_BUCKET_CREATION = " already exists,
skipping bucket creation...";
+ private static final String SKIPPING_IMPORT = " already exists, skipping
import...";
+ private static final String SKIPPING_FLOW_CREATION = " already exists,
skipping flow creation...";
+ private static final String IMPORT_COMPLETED = "Import completed...";
+ private static final String ALL_BUCKETS_COLLECTED = "All buckets
collected...";
+ private static final String ALL_FLOWS_COLLECTED = "All flows collected...";
+ private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow
versions collected...";
+ private final ListBuckets listBuckets;
+ private final ListFlows listFlows;
+ private final ListFlowVersions listFlowVersions;
+ private final ImportFlowVersion importFlowVersion;
+
+ public ImportAllFlows() {
+ super("import-all-flows", StringResult.class);
+ this.listBuckets = new ListBuckets();
+ this.listFlows = new ListFlows();
+ this.listFlowVersions = new ListFlowVersions();
+ this.importFlowVersion = new ImportFlowVersion();
+ }
+
+ @Override
+ protected void doInitialize(Context context) {
+ addOption(CommandOption.INPUT_SOURCE.createOption());
+ addOption(CommandOption.SKIP.createOption());
+
+ listBuckets.initialize(context);
+ listFlows.initialize(context);
+ listFlowVersions.initialize(context);
+ importFlowVersion.initialize(context);
+ }
+
+ @Override
+ public String getDescription() {
+ return "From a provided directory as input, the directory content must
be generated by the export-all-flows command, " +
+ "based on the file contents, the corresponding buckets, flows
and flow versions will be created." +
+ "If not configured otherwise, already existing objects will be
skipped.";
+ }
+
+ @Override
+ public StringResult doExecute(final NiFiRegistryClient client, final
Properties properties) throws IOException, NiFiRegistryException,
ParseException, CommandException {
+ final boolean skip = Boolean.parseBoolean(getRequiredArg(properties,
CommandOption.SKIP));
+ final boolean isInteractive = getContext().isInteractive();
+
+ //Gather all buckets and create a map for easier search by bucket name
+ final Map<String, String> bucketMap = getBucketMap(client,
isInteractive);
+
+ // Gather all flows and create a map for easier search by flow name.
+ // As flow name is only unique within the same bucket we need to use
the bucket id in the key as well
+ final Map<Pair<String, String>, String> flowMap = getFlowMap(client,
bucketMap, isInteractive);
+ final Map<Pair<String, String>, String> flowCreated = new HashMap<>();
+
+ // Gather all flow versions and create a map for easier search by flow
id
+ final Map<String, List<Integer>> versionMap = getVersionMap(client,
flowMap, isInteractive);
+
+ // Create file path list
+ final List<String> files = getFilePathList(properties);
+
+ // Deserialize file content
+ final List<ImportedSnapshot> importedSnapshots =
deserializeSnapshots(files);
+
+ // As we need to keep the version order the snapshot list needs to be
sorted
+ importedSnapshots.sort((o1, o2) -> ComparisonChain.start()
+ .compare(o1.getSnapshot().getBucket().getName(),
o2.getSnapshot().getBucket().getName())
+ .compare(o1.getSnapshot().getFlow().getName(),
o2.getSnapshot().getFlow().getName())
+ .compare(o1.getSnapshot().getSnapshotMetadata().getVersion(),
o2.getSnapshot().getSnapshotMetadata().getVersion())
+ .result());
+
+ for (final ImportedSnapshot snapshot : importedSnapshots) {
+
+ final String inputSource = snapshot.getInputSource();
+ final String bucketName =
snapshot.getSnapshot().getBucket().getName();
+ final String bucketDescription =
snapshot.getSnapshot().getBucket().getDescription();
+ final String flowName = snapshot.getSnapshot().getFlow().getName();
+ final String flowDescription =
snapshot.getSnapshot().getFlow().getDescription();
+ final int flowVersion =
snapshot.getSnapshot().getSnapshotMetadata().getVersion();
+ // The original bucket and flow ids must be kept otherwise NiFi
won't be able to synchronize with the NiFi Registry
+ final String bucketId =
snapshot.getSnapshot().getBucket().getIdentifier();
+ final String flowId =
snapshot.getSnapshot().getFlow().getIdentifier();
+
+ // Create bucket if missing
+ if (bucketMap.containsKey(bucketName)) {
+ printMessage(isInteractive, bucketName +
SKIPPING_BUCKET_CREATION);
+ } else {
+ createBucket(client, bucketMap, bucketName, bucketDescription,
bucketId);
+ }
+
+ // Create flow if missing
+ if (flowMap.containsKey(new ImmutablePair<>(bucketId, flowName))) {
+ if (skip) {
+ printMessage(isInteractive, flowName + SKIPPING_IMPORT);
+ continue;
+ } else {
+ printMessage(isInteractive, flowName +
SKIPPING_FLOW_CREATION);
+ }
+ } else if (!flowCreated.containsKey(new ImmutablePair<>(bucketId,
flowName))) {
+ createFlow(client, flowCreated, flowId, flowName,
flowDescription, bucketId);
+ }
+
+ // Create missing flow versions
+ if (!versionMap.getOrDefault(flowId,
Collections.emptyList()).contains(flowVersion)) {
+ createFlowVersion(client, inputSource, flowId);
+ }
+ }
+ return new StringResult(IMPORT_COMPLETED,
getContext().isInteractive());
+ }
+
+ private Map<String, String> getBucketMap(final NiFiRegistryClient client,
final boolean isInteractive) throws IOException, NiFiRegistryException {
+ printMessage(isInteractive, ALL_BUCKETS_COLLECTED);
+
+ return listBuckets.doExecute(client, new Properties())
+ .getResult()
+ .stream()
+ .collect(Collectors.toMap(Bucket::getName,
Bucket::getIdentifier));
+ }
+
+ private Map<Pair<String, String>, String> getFlowMap(final
NiFiRegistryClient client, final Map<String, String> bucketMap,
+ final boolean
isInteractive) throws ParseException, IOException, NiFiRegistryException {
+ printMessage(isInteractive, ALL_FLOWS_COLLECTED);
+ return getVersionedFlows(client, bucketMap)
+ .stream()
+ .collect(Collectors.toMap(e -> new
ImmutablePair<>(e.getBucketIdentifier(), e.getName()),
+ VersionedFlow::getIdentifier));
+ }
+
+ private List<VersionedFlow> getVersionedFlows(final NiFiRegistryClient
client, final Map<String, String> bucketMap) throws ParseException,
IOException, NiFiRegistryException {
+ final List<VersionedFlow> flows = new ArrayList<>();
+ for (final String id : bucketMap.values()) {
+ final Properties flowProperties = new Properties();
+ flowProperties.setProperty(CommandOption.BUCKET_ID.getLongName(),
id);
+
+ flows.addAll(listFlows.doExecute(client,
flowProperties).getResult());
+ }
+ return flows;
+ }
+
+ private Map<String, List<Integer>> getVersionMap(final NiFiRegistryClient
client, final Map<Pair<String, String>, String> flowMap,
+ final boolean
isInteractive) throws ParseException, IOException, NiFiRegistryException {
+ printMessage(isInteractive, ALL_FLOW_VERSIONS_COLLECTED);
+ return getVersionedFlowSnapshotMetadataList(client, flowMap)
+ .stream()
+
.collect(Collectors.groupingBy(VersionedFlowSnapshotMetadata::getFlowIdentifier,
+
Collectors.mapping(VersionedFlowSnapshotMetadata::getVersion,
Collectors.toList())));
+ }
+
+ private List<VersionedFlowSnapshotMetadata>
getVersionedFlowSnapshotMetadataList(final NiFiRegistryClient client,
+
final Map<Pair<String, String>, String> flowMap) throws ParseException,
IOException, NiFiRegistryException {
+ final List<VersionedFlowSnapshotMetadata> versions = new ArrayList<>();
+ for (final String flowIds : flowMap.values()) {
+ final Properties flowVersionProperties = new Properties();
+
flowVersionProperties.setProperty(CommandOption.FLOW_ID.getLongName(), flowIds);
+
+ versions.addAll(listFlowVersions.doExecute(client,
flowVersionProperties).getResult());
+ }
+ return versions;
+ }
+
+ private List<String> getFilePathList(final Properties properties) throws
MissingOptionException, NiFiRegistryException {
+ final String directory = getRequiredArg(properties,
CommandOption.INPUT_SOURCE);
+ final List<String> files;
+
+ try (final Stream<Path> paths = Files.list(Paths.get(directory))) {
+ files = paths
+ .filter(Files::isRegularFile)
+ .filter(path ->
path.getFileName().toString().startsWith(FILE_NAME_PREFIX))
+ .map(Path::toString)
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new NiFiRegistryException("File listing not possible due to
", e);
+ }
+ return files;
+ }
+
+ private List<ImportedSnapshot> deserializeSnapshots(final List<String>
files) throws IOException {
+ final List<ImportedSnapshot> importedSnapshots = new ArrayList<>();
+ final ObjectMapper mapper = JacksonUtils.getObjectMapper();
Review Comment:
`ObjectMapper` is thread safe, so this can be declared as a member variable
on the class.
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.toolkit.cli.impl.command.registry.flow;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.cli.MissingOptionException;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.curator.shaded.com.google.common.collect.ComparisonChain;
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.client.BucketClient;
+import org.apache.nifi.registry.client.FlowClient;
+import org.apache.nifi.registry.client.NiFiRegistryClient;
+import org.apache.nifi.registry.client.NiFiRegistryException;
+import org.apache.nifi.registry.flow.VersionedFlow;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
+import org.apache.nifi.toolkit.cli.api.CommandException;
+import org.apache.nifi.toolkit.cli.api.Context;
+import org.apache.nifi.toolkit.cli.impl.command.CommandOption;
+import
org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand;
+import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets;
+import org.apache.nifi.toolkit.cli.impl.result.StringResult;
+import org.apache.nifi.toolkit.cli.impl.util.JacksonUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ImportAllFlows extends AbstractNiFiRegistryCommand<StringResult> {
+ private static final String FILE_NAME_PREFIX =
"toolkit_registry_export_all_";
+ private static final String SKIPPING_BUCKET_CREATION = " already exists,
skipping bucket creation...";
+ private static final String SKIPPING_IMPORT = " already exists, skipping
import...";
+ private static final String SKIPPING_FLOW_CREATION = " already exists,
skipping flow creation...";
+ private static final String IMPORT_COMPLETED = "Import completed...";
+ private static final String ALL_BUCKETS_COLLECTED = "All buckets
collected...";
+ private static final String ALL_FLOWS_COLLECTED = "All flows collected...";
+ private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow
versions collected...";
+ private final ListBuckets listBuckets;
+ private final ListFlows listFlows;
+ private final ListFlowVersions listFlowVersions;
+ private final ImportFlowVersion importFlowVersion;
+
+ public ImportAllFlows() {
+ super("import-all-flows", StringResult.class);
+ this.listBuckets = new ListBuckets();
+ this.listFlows = new ListFlows();
+ this.listFlowVersions = new ListFlowVersions();
+ this.importFlowVersion = new ImportFlowVersion();
+ }
+
+ @Override
+ protected void doInitialize(Context context) {
+ addOption(CommandOption.INPUT_SOURCE.createOption());
+ addOption(CommandOption.SKIP.createOption());
+
+ listBuckets.initialize(context);
+ listFlows.initialize(context);
+ listFlowVersions.initialize(context);
+ importFlowVersion.initialize(context);
+ }
+
+ @Override
+ public String getDescription() {
+ return "From a provided directory as input, the directory content must
be generated by the export-all-flows command, " +
+ "based on the file contents, the corresponding buckets, flows
and flow versions will be created." +
+ "If not configured otherwise, already existing objects will be
skipped.";
+ }
+
+ @Override
+ public StringResult doExecute(final NiFiRegistryClient client, final
Properties properties) throws IOException, NiFiRegistryException,
ParseException, CommandException {
+ final boolean skip = Boolean.parseBoolean(getRequiredArg(properties,
CommandOption.SKIP));
+ final boolean isInteractive = getContext().isInteractive();
+
+ //Gather all buckets and create a map for easier search by bucket name
+ final Map<String, String> bucketMap = getBucketMap(client,
isInteractive);
+
+ // Gather all flows and create a map for easier search by flow name.
+ // As flow name is only unique within the same bucket we need to use
the bucket id in the key as well
+ final Map<Pair<String, String>, String> flowMap = getFlowMap(client,
bucketMap, isInteractive);
+ final Map<Pair<String, String>, String> flowCreated = new HashMap<>();
+
+ // Gather all flow versions and create a map for easier search by flow
id
+ final Map<String, List<Integer>> versionMap = getVersionMap(client,
flowMap, isInteractive);
+
+ // Create file path list
+ final List<String> files = getFilePathList(properties);
+
+ // Deserialize file content
+ final List<ImportedSnapshot> importedSnapshots =
deserializeSnapshots(files);
Review Comment:
Instead of serializing all snapshots, is it possible to iterate through each
one? It looks like that could create problems for sorting, so perhaps there is
no other way to avoid loading everything.
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.toolkit.cli.impl.command.registry.flow;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.cli.MissingOptionException;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.curator.shaded.com.google.common.collect.ComparisonChain;
+import org.apache.nifi.registry.bucket.Bucket;
+import org.apache.nifi.registry.client.BucketClient;
+import org.apache.nifi.registry.client.FlowClient;
+import org.apache.nifi.registry.client.NiFiRegistryClient;
+import org.apache.nifi.registry.client.NiFiRegistryException;
+import org.apache.nifi.registry.flow.VersionedFlow;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
+import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
+import org.apache.nifi.toolkit.cli.api.CommandException;
+import org.apache.nifi.toolkit.cli.api.Context;
+import org.apache.nifi.toolkit.cli.impl.command.CommandOption;
+import
org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand;
+import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets;
+import org.apache.nifi.toolkit.cli.impl.result.StringResult;
+import org.apache.nifi.toolkit.cli.impl.util.JacksonUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ImportAllFlows extends AbstractNiFiRegistryCommand<StringResult> {
+ private static final String FILE_NAME_PREFIX =
"toolkit_registry_export_all_";
+ private static final String SKIPPING_BUCKET_CREATION = " already exists,
skipping bucket creation...";
+ private static final String SKIPPING_IMPORT = " already exists, skipping
import...";
+ private static final String SKIPPING_FLOW_CREATION = " already exists,
skipping flow creation...";
+ private static final String IMPORT_COMPLETED = "Import completed...";
+ private static final String ALL_BUCKETS_COLLECTED = "All buckets
collected...";
+ private static final String ALL_FLOWS_COLLECTED = "All flows collected...";
+ private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow
versions collected...";
+ private final ListBuckets listBuckets;
+ private final ListFlows listFlows;
+ private final ListFlowVersions listFlowVersions;
+ private final ImportFlowVersion importFlowVersion;
+
+ public ImportAllFlows() {
+ super("import-all-flows", StringResult.class);
+ this.listBuckets = new ListBuckets();
+ this.listFlows = new ListFlows();
+ this.listFlowVersions = new ListFlowVersions();
+ this.importFlowVersion = new ImportFlowVersion();
+ }
+
+ @Override
+ protected void doInitialize(Context context) {
+ addOption(CommandOption.INPUT_SOURCE.createOption());
+ addOption(CommandOption.SKIP.createOption());
+
+ listBuckets.initialize(context);
+ listFlows.initialize(context);
+ listFlowVersions.initialize(context);
+ importFlowVersion.initialize(context);
+ }
+
+ @Override
+ public String getDescription() {
+ return "From a provided directory as input, the directory content must
be generated by the export-all-flows command, " +
+ "based on the file contents, the corresponding buckets, flows
and flow versions will be created." +
+ "If not configured otherwise, already existing objects will be
skipped.";
+ }
+
+ @Override
+ public StringResult doExecute(final NiFiRegistryClient client, final
Properties properties) throws IOException, NiFiRegistryException,
ParseException, CommandException {
+ final boolean skip = Boolean.parseBoolean(getRequiredArg(properties,
CommandOption.SKIP));
+ final boolean isInteractive = getContext().isInteractive();
+
+ //Gather all buckets and create a map for easier search by bucket name
+ final Map<String, String> bucketMap = getBucketMap(client,
isInteractive);
+
+ // Gather all flows and create a map for easier search by flow name.
+ // As flow name is only unique within the same bucket we need to use
the bucket id in the key as well
+ final Map<Pair<String, String>, String> flowMap = getFlowMap(client,
bucketMap, isInteractive);
+ final Map<Pair<String, String>, String> flowCreated = new HashMap<>();
+
+ // Gather all flow versions and create a map for easier search by flow
id
+ final Map<String, List<Integer>> versionMap = getVersionMap(client,
flowMap, isInteractive);
+
+ // Create file path list
+ final List<String> files = getFilePathList(properties);
+
+ // Deserialize file content
+ final List<ImportedSnapshot> importedSnapshots =
deserializeSnapshots(files);
+
+ // As we need to keep the version order the snapshot list needs to be
sorted
+ importedSnapshots.sort((o1, o2) -> ComparisonChain.start()
+ .compare(o1.getSnapshot().getBucket().getName(),
o2.getSnapshot().getBucket().getName())
+ .compare(o1.getSnapshot().getFlow().getName(),
o2.getSnapshot().getFlow().getName())
+ .compare(o1.getSnapshot().getSnapshotMetadata().getVersion(),
o2.getSnapshot().getSnapshotMetadata().getVersion())
+ .result());
+
+ for (final ImportedSnapshot snapshot : importedSnapshots) {
+
+ final String inputSource = snapshot.getInputSource();
+ final String bucketName =
snapshot.getSnapshot().getBucket().getName();
+ final String bucketDescription =
snapshot.getSnapshot().getBucket().getDescription();
+ final String flowName = snapshot.getSnapshot().getFlow().getName();
+ final String flowDescription =
snapshot.getSnapshot().getFlow().getDescription();
+ final int flowVersion =
snapshot.getSnapshot().getSnapshotMetadata().getVersion();
+ // The original bucket and flow ids must be kept otherwise NiFi
won't be able to synchronize with the NiFi Registry
+ final String bucketId =
snapshot.getSnapshot().getBucket().getIdentifier();
+ final String flowId =
snapshot.getSnapshot().getFlow().getIdentifier();
+
+ // Create bucket if missing
+ if (bucketMap.containsKey(bucketName)) {
+ printMessage(isInteractive, bucketName +
SKIPPING_BUCKET_CREATION);
+ } else {
+ createBucket(client, bucketMap, bucketName, bucketDescription,
bucketId);
+ }
+
+ // Create flow if missing
+ if (flowMap.containsKey(new ImmutablePair<>(bucketId, flowName))) {
+ if (skip) {
+ printMessage(isInteractive, flowName + SKIPPING_IMPORT);
+ continue;
+ } else {
+ printMessage(isInteractive, flowName +
SKIPPING_FLOW_CREATION);
+ }
+ } else if (!flowCreated.containsKey(new ImmutablePair<>(bucketId,
flowName))) {
+ createFlow(client, flowCreated, flowId, flowName,
flowDescription, bucketId);
+ }
+
+ // Create missing flow versions
+ if (!versionMap.getOrDefault(flowId,
Collections.emptyList()).contains(flowVersion)) {
+ createFlowVersion(client, inputSource, flowId);
+ }
+ }
+ return new StringResult(IMPORT_COMPLETED,
getContext().isInteractive());
+ }
+
+ private Map<String, String> getBucketMap(final NiFiRegistryClient client,
final boolean isInteractive) throws IOException, NiFiRegistryException {
+ printMessage(isInteractive, ALL_BUCKETS_COLLECTED);
+
+ return listBuckets.doExecute(client, new Properties())
+ .getResult()
+ .stream()
+ .collect(Collectors.toMap(Bucket::getName,
Bucket::getIdentifier));
+ }
+
+ private Map<Pair<String, String>, String> getFlowMap(final
NiFiRegistryClient client, final Map<String, String> bucketMap,
+ final boolean
isInteractive) throws ParseException, IOException, NiFiRegistryException {
+ printMessage(isInteractive, ALL_FLOWS_COLLECTED);
+ return getVersionedFlows(client, bucketMap)
+ .stream()
+ .collect(Collectors.toMap(e -> new
ImmutablePair<>(e.getBucketIdentifier(), e.getName()),
+ VersionedFlow::getIdentifier));
+ }
+
+ private List<VersionedFlow> getVersionedFlows(final NiFiRegistryClient
client, final Map<String, String> bucketMap) throws ParseException,
IOException, NiFiRegistryException {
+ final List<VersionedFlow> flows = new ArrayList<>();
+ for (final String id : bucketMap.values()) {
+ final Properties flowProperties = new Properties();
+ flowProperties.setProperty(CommandOption.BUCKET_ID.getLongName(),
id);
+
+ flows.addAll(listFlows.doExecute(client,
flowProperties).getResult());
+ }
+ return flows;
+ }
+
+ private Map<String, List<Integer>> getVersionMap(final NiFiRegistryClient
client, final Map<Pair<String, String>, String> flowMap,
+ final boolean
isInteractive) throws ParseException, IOException, NiFiRegistryException {
+ printMessage(isInteractive, ALL_FLOW_VERSIONS_COLLECTED);
+ return getVersionedFlowSnapshotMetadataList(client, flowMap)
+ .stream()
+
.collect(Collectors.groupingBy(VersionedFlowSnapshotMetadata::getFlowIdentifier,
+
Collectors.mapping(VersionedFlowSnapshotMetadata::getVersion,
Collectors.toList())));
+ }
+
+ private List<VersionedFlowSnapshotMetadata>
getVersionedFlowSnapshotMetadataList(final NiFiRegistryClient client,
+
final Map<Pair<String, String>, String> flowMap) throws ParseException,
IOException, NiFiRegistryException {
+ final List<VersionedFlowSnapshotMetadata> versions = new ArrayList<>();
+ for (final String flowIds : flowMap.values()) {
+ final Properties flowVersionProperties = new Properties();
+
flowVersionProperties.setProperty(CommandOption.FLOW_ID.getLongName(), flowIds);
+
+ versions.addAll(listFlowVersions.doExecute(client,
flowVersionProperties).getResult());
+ }
+ return versions;
+ }
+
+ private List<String> getFilePathList(final Properties properties) throws
MissingOptionException, NiFiRegistryException {
+ final String directory = getRequiredArg(properties,
CommandOption.INPUT_SOURCE);
+ final List<String> files;
+
+ try (final Stream<Path> paths = Files.list(Paths.get(directory))) {
+ files = paths
+ .filter(Files::isRegularFile)
+ .filter(path ->
path.getFileName().toString().startsWith(FILE_NAME_PREFIX))
+ .map(Path::toString)
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new NiFiRegistryException("File listing not possible due to
", e);
Review Comment:
This approach to error messages needs to be changed. The message should
stand by itself, it will not be concatenated with the exception. Recommend the
following:
```suggestion
throw new NiFiRegistryException("File listing failed", e);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
