You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2015/01/16 03:29:32 UTC
[15/51] [partial] incubator-nifi git commit: Reworked overall
directory structure to make releasing nifi vs maven plugis easier
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java
deleted file mode 100644
index 72b594a..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java
+++ /dev/null
@@ -1,600 +0,0 @@
-/*
- * 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.flow.impl;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.FilenameFilter;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.util.Arrays;
-import java.util.UUID;
-
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.JAXBException;
-import javax.xml.bind.Marshaller;
-import javax.xml.bind.Unmarshaller;
-import javax.xml.bind.annotation.XmlRootElement;
-import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
-import javax.xml.parsers.DocumentBuilder;
-import javax.xml.parsers.DocumentBuilderFactory;
-import javax.xml.transform.OutputKeys;
-import javax.xml.transform.Transformer;
-import javax.xml.transform.TransformerFactory;
-import javax.xml.transform.dom.DOMSource;
-import javax.xml.transform.stream.StreamResult;
-
-import org.apache.commons.compress.archivers.ArchiveEntry;
-import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
-import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
-import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
-import org.apache.nifi.cluster.flow.ClusterDataFlow;
-import org.apache.nifi.cluster.flow.DaoException;
-import org.apache.nifi.cluster.flow.DataFlowDao;
-import org.apache.nifi.cluster.flow.PersistedFlowState;
-import org.apache.nifi.cluster.protocol.DataFlow;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.cluster.protocol.StandardDataFlow;
-import org.apache.nifi.cluster.protocol.jaxb.message.NodeIdentifierAdapter;
-import org.apache.nifi.logging.NiFiLog;
-import org.apache.nifi.stream.io.BufferedInputStream;
-import org.apache.nifi.stream.io.BufferedOutputStream;
-import org.apache.nifi.stream.io.ByteArrayInputStream;
-import org.apache.nifi.stream.io.StreamUtils;
-import org.apache.nifi.util.file.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.w3c.dom.Document;
-import org.w3c.dom.Element;
-
-/**
- * Implements the FlowDao interface. The implementation tracks the state of the
- * dataflow by annotating the filename of the flow state file. Specifically, the
- * implementation correlates PersistedFlowState states to filename extensions.
- * The correlation is as follows:
- * <ul>
- * <li> CURRENT maps to flow.xml </li>
- * <li> STALE maps to flow.xml.stale </li>
- * <li> UNKNOWN maps to flow.xml.unknown </li>
- * </ul>
- * Whenever the flow state changes, the flow state file's name is updated to
- * denote its state.
- *
- * The implementation also provides for a restore directory that may be
- * configured for higher availability. At instance creation, if the primary or
- * restore directories have multiple flow state files, an exception is thrown.
- * If the primary directory has a current flow state file, but the restore
- * directory does not, then the primary flow state file is copied to the restore
- * directory. If the restore directory has a current flow state file, but the
- * primary directory does not, then the restore flow state file is copied to the
- * primary directory. If both the primary and restore directories have a current
- * flow state file and the files are different, then an exception is thrown.
- *
- * When the flow state file is saved, it is always saved first to the restore
- * directory followed by a save to the primary directory. When the flow state
- * file is loaded, a check is made to verify that the primary and restore flow
- * state files are both current. If either is not current, then an exception is
- * thrown. The primary flow state file is always read when the load method is
- * called.
- *
- * @author unattributed
- */
-public class DataFlowDaoImpl implements DataFlowDao {
-
- private final File primaryDirectory;
- private final File restoreDirectory;
- private final boolean autoStart;
- private final String generatedRootGroupId = UUID.randomUUID().toString();
-
- public static final String STALE_EXT = ".stale";
- public static final String UNKNOWN_EXT = ".unknown";
- public static final String FLOW_PACKAGE = "flow.tar";
- public static final String FLOW_XML_FILENAME = "flow.xml";
- public static final String TEMPLATES_FILENAME = "templates.xml";
- public static final String SNIPPETS_FILENAME = "snippets.xml";
- public static final String CLUSTER_INFO_FILENAME = "cluster-info.xml";
-
- private static final Logger logger = new NiFiLog(LoggerFactory.getLogger(DataFlowDaoImpl.class));
-
- public DataFlowDaoImpl(final File primaryDirectory) throws DaoException {
- this(primaryDirectory, null, false);
- }
-
- public DataFlowDaoImpl(final File primaryDirectory, final File restoreDirectory, final boolean autoStart) throws DaoException {
-
- // sanity check that primary directory is a directory, creating it if necessary
- if (primaryDirectory == null) {
- throw new IllegalArgumentException("Primary directory may not be null.");
- } else if (!primaryDirectory.exists()) {
- if (!primaryDirectory.mkdir()) {
- throw new DaoException(String.format("Failed to create primary directory '%s'", primaryDirectory.getAbsolutePath()));
- }
- } else if (!primaryDirectory.isDirectory()) {
- throw new IllegalArgumentException("Primary directory must be a directory.");
- }
-
- this.autoStart = autoStart;
-
- try {
- this.primaryDirectory = primaryDirectory;
- this.restoreDirectory = restoreDirectory;
-
- if (restoreDirectory == null) {
- // check that we have exactly one current flow state file
- ensureSingleCurrentStateFile(primaryDirectory);
- } else {
-
- // check that restore directory is a directory, creating it if necessary
- FileUtils.ensureDirectoryExistAndCanAccess(restoreDirectory);
-
- // check that restore directory is not the same as the primary directory
- if (primaryDirectory.getAbsolutePath().equals(restoreDirectory.getAbsolutePath())) {
- throw new IllegalArgumentException(String.format("Primary directory '%s' is the same as restore directory '%s' ",
- primaryDirectory.getAbsolutePath(), restoreDirectory.getAbsolutePath()));
- }
-
- final File[] primaryFlowStateFiles = getFlowStateFiles(primaryDirectory);
- final File[] restoreFlowStateFiles = getFlowStateFiles(restoreDirectory);
-
- // if more than one state file in either primary or restore, then throw exception
- if (primaryFlowStateFiles.length > 1) {
- throw new IllegalStateException(String.format("Found multiple dataflow state files in primary directory '%s'", primaryDirectory));
- } else if (restoreFlowStateFiles.length > 1) {
- throw new IllegalStateException(String.format("Found multiple dataflow state files in restore directory '%s'", restoreDirectory));
- }
-
- // check that the single primary state file we found is current or create a new one
- final File primaryFlowStateFile = ensureSingleCurrentStateFile(primaryDirectory);
-
- // check that the single restore state file we found is current or create a new one
- final File restoreFlowStateFile = ensureSingleCurrentStateFile(restoreDirectory);
-
- // if there was a difference in flow state file directories, then copy the appropriate files
- if (restoreFlowStateFiles.length == 0 && primaryFlowStateFiles.length != 0) {
- // copy primary state file to restore
- FileUtils.copyFile(primaryFlowStateFile, restoreFlowStateFile, false, false, logger);
- } else if (primaryFlowStateFiles.length == 0 && restoreFlowStateFiles.length != 0) {
- // copy restore state file to primary
- FileUtils.copyFile(restoreFlowStateFile, primaryFlowStateFile, false, false, logger);
- } else {
- // sync the primary copy with the restore copy
- syncWithRestore(primaryFlowStateFile, restoreFlowStateFile);
- }
-
- }
- } catch (final IOException | IllegalArgumentException | IllegalStateException | JAXBException ex) {
- throw new DaoException(ex);
- }
- }
-
-
- private void syncWithRestore(final File primaryFile, final File restoreFile) throws IOException {
- try (final FileInputStream primaryFis = new FileInputStream(primaryFile);
- final TarArchiveInputStream primaryIn = new TarArchiveInputStream(primaryFis);
- final FileInputStream restoreFis = new FileInputStream(restoreFile);
- final TarArchiveInputStream restoreIn = new TarArchiveInputStream(restoreFis)) {
-
- final ArchiveEntry primaryEntry = primaryIn.getNextEntry();
- final ArchiveEntry restoreEntry = restoreIn.getNextEntry();
-
- if ( primaryEntry == null && restoreEntry == null ) {
- return;
- }
-
- if ( (primaryEntry == null && restoreEntry != null) || (primaryEntry != null && restoreEntry == null) ) {
- throw new IllegalStateException(String.format("Primary file '%s' is different than restore file '%s'",
- primaryFile.getAbsoluteFile(), restoreFile.getAbsolutePath()));
- }
-
- final byte[] primaryMd5 = calculateMd5(primaryIn);
- final byte[] restoreMd5 = calculateMd5(restoreIn);
-
- if ( !Arrays.equals(primaryMd5, restoreMd5) ) {
- throw new IllegalStateException(String.format("Primary file '%s' is different than restore file '%s'",
- primaryFile.getAbsoluteFile(), restoreFile.getAbsolutePath()));
- }
- }
- }
-
- private byte[] calculateMd5(final InputStream in) throws IOException {
- final MessageDigest digest;
- try {
- digest = MessageDigest.getInstance("MD5");
- } catch (final NoSuchAlgorithmException nsae) {
- throw new IOException(nsae);
- }
-
- int len;
- final byte[] buffer = new byte[8192];
- while ((len = in.read(buffer)) > -1) {
- if (len > 0) {
- digest.update(buffer, 0, len);
- }
- }
- return digest.digest();
- }
-
- @Override
- public ClusterDataFlow loadDataFlow() throws DaoException {
- try {
- return parseDataFlow(getExistingFlowStateFile(primaryDirectory));
- } catch (final IOException | JAXBException ex) {
- throw new DaoException(ex);
- }
- }
-
- @Override
- public void saveDataFlow(final ClusterDataFlow dataFlow) throws DaoException {
- try {
-
- final File primaryStateFile = getFlowStateFile(primaryDirectory);
-
- // write to restore before writing to primary in case primary experiences problems
- if (restoreDirectory != null) {
- final File restoreStateFile = getFlowStateFile(restoreDirectory);
- if (restoreStateFile == null) {
- if (primaryStateFile == null) {
- writeDataFlow(createNewFlowStateFile(restoreDirectory), dataFlow);
- } else {
- throw new DaoException(String.format("Unable to save dataflow because dataflow state file in primary directory '%s' exists, but it does not exist in the restore directory '%s'",
- primaryDirectory.getAbsolutePath(), restoreDirectory.getAbsolutePath()));
- }
- } else {
- if (primaryStateFile == null) {
- throw new DaoException(String.format("Unable to save dataflow because dataflow state file in restore directory '%s' exists, but it does not exist in the primary directory '%s'",
- restoreDirectory.getAbsolutePath(), primaryDirectory.getAbsolutePath()));
- } else {
- final PersistedFlowState primaryFlowState = getPersistedFlowState(primaryStateFile);
- final PersistedFlowState restoreFlowState = getPersistedFlowState(restoreStateFile);
- if (primaryFlowState == restoreFlowState) {
- writeDataFlow(restoreStateFile, dataFlow);
- } else {
- throw new DaoException(String.format("Unable to save dataflow because state file in primary directory '%s' has state '%s', but the state file in the restore directory '%s' has state '%s'",
- primaryDirectory.getAbsolutePath(), primaryFlowState, restoreDirectory.getAbsolutePath(), restoreFlowState));
- }
- }
- }
- }
-
- // write dataflow to primary
- if (primaryStateFile == null) {
- writeDataFlow(createNewFlowStateFile(primaryDirectory), dataFlow);
- } else {
- writeDataFlow(primaryStateFile, dataFlow);
- }
-
- } catch (final IOException | JAXBException ex) {
- throw new DaoException(ex);
- }
- }
-
- @Override
- public PersistedFlowState getPersistedFlowState() {
- // trust restore over primary if configured for restore
- if (restoreDirectory == null) {
- return getPersistedFlowState(getExistingFlowStateFile(primaryDirectory));
- } else {
- return getPersistedFlowState(getExistingFlowStateFile(restoreDirectory));
- }
- }
-
- @Override
- public void setPersistedFlowState(final PersistedFlowState flowState) throws DaoException {
- // rename restore before primary if configured for restore
- if (restoreDirectory != null) {
- renameFlowStateFile(getExistingFlowStateFile(restoreDirectory), flowState);
- }
- renameFlowStateFile(getExistingFlowStateFile(primaryDirectory), flowState);
- }
-
- private File ensureSingleCurrentStateFile(final File dir) throws IOException, JAXBException {
-
- // ensure that we have at most one state file and if we have one, it is current
- final File[] directoryFlowStateFiles = getFlowStateFiles(dir);
- if (directoryFlowStateFiles.length > 1) {
- throw new DaoException(String.format("Found multiple dataflow state files in directory '%s'", dir));
- } else if (directoryFlowStateFiles.length == 0) {
- // create a new file if none exist
- return createNewFlowStateFile(dir);
- } else {
- // check that the single flow state file is current
- final PersistedFlowState flowState = getPersistedFlowState(directoryFlowStateFiles[0]);
- if (PersistedFlowState.CURRENT == flowState) {
- return directoryFlowStateFiles[0];
- } else {
- throw new DaoException(String.format("Dataflow state file '%s' must be current.", directoryFlowStateFiles[0].getAbsolutePath()));
- }
- }
-
- }
-
- private PersistedFlowState getPersistedFlowState(final File file) {
- final String path = file.getAbsolutePath();
- if (path.endsWith(STALE_EXT)) {
- return PersistedFlowState.STALE;
- } else if (path.endsWith(UNKNOWN_EXT)) {
- return PersistedFlowState.UNKNOWN;
- } else {
- return PersistedFlowState.CURRENT;
- }
- }
-
- private File getFlowStateFile(final File dir) {
- final File[] files = getFlowStateFiles(dir);
- if (files.length > 1) {
- throw new IllegalStateException(String.format("Expected at most one dataflow state file, but found %s files.", files.length));
- } else if (files.length == 0) {
- return null;
- } else {
- return files[0];
- }
- }
-
- private File getExistingFlowStateFile(final File dir) {
- final File file = getFlowStateFile(dir);
- if (file == null) {
- throw new IllegalStateException(String.format("Expected a dataflow state file, but none existed in directory '%s'", dir.getAbsolutePath()));
- }
- return file;
- }
-
- private File[] getFlowStateFiles(final File dir) {
- final File[] files = dir.listFiles(new FilenameFilter() {
- @Override
- public boolean accept(File dir, String name) {
- return (name.equals(FLOW_PACKAGE) || name.endsWith(STALE_EXT) || name.endsWith(UNKNOWN_EXT));
- }
- });
-
- if (files == null) {
- return new File[0];
- } else {
- return files;
- }
- }
-
- private File removeStateFileExtension(final File file) {
-
- final String path = file.getAbsolutePath();
- final int stateFileExtIndex;
- if (path.endsWith(STALE_EXT)) {
- stateFileExtIndex = path.lastIndexOf(STALE_EXT);
- } else if (path.endsWith(UNKNOWN_EXT)) {
- stateFileExtIndex = path.lastIndexOf(UNKNOWN_EXT);
- } else {
- stateFileExtIndex = path.length();
- }
-
- return new File(path.substring(0, stateFileExtIndex));
- }
-
- private File addStateFileExtension(final File file, final PersistedFlowState state) {
- switch (state) {
- case CURRENT: {
- return file;
- }
- case STALE: {
- return new File(file.getAbsolutePath() + STALE_EXT);
- }
- case UNKNOWN: {
- return new File(file.getAbsolutePath() + UNKNOWN_EXT);
- }
- default: {
- throw new RuntimeException("Unsupported PersistedFlowState Enum value: " + state);
- }
- }
- }
-
- private File createNewFlowStateFile(final File dir) throws IOException, JAXBException {
- final File stateFile = new File(dir, FLOW_PACKAGE);
- stateFile.createNewFile();
-
- final byte[] flowBytes = getEmptyFlowBytes();
- final byte[] templateBytes = new byte[0];
- final byte[] snippetBytes = new byte[0];
- final DataFlow dataFlow = new StandardDataFlow(flowBytes, templateBytes, snippetBytes);
-
- final ClusterMetadata clusterMetadata = new ClusterMetadata();
- writeDataFlow(stateFile, dataFlow, clusterMetadata);
-
- return stateFile;
- }
-
- private byte[] getEmptyFlowBytes() throws IOException {
- try {
- final DocumentBuilder docBuilder = DocumentBuilderFactory.newInstance().newDocumentBuilder();
- final Document document = docBuilder.newDocument();
-
- final Element controller = document.createElement("flowController");
- document.appendChild(controller);
-
- controller.appendChild(createTextElement(document, "maxThreadCount", "15"));
-
- final Element rootGroup = document.createElement("rootGroup");
- rootGroup.appendChild(createTextElement(document, "id", generatedRootGroupId));
- rootGroup.appendChild(createTextElement(document, "name", "NiFi Flow"));
-
- // create the position element
- final Element positionElement = createTextElement(document, "position", "");
- positionElement.setAttribute("x", "0.0");
- positionElement.setAttribute("y", "0.0");
- rootGroup.appendChild(positionElement);
-
- rootGroup.appendChild(createTextElement(document, "comment", ""));
- controller.appendChild(rootGroup);
-
- final Transformer transformer = TransformerFactory.newInstance().newTransformer();
- transformer.setOutputProperty("{http://xml.apache.org/xslt}indent-amount", "2");
- transformer.setOutputProperty(OutputKeys.INDENT, "yes");
-
- final DOMSource source = new DOMSource(document);
- final ByteArrayOutputStream baos = new ByteArrayOutputStream();
- final StreamResult result = new StreamResult(baos);
- transformer.transform(source, result);
-
- return baos.toByteArray();
- } catch (final Exception e) {
- throw new IOException(e);
- }
- }
-
- private Element createTextElement(final Document document, final String elementName, final String value) {
- final Element element = document.createElement(elementName);
- element.setTextContent(value);
- return element;
- }
-
- private void renameFlowStateFile(final File flowStateFile, final PersistedFlowState newState) throws DaoException {
- final PersistedFlowState existingState = getPersistedFlowState(flowStateFile);
- if (existingState != newState) {
- final File newFlowStateFile = addStateFileExtension(removeStateFileExtension(flowStateFile), newState);
- if (flowStateFile.renameTo(newFlowStateFile) == false) {
- throw new DaoException(
- String.format("Failed to rename flow state file '%s' to new name '%s'", flowStateFile.getAbsolutePath(), newFlowStateFile.getAbsolutePath()));
- }
- }
- }
-
- private ClusterDataFlow parseDataFlow(final File file) throws IOException, JAXBException, DaoException {
- byte[] flowBytes = new byte[0];
- byte[] templateBytes = new byte[0];
- byte[] snippetBytes = new byte[0];
- byte[] clusterInfoBytes = new byte[0];
-
- try (final InputStream inStream = new FileInputStream(file);
- final TarArchiveInputStream tarIn = new TarArchiveInputStream(new BufferedInputStream(inStream))) {
- TarArchiveEntry tarEntry;
- while ((tarEntry = tarIn.getNextTarEntry()) != null) {
- switch (tarEntry.getName()) {
- case FLOW_XML_FILENAME:
- flowBytes = new byte[(int) tarEntry.getSize()];
- StreamUtils.fillBuffer(tarIn, flowBytes, true);
- break;
- case TEMPLATES_FILENAME:
- templateBytes = new byte[(int) tarEntry.getSize()];
- StreamUtils.fillBuffer(tarIn, templateBytes, true);
- break;
- case SNIPPETS_FILENAME:
- snippetBytes = new byte[(int) tarEntry.getSize()];
- StreamUtils.fillBuffer(tarIn, snippetBytes, true);
- break;
- case CLUSTER_INFO_FILENAME:
- clusterInfoBytes = new byte[(int) tarEntry.getSize()];
- StreamUtils.fillBuffer(tarIn, clusterInfoBytes, true);
- break;
- default:
- throw new DaoException("Found Unexpected file in dataflow configuration: " + tarEntry.getName());
- }
- }
- }
-
- final ClusterMetadata clusterMetadata;
- if (clusterInfoBytes.length == 0) {
- clusterMetadata = null;
- } else {
- final Unmarshaller clusterMetadataUnmarshaller = ClusterMetadata.jaxbCtx.createUnmarshaller();
- clusterMetadata = (ClusterMetadata) clusterMetadataUnmarshaller.unmarshal(new ByteArrayInputStream(clusterInfoBytes));
- }
-
- final StandardDataFlow dataFlow = new StandardDataFlow(flowBytes, templateBytes, snippetBytes);
- dataFlow.setAutoStartProcessors(autoStart);
-
- return new ClusterDataFlow(dataFlow, (clusterMetadata == null) ? null : clusterMetadata.getPrimaryNodeId());
- }
-
- private void writeDataFlow(final File file, final ClusterDataFlow clusterDataFlow) throws IOException, JAXBException {
-
- // get the data flow
- DataFlow dataFlow = clusterDataFlow.getDataFlow();
-
- // if no dataflow, then write a new dataflow
- if (dataFlow == null) {
- dataFlow = new StandardDataFlow(new byte[0], new byte[0], new byte[0]);
- }
-
- // setup the cluster metadata
- final ClusterMetadata clusterMetadata = new ClusterMetadata();
- clusterMetadata.setPrimaryNodeId(clusterDataFlow.getPrimaryNodeId());
-
- // write to disk
- writeDataFlow(file, dataFlow, clusterMetadata);
- }
-
- private void writeTarEntry(final TarArchiveOutputStream tarOut, final String filename, final byte[] bytes) throws IOException {
- final TarArchiveEntry flowEntry = new TarArchiveEntry(filename);
- flowEntry.setSize(bytes.length);
- tarOut.putArchiveEntry(flowEntry);
- tarOut.write(bytes);
- tarOut.closeArchiveEntry();
- }
-
- private void writeDataFlow(final File file, final DataFlow dataFlow, final ClusterMetadata clusterMetadata) throws IOException, JAXBException {
-
- try (final OutputStream fos = new FileOutputStream(file);
- final TarArchiveOutputStream tarOut = new TarArchiveOutputStream(new BufferedOutputStream(fos))) {
-
- writeTarEntry(tarOut, FLOW_XML_FILENAME, dataFlow.getFlow());
- writeTarEntry(tarOut, TEMPLATES_FILENAME, dataFlow.getTemplates());
- writeTarEntry(tarOut, SNIPPETS_FILENAME, dataFlow.getSnippets());
-
- final ByteArrayOutputStream baos = new ByteArrayOutputStream(256);
- writeClusterMetadata(clusterMetadata, baos);
- final byte[] clusterInfoBytes = baos.toByteArray();
-
- writeTarEntry(tarOut, CLUSTER_INFO_FILENAME, clusterInfoBytes);
- }
- }
-
- private void writeClusterMetadata(final ClusterMetadata clusterMetadata, final OutputStream os) throws IOException, JAXBException {
- // write cluster metadata to output stream
- final Marshaller marshaller = ClusterMetadata.jaxbCtx.createMarshaller();
- marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
- marshaller.setProperty(Marshaller.JAXB_FRAGMENT, true);
- marshaller.setProperty(Marshaller.JAXB_ENCODING, "UTF-8");
- marshaller.marshal(clusterMetadata, os);
- }
-
- @XmlRootElement(name = "clusterMetadata")
- private static class ClusterMetadata {
-
- private NodeIdentifier primaryNodeId;
-
- private static final JAXBContext jaxbCtx;
-
- static {
- try {
- jaxbCtx = JAXBContext.newInstance(ClusterMetadata.class);
- } catch (final JAXBException je) {
- throw new RuntimeException(je);
- }
- }
-
- @XmlJavaTypeAdapter(NodeIdentifierAdapter.class)
- public NodeIdentifier getPrimaryNodeId() {
- return primaryNodeId;
- }
-
- public void setPrimaryNodeId(final NodeIdentifier primaryNodeId) {
- this.primaryNodeId = primaryNodeId;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java
deleted file mode 100644
index e135af3..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java
+++ /dev/null
@@ -1,356 +0,0 @@
-/*
- * 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.flow.impl;
-
-import java.util.Collections;
-import java.util.Date;
-import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.nifi.cluster.flow.ClusterDataFlow;
-import org.apache.nifi.cluster.flow.DaoException;
-import org.apache.nifi.cluster.flow.DataFlowDao;
-import org.apache.nifi.cluster.flow.DataFlowManagementService;
-import org.apache.nifi.cluster.flow.PersistedFlowState;
-import org.apache.nifi.cluster.protocol.ClusterManagerProtocolSender;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.cluster.protocol.StandardDataFlow;
-import org.apache.nifi.cluster.protocol.message.FlowRequestMessage;
-import org.apache.nifi.cluster.protocol.message.FlowResponseMessage;
-import org.apache.nifi.logging.NiFiLog;
-import org.apache.nifi.util.FormatUtils;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Implements FlowManagementService interface. The service tries to keep the
- * cluster's flow current with regards to the available nodes.
- *
- * The instance may be configured with a retrieval delay, which will reduce the
- * number of retrievals performed by the service at the expense of increasing
- * the chances that the service will not be able to provide a current flow to
- * the caller.
- *
- * By default, the service will try to update the flow as quickly as possible.
- * Configuring a delay enables a less aggressive retrieval strategy.
- * Specifically, the eligible retrieval time is reset every time the flow state
- * is set to STALE. If the state is set to UNKNOWN or CURRENT, then the flow
- * will not be retrieved.
- *
- * @author unattributed
- */
-public class DataFlowManagementServiceImpl implements DataFlowManagementService {
-
- /*
- * Developer Note:
- *
- * This class maintains an ExecutorService and a Runnable.
- * Although the class is not externally threadsafe, its internals are protected to
- * accommodate multithread access between the ExecutorServer and the Runnable.
- *
- */
- private static final Logger logger = new NiFiLog(LoggerFactory.getLogger(DataFlowManagementServiceImpl.class));
-
- private final DataFlowDao flowDao;
-
- private final ClusterManagerProtocolSender sender;
-
- private final Set<NodeIdentifier> nodeIds = new CopyOnWriteArraySet<>();
-
- private final AtomicBoolean stopRequested = new AtomicBoolean(false);
-
- private final AtomicLong lastRetrievalTime = new AtomicLong(-1);
-
- private Timer flowRetriever;
-
- private long retrievableAfterTime = 0L;
-
- private AtomicInteger retrievalDelaySeconds = new AtomicInteger(0);
-
- private final TimingReentrantLock resourceLock = new TimingReentrantLock(new ReentrantLock());
-
- public DataFlowManagementServiceImpl(final DataFlowDao flowDao, final ClusterManagerProtocolSender sender) {
- if (flowDao == null) {
- throw new IllegalArgumentException("Flow DAO may not be null.");
- } else if (sender == null) {
- throw new IllegalArgumentException("Cluster Manager Protocol Sender may not be null.");
- }
- this.flowDao = flowDao;
- this.sender = sender;
- }
-
- @Override
- public void start() {
-
- if (isRunning()) {
- throw new IllegalArgumentException("Instance is already running.");
- }
-
- // reset stop requested
- stopRequested.set(false);
-
- // setup flow retreiver timer
- flowRetriever = new Timer("Flow Management Service", /* is daemon */ true);
- flowRetriever.schedule(new FlowRetrieverTimerTask(), 0, 500);
- }
-
- @Override
- public boolean isRunning() {
- return (flowRetriever != null);
- }
-
- @Override
- public void stop() {
-
- if (isRunning() == false) {
- throw new IllegalArgumentException("Instance is already stopped.");
- }
-
- // record stop request
- stopRequested.set(true);
-
- flowRetriever.cancel();
- flowRetriever = null;
-
- }
-
- @Override
- public ClusterDataFlow loadDataFlow() throws DaoException {
- resourceLock.lock();
- try {
- return flowDao.loadDataFlow();
- } finally {
- resourceLock.unlock("loadDataFlow");
- }
- }
-
- @Override
- public void updatePrimaryNode(final NodeIdentifier nodeId) {
- resourceLock.lock();
- try {
- final ClusterDataFlow existingClusterDataFlow = flowDao.loadDataFlow();
-
- final StandardDataFlow dataFlow;
- if (existingClusterDataFlow == null) {
- dataFlow = null;
- } else {
- dataFlow = existingClusterDataFlow.getDataFlow();
- }
-
- flowDao.saveDataFlow(new ClusterDataFlow(dataFlow, nodeId));
- } finally {
- resourceLock.unlock("updatePrimaryNode");
- }
- }
-
- @Override
- public PersistedFlowState getPersistedFlowState() {
- resourceLock.lock();
- try {
- return flowDao.getPersistedFlowState();
- } finally {
- resourceLock.unlock("getPersistedFlowState");
- }
- }
-
- @Override
- public boolean isFlowCurrent() {
- return PersistedFlowState.CURRENT == getPersistedFlowState();
- }
-
- @Override
- public void setPersistedFlowState(final PersistedFlowState flowState) {
- // lock to ensure state change and retrievable time update are atomic
- resourceLock.lock();
- try {
- flowDao.setPersistedFlowState(flowState);
- if (PersistedFlowState.STALE == flowState) {
- retrievableAfterTime = new Date().getTime() + (getRetrievalDelaySeconds() * 1000);
- } else if (PersistedFlowState.UNKNOWN == flowState || PersistedFlowState.CURRENT == flowState) {
- retrievableAfterTime = Long.MAX_VALUE;
- }
- } finally {
- resourceLock.unlock("setPersistedFlowState");
- }
- }
-
- @Override
- public Set<NodeIdentifier> getNodeIds() {
- return Collections.unmodifiableSet(nodeIds);
- }
-
- @Override
- public void setNodeIds(final Set<NodeIdentifier> nodeIds) {
-
- if (nodeIds == null) {
- throw new IllegalArgumentException("Node IDs may not be null.");
- }
-
- resourceLock.lock();
- try {
-
- if (this.nodeIds.equals(nodeIds)) {
- return;
- }
-
- this.nodeIds.clear();
- this.nodeIds.addAll(nodeIds);
-
- } finally {
- resourceLock.unlock("setNodeIds");
- }
-
- }
-
- @Override
- public int getRetrievalDelaySeconds() {
- return retrievalDelaySeconds.get();
- }
-
- @Override
- public void setRetrievalDelay(final String retrievalDelay) {
- this.retrievalDelaySeconds.set((int) FormatUtils.getTimeDuration(retrievalDelay, TimeUnit.SECONDS));
- }
-
- public ClusterManagerProtocolSender getSender() {
- return sender;
- }
-
- public long getLastRetrievalTime() {
- return lastRetrievalTime.get();
- }
-
- /**
- * A timer task for issuing FlowRequestMessage messages to nodes to retrieve
- * an updated flow.
- */
- private class FlowRetrieverTimerTask extends TimerTask {
-
- @Override
- public void run() {
-
- resourceLock.lock();
- try {
- // if flow is current, then we're done
- if (isFlowCurrent()) {
- return;
- }
- } catch (final Exception ex) {
- logger.info("Encountered exception checking if flow is current caused by " + ex, ex);
- } finally {
- resourceLock.unlock("FlowRetrieverTimerTask - isFlowCurrent");
- }
-
- final FlowRequestMessage request = new FlowRequestMessage();
- for (final NodeIdentifier nodeId : getNodeIds()) {
- try {
- // setup request
- request.setNodeId(nodeId);
-
- // record request time
- final long requestSentTime = new Date().getTime();
-
- resourceLock.lock();
- try {
- // sanity checks before making request
- if (stopRequested.get()) { // did we receive a stop request
- logger.debug("Stopping runnable prematurely because a request to stop was issued.");
- return;
- } else if (requestSentTime < retrievableAfterTime) {
- /*
- * Retrievable after time was updated while obtaining
- * the lock, so try again later
- */
- return;
- }
- } finally {
- resourceLock.unlock("FlowRetrieverTimerTask - check stopRequested");
- }
-
- // send request
- final FlowResponseMessage response = sender.requestFlow(request);
-
- resourceLock.lock();
- try {
- // check if the retrieved flow is still valid
- if (requestSentTime > retrievableAfterTime) {
- logger.info("Saving retrieved flow.");
-
- final StandardDataFlow dataFlow = response.getDataFlow();
- final ClusterDataFlow existingClusterDataFlow = flowDao.loadDataFlow();
- final ClusterDataFlow currentClusterDataFlow;
- if (existingClusterDataFlow == null) {
- currentClusterDataFlow = new ClusterDataFlow(dataFlow, null);
- } else {
- currentClusterDataFlow = new ClusterDataFlow(dataFlow, existingClusterDataFlow.getPrimaryNodeId());
- }
- flowDao.saveDataFlow(currentClusterDataFlow);
- flowDao.setPersistedFlowState(PersistedFlowState.CURRENT);
- lastRetrievalTime.set(new Date().getTime());
- }
-
- /*
- * Retrievable after time was updated while requesting
- * the flow, so try again later.
- */
- } finally {
- resourceLock.unlock("FlowRetrieverTimerTask - saveDataFlow");
- }
-
- } catch (final Throwable t) {
- logger.info("Encountered exception retrieving flow from node " + nodeId + " caused by " + t, t);
- }
- }
- }
- }
-
- private static class TimingReentrantLock {
-
- private final Lock lock;
- private static final Logger logger = LoggerFactory.getLogger("dataFlowManagementService.lock");
-
- private final ThreadLocal<Long> lockTime = new ThreadLocal<>();
-
- public TimingReentrantLock(final Lock lock) {
- this.lock = lock;
- }
-
- public void lock() {
- lock.lock();
- lockTime.set(System.nanoTime());
- }
-
- public void unlock(final String task) {
- final long nanosLocked = System.nanoTime() - lockTime.get();
- lock.unlock();
-
- final long millisLocked = TimeUnit.MILLISECONDS.convert(nanosLocked, TimeUnit.NANOSECONDS);
- if (millisLocked > 100L) {
- logger.debug("Lock held for {} milliseconds for task: {}", millisLocked, task);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
deleted file mode 100644
index 0fcac8c..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
- * 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.manager;
-
-import org.apache.nifi.cluster.manager.exception.PrimaryRoleAssignmentException;
-import org.apache.nifi.cluster.manager.exception.IllegalNodeDisconnectionException;
-import org.apache.nifi.cluster.manager.exception.UnknownNodeException;
-import org.apache.nifi.cluster.manager.exception.IneligiblePrimaryNodeException;
-import org.apache.nifi.cluster.manager.exception.IllegalNodeDeletionException;
-import org.apache.nifi.cluster.manager.exception.IllegalNodeReconnectionException;
-import org.apache.nifi.cluster.manager.exception.NodeDisconnectionException;
-import org.apache.nifi.cluster.NodeInformant;
-import org.apache.nifi.cluster.event.Event;
-import org.apache.nifi.cluster.node.Node;
-import org.apache.nifi.cluster.node.Node.Status;
-import org.apache.nifi.cluster.protocol.ConnectionRequest;
-import org.apache.nifi.cluster.protocol.ConnectionResponse;
-import org.apache.nifi.cluster.protocol.Heartbeat;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.controller.status.ProcessGroupStatus;
-import org.apache.nifi.diagnostics.SystemDiagnostics;
-import org.apache.nifi.reporting.BulletinRepository;
-
-import java.util.List;
-import java.util.Set;
-
-/**
- * Defines the interface for a ClusterManager. The cluster manager is a
- * threadsafe centralized manager for a cluster. Members of a cluster are nodes.
- * A member becomes a node by issuing a connection request to the manager. The
- * manager maintains the set of nodes. Nodes may be disconnected, reconnected,
- * and deleted.
- *
- * Nodes are responsible for sending heartbeats to the manager to indicate their
- * liveliness. A manager may disconnect a node if it does not receive a
- * heartbeat within a configurable time period. A cluster manager instance may
- * be configured with how often to monitor received heartbeats
- * (getHeartbeatMonitoringIntervalSeconds()) and the maximum time that may
- * elapse between node heartbeats before disconnecting the node
- * (getMaxHeartbeatGapSeconds()).
- *
- * Since only a single node may execute isolated processors, the cluster manager
- * maintains the notion of a primary node. The primary node is chosen at cluster
- * startup and retains the role until a user requests a different node to be the
- * primary node.
- *
- * @author unattributed
- */
-public interface ClusterManager extends NodeInformant {
-
- /**
- * Handles a node's heartbeat.
- *
- * @param heartbeat a heartbeat
- *
- */
- void handleHeartbeat(Heartbeat heartbeat);
-
- /**
- * @param statuses the statuses of the nodes
- * @return the set of nodes
- */
- Set<Node> getNodes(Status... statuses);
-
- /**
- * @param nodeId
- * @return returns the node with the given identifier or null if node does
- * not exist
- */
- Node getNode(String nodeId);
-
- /**
- * @param statuses
- * @return the set of node identifiers with the given node status
- */
- Set<NodeIdentifier> getNodeIds(Status... statuses);
-
- /**
- * Deletes the node with the given node identifier. If the given node is the
- * primary node, then a subsequent request may be made to the manager to set
- * a new primary node.
- *
- * @param nodeId the node identifier
- * @param userDn the Distinguished Name of the user requesting the node be
- * deleted from the cluster
- *
- * @throws UnknownNodeException if the node does not exist
- * @throws IllegalNodeDeletionException if the node is not in a disconnected
- * state
- */
- void deleteNode(String nodeId, String userDn) throws UnknownNodeException, IllegalNodeDeletionException;
-
- /**
- * Requests a connection to the cluster.
- *
- * @param request the request
- *
- * @return the response
- */
- ConnectionResponse requestConnection(ConnectionRequest request);
-
- /**
- * Services reconnection requests for a given node. If the node indicates
- * reconnection failure, then the node will be set to disconnected.
- * Otherwise, a reconnection request will be sent to the node, initiating
- * the connection handshake.
- *
- * @param nodeId a node identifier
- * @param userDn the Distinguished Name of the user requesting the
- * reconnection
- *
- * @throws UnknownNodeException if the node does not exist
- * @throws IllegalNodeReconnectionException if the node is not disconnected
- */
- void requestReconnection(String nodeId, String userDn) throws UnknownNodeException, IllegalNodeReconnectionException;
-
- /**
- * Requests the node with the given identifier be disconnected.
- *
- * @param nodeId the node identifier
- * @param userDn the Distinguished Name of the user requesting the
- * disconnection
- *
- * @throws UnknownNodeException if the node does not exist
- * @throws IllegalNodeDisconnectionException if the node cannot be
- * disconnected due to the cluster's state (e.g., node is last connected
- * node or node is primary)
- * @throws UnknownNodeException if the node does not exist
- * @throws IllegalNodeDisconnectionException if the node is not disconnected
- * @throws NodeDisconnectionException if the disconnection failed
- */
- void requestDisconnection(String nodeId, String userDn) throws UnknownNodeException, IllegalNodeDisconnectionException, NodeDisconnectionException;
-
- /**
- * @return the time in seconds to wait between successive executions of
- * heartbeat monitoring
- */
- int getHeartbeatMonitoringIntervalSeconds();
-
- /**
- * @return the maximum time in seconds that is allowed between successive
- * heartbeats of a node before disconnecting the node
- */
- int getMaxHeartbeatGapSeconds();
-
- /**
- * Returns a list of node events for the node with the given identifier. The
- * events will be returned in order of most recent to least recent according
- * to the creation date of the event.
- *
- * @param nodeId the node identifier
- *
- * @return the list of events or an empty list if no node exists with the
- * given identifier
- */
- List<Event> getNodeEvents(final String nodeId);
-
- /**
- * Revokes the primary role from the current primary node and assigns the
- * primary role to given given node ID.
- *
- * If role revocation fails, then the current primary node is set to
- * disconnected while retaining the primary role and no role assignment is
- * performed.
- *
- * If role assignment fails, then the given node is set to disconnected and
- * is given the primary role.
- *
- * @param nodeId the node identifier
- * @param userDn the Distinguished Name of the user requesting that the
- * Primary Node be assigned
- *
- * @throws UnknownNodeException if the node with the given identifier does
- * not exist
- * @throws IneligiblePrimaryNodeException if the node with the given
- * identifier is not eligible to be the primary node
- * @throws PrimaryRoleAssignmentException if the cluster was unable to
- * change the primary role to the requested node
- */
- void setPrimaryNode(String nodeId, String userDn) throws UnknownNodeException, IneligiblePrimaryNodeException, PrimaryRoleAssignmentException;
-
- /**
- * @return the primary node of the cluster or null if no primary node exists
- */
- Node getPrimaryNode();
-
- /**
- * Returns the bulletin repository.
- *
- * @return
- */
- BulletinRepository getBulletinRepository();
-
- /**
- * Returns a {@link ProcessGroupStatus} that represents the status of all
- * nodes with the given {@link Status}es for the given ProcessGroup id, or
- * null if no nodes exist with the given statuses
- *
- * @param groupId
- * @return
- */
- ProcessGroupStatus getProcessGroupStatus(String groupId);
-
- /**
- * Returns a merged representation of the System Diagnostics for all nodes
- * in the cluster
- *
- * @return
- */
- SystemDiagnostics getSystemDiagnostics();
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java
deleted file mode 100644
index 2cf5812..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/*
- * 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.manager;
-
-import org.apache.nifi.cluster.manager.exception.DisconnectedNodeMutableRequestException;
-import org.apache.nifi.cluster.manager.exception.UriConstructionException;
-import org.apache.nifi.cluster.manager.exception.ConnectingNodeMutableRequestException;
-import org.apache.nifi.cluster.manager.exception.NoConnectedNodesException;
-import org.apache.nifi.cluster.manager.exception.NoResponseFromNodesException;
-import org.apache.nifi.cluster.manager.exception.SafeModeMutableRequestException;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-
-import java.net.URI;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * Extends the ClusterManager interface to define how requests issued to the
- * cluster manager are federated to the nodes. Specifically, the HTTP protocol
- * is used for communicating requests to the cluster manager and to the nodes.
- *
- * @author unattributed
- */
-public interface HttpClusterManager extends ClusterManager {
-
- /**
- * Federates the HTTP request to all connected nodes in the cluster. The
- * given URI's host and port will not be used and instead will be adjusted
- * for each node's host and port. The node URIs are guaranteed to be
- * constructed before issuing any requests, so if a UriConstructionException
- * is thrown, then it is guaranteed that no request was issued.
- *
- * @param method the HTTP method (e.g., GET, POST, PUT, DELETE, HEAD)
- * @param uri the base request URI (up to, but not including, the query
- * string)
- * @param parameters the request parameters
- * @param headers the request headers
- *
- * @return the client response
- *
- * @throws NoConnectedNodesException if no nodes are connected as results of
- * the request
- * @throws NoResponseFromNodesException if no response could be obtained
- * @throws UriConstructionException if there was an issue constructing the
- * URIs tailored for each individual node
- * @throws ConnectingNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is connecting to the cluster
- * @throws DisconnectedNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is disconnected from the cluster
- * @throws SafeModeMutableRequestException if the request was a PUT, POST,
- * DELETE and a the cluster is in safe mode
- */
- NodeResponse applyRequest(String method, URI uri, Map<String, List<String>> parameters, Map<String, String> headers)
- throws NoConnectedNodesException, NoResponseFromNodesException, UriConstructionException, ConnectingNodeMutableRequestException,
- DisconnectedNodeMutableRequestException, SafeModeMutableRequestException;
-
- /**
- * Federates the HTTP request to the nodes specified. The given URI's host
- * and port will not be used and instead will be adjusted for each node's
- * host and port. The node URIs are guaranteed to be constructed before
- * issuing any requests, so if a UriConstructionException is thrown, then it
- * is guaranteed that no request was issued.
- *
- * @param method the HTTP method (e.g., GET, POST, PUT, DELETE, HEAD)
- * @param uri the base request URI (up to, but not including, the query
- * string)
- * @param parameters the request parameters
- * @param headers the request headers
- * @param nodeIdentifiers the NodeIdentifier for each node that the request
- * should be replaced to
- *
- * @return the client response
- *
- * @throws NoConnectedNodesException if no nodes are connected as results of
- * the request
- * @throws NoResponseFromNodesException if no response could be obtained
- * @throws UriConstructionException if there was an issue constructing the
- * URIs tailored for each individual node
- * @throws ConnectingNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is connecting to the cluster
- * @throws DisconnectedNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is disconnected from the cluster
- * @throws SafeModeMutableRequestException if the request was a PUT, POST,
- * DELETE and a the cluster is in safe mode
- */
- NodeResponse applyRequest(String method, URI uri, Map<String, List<String>> parameters, Map<String, String> headers,
- Set<NodeIdentifier> nodeIdentifiers)
- throws NoConnectedNodesException, NoResponseFromNodesException, UriConstructionException, ConnectingNodeMutableRequestException,
- DisconnectedNodeMutableRequestException, SafeModeMutableRequestException;
-
- /**
- * Federates the HTTP request to all connected nodes in the cluster. The
- * given URI's host and port will not be used and instead will be adjusted
- * for each node's host and port. The node URIs are guaranteed to be
- * constructed before issuing any requests, so if a UriConstructionException
- * is thrown, then it is guaranteed that no request was issued.
- *
- * @param method the HTTP method (e.g., GET, POST, PUT, DELETE, HEAD)
- * @param uri the base request URI (up to, but not including, the query
- * string)
- * @param entity the HTTP request entity
- * @param headers the request headers
- *
- * @return the client response
- *
- * @throws NoConnectedNodesException if no nodes are connected as results of
- * the request
- * @throws NoResponseFromNodesException if no response could be obtained
- * @throws UriConstructionException if there was an issue constructing the
- * URIs tailored for each individual node
- * @throws ConnectingNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is connecting to the cluster
- * @throws DisconnectedNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is disconnected from the cluster
- * @throws SafeModeMutableRequestException if the request was a PUT, POST,
- * DELETE and a the cluster is in safe mode
- */
- NodeResponse applyRequest(String method, URI uri, Object entity, Map<String, String> headers)
- throws NoConnectedNodesException, NoResponseFromNodesException, UriConstructionException, ConnectingNodeMutableRequestException,
- DisconnectedNodeMutableRequestException, SafeModeMutableRequestException;
-
- /**
- * Federates the HTTP request to the nodes specified. The given URI's host
- * and port will not be used and instead will be adjusted for each node's
- * host and port. The node URIs are guaranteed to be constructed before
- * issuing any requests, so if a UriConstructionException is thrown, then it
- * is guaranteed that no request was issued.
- *
- * @param method the HTTP method (e.g., GET, POST, PUT, DELETE, HEAD)
- * @param uri the base request URI (up to, but not including, the query
- * string)
- * @param entity the HTTP request entity
- * @param headers the request headers
- * @param nodeIdentifiers the NodeIdentifier for each node that the request
- * should be replaced to
- *
- * @return the client response
- *
- * @throws NoConnectedNodesException if no nodes are connected as results of
- * the request
- * @throws NoResponseFromNodesException if no response could be obtained
- * @throws UriConstructionException if there was an issue constructing the
- * URIs tailored for each individual node
- * @throws ConnectingNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is connecting to the cluster
- * @throws DisconnectedNodeMutableRequestException if the request was a PUT,
- * POST, DELETE and a node is disconnected from the cluster
- * @throws SafeModeMutableRequestException if the request was a PUT, POST,
- * DELETE and a the cluster is in safe mode
- */
- NodeResponse applyRequest(String method, URI uri, Object entity, Map<String, String> headers, Set<NodeIdentifier> nodeIdentifiers)
- throws NoConnectedNodesException, NoResponseFromNodesException, UriConstructionException, ConnectingNodeMutableRequestException,
- DisconnectedNodeMutableRequestException, SafeModeMutableRequestException;
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java
deleted file mode 100644
index fb57622..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- * 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.manager;
-
-import org.apache.nifi.cluster.manager.exception.UriConstructionException;
-import java.net.URI;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-
-/**
- * A service for managing the replication of requests to nodes. It is up to the
- * implementing class to decide if requests are sent concurrently or serially.
- *
- * Clients must call start() and stop() to initialize and shutdown the instance.
- * The instance must be started before issuing any replication requests.
- *
- * @author unattributed
- */
-public interface HttpRequestReplicator {
-
- /**
- * Starts the instance for replicating requests. Start may only be called if
- * the instance is not running.
- */
- void start();
-
- /**
- * Stops the instance from replicating requests. Stop may only be called if
- * the instance is running.
- */
- void stop();
-
- /**
- * @return true if the instance is started; false otherwise.
- */
- boolean isRunning();
-
- /**
- * Requests are sent to each node in the cluster. If the request results in
- * an exception, then the NodeResourceResponse will contain the exception.
- *
- * HTTP DELETE and OPTIONS methods must supply an empty parameters map or
- * else and IllegalArgumentException is thrown.
- *
- * @param nodeIds the node identifiers
- * @param method the HTTP method (e.g., GET, POST, PUT, DELETE, HEAD,
- * OPTIONS)
- * @param uri the base request URI (up to, but not including, the query
- * string)
- * @param parameters any request parameters
- * @param headers any HTTP headers
- *
- * @return the set of node responses
- *
- * @throws UriConstructionException if a request for a node failed to be
- * constructed from the given prototype URI. If thrown, it is guaranteed
- * that no request was sent.
- */
- Set<NodeResponse> replicate(Set<NodeIdentifier> nodeIds, String method, URI uri, Map<String, List<String>> parameters, Map<String, String> headers) throws UriConstructionException;
-
- /**
- * Requests are sent to each node in the cluster. If the request results in
- * an exception, then the NodeResourceResponse will contain the exception.
- *
- * HTTP DELETE, GET, HEAD, and OPTIONS methods will throw an
- * IllegalArgumentException if used.
- *
- * @param nodeIds the node identifiers
- * @param method the HTTP method (e.g., POST, PUT)
- * @param uri the base request URI (up to, but not including, the query
- * string)
- * @param entity an entity
- * @param headers any HTTP headers
- *
- * @return the set of node responses
- *
- * @throws UriConstructionException if a request for a node failed to be
- * constructed from the given prototype URI. If thrown, it is guaranteed
- * that no request was sent.
- */
- Set<NodeResponse> replicate(Set<NodeIdentifier> nodeIds, String method, URI uri, Object entity, Map<String, String> headers) throws UriConstructionException;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java
deleted file mode 100644
index 843a666..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.manager;
-
-import java.net.URI;
-import java.util.Map;
-import java.util.Set;
-import org.apache.nifi.cluster.node.Node.Status;
-
-/**
- * Maps a HTTP response to a node status.
- *
- * @author unattributed
- */
-public interface HttpResponseMapper {
-
- /**
- * Maps a HTTP response to a node response and the corresponding node
- * status.
- *
- * @param requestURI the original request URI
- * @param nodeResponses a set of node resource responses
- *
- * @return a map associating the node response to the node status
- */
- Map<NodeResponse, Status> map(URI requestURI, Set<NodeResponse> nodeResponses);
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
deleted file mode 100644
index 3f966e5..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
+++ /dev/null
@@ -1,329 +0,0 @@
-/*
- * 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.manager;
-
-import com.sun.jersey.api.client.ClientResponse;
-import java.io.BufferedInputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.net.URI;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
-
-import javax.ws.rs.HttpMethod;
-import javax.ws.rs.WebApplicationException;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.Response.ResponseBuilder;
-import javax.ws.rs.core.Response.Status;
-import javax.ws.rs.core.StreamingOutput;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.web.api.entity.Entity;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Encapsulates a node's response in regards to receiving a external API
- * request.
- *
- * Both the ClientResponse and (server) Response may be obtained from this
- * instance. The ClientResponse is stored as it is received from the node. This
- * includes the entity input stream. The Response is constructed on demand when
- * mapping a ClientResponse to the Response. The ClientResponse to Response
- * mapping includes copying the ClientResponse's input stream to the Response.
- * Therefore, the getResponse() method should not be called more than once.
- * Furthermore, the method should not be called if the caller has already read
- * the ClientResponse's input stream.
- *
- * If a ClientResponse was unable to be created, then a NodeResponse will store
- * the Throwable, which may be obtained by calling getThrowable().
- *
- * This class overrides hashCode and equals and considers two instances to be
- * equal if they have the equal NodeIdentifiers.
- *
- * @author unattributed
- */
-public class NodeResponse {
-
- private static final Logger logger = LoggerFactory.getLogger(NodeResponse.class);
- private final String httpMethod;
- private final URI requestUri;
- private final ClientResponse clientResponse;
- private final NodeIdentifier nodeId;
- private final Throwable throwable;
- private boolean hasCreatedResponse = false;
- private final Entity updatedEntity;
- private final long requestDurationNanos;
- private final String requestId;
-
- public NodeResponse(final NodeIdentifier nodeId, final String httpMethod, final URI requestUri, final ClientResponse clientResponse, final long requestDurationNanos, final String requestId) {
- if (nodeId == null) {
- throw new IllegalArgumentException("Node identifier may not be null.");
- } else if (StringUtils.isBlank(httpMethod)) {
- throw new IllegalArgumentException("Http method may not be null or empty.");
- } else if (requestUri == null) {
- throw new IllegalArgumentException("Request URI may not be null.");
- } else if (clientResponse == null) {
- throw new IllegalArgumentException("ClientResponse may not be null.");
- }
- this.nodeId = nodeId;
- this.httpMethod = httpMethod;
- this.requestUri = requestUri;
- this.clientResponse = clientResponse;
- this.throwable = null;
- this.updatedEntity = null;
- this.requestDurationNanos = requestDurationNanos;
- this.requestId = requestId;
- }
-
- public NodeResponse(final NodeIdentifier nodeId, final String httpMethod, final URI requestUri, final Throwable throwable) {
- if (nodeId == null) {
- throw new IllegalArgumentException("Node identifier may not be null.");
- } else if (StringUtils.isBlank(httpMethod)) {
- throw new IllegalArgumentException("Http method may not be null or empty.");
- } else if (requestUri == null) {
- throw new IllegalArgumentException("Request URI may not be null.");
- } else if (throwable == null) {
- throw new IllegalArgumentException("Throwable may not be null.");
- }
- this.nodeId = nodeId;
- this.httpMethod = httpMethod;
- this.requestUri = requestUri;
- this.clientResponse = null;
- this.throwable = throwable;
- this.updatedEntity = null;
- this.requestDurationNanos = -1L;
- this.requestId = null;
- }
-
- public NodeResponse(final NodeResponse example, final Entity updatedEntity) {
- Objects.requireNonNull(example, "NodeResponse cannot be null");
- Objects.requireNonNull(updatedEntity, "UpdatedEntity cannot be null");
-
- this.nodeId = example.nodeId;
- this.httpMethod = example.httpMethod;
- this.requestUri = example.requestUri;
- this.clientResponse = example.clientResponse;
- this.throwable = example.throwable;
- this.updatedEntity = updatedEntity;
- this.requestDurationNanos = example.requestDurationNanos;
- this.requestId = null;
- }
-
- public NodeIdentifier getNodeId() {
- return nodeId;
- }
-
- public String getHttpMethod() {
- return httpMethod;
- }
-
- public URI getRequestUri() {
- return requestUri;
- }
-
- /**
- * @return the HTTP response status code
- */
- public int getStatus() {
- if (hasThrowable()) {
- /*
- * since there is a throwable, there is no client input stream to
- * worry about maintaining, so we can call getResponse() method
- */
- return getResponse().getStatus();
- } else {
- /*
- * use client response's status instead of calling getResponse().getStatus()
- * so that we don't read the client's input stream as part of creating
- * the response in the getResponse() method
- */
- return clientResponse.getStatus();
- }
- }
-
- /**
- * Returns true if the response status is 2xx, false otherwise.
- *
- * @return
- */
- public boolean is2xx() {
- final int statusCode = getStatus();
- return (200 <= statusCode && statusCode <= 299);
- }
-
- /**
- * Returns true if the response status is 5xx, false otherwise.
- *
- * @return
- */
- public boolean is5xx() {
- final int statusCode = getStatus();
- return (500 <= statusCode && statusCode <= 599);
- }
-
- /**
- * Returns null if hasThrowable() is true; otherwise the client's response
- * is returned.
- *
- * The ClientResponse's input stream can only be read once.
- *
- * @return the client's response
- */
- public ClientResponse getClientResponse() {
- return clientResponse;
- }
-
- /**
- * Creates a Response by mapping the ClientResponse values to it. Since the
- * ClientResponse's input stream can only be read once, this method should
- * only be called once. Furthermore, the caller should not have already read
- * the ClientResponse's input stream.
- *
- * @return the response
- */
- public Response getResponse() {
- // if the response encapsulates a throwable, then the input stream is never read and the below warning is irrelevant
- if (hasCreatedResponse && !hasThrowable()) {
- logger.warn("ClientResponse's input stream has already been read. The created response will not contain this data.");
- }
- hasCreatedResponse = true;
- return createResponse();
- }
-
- /**
- * Returns the throwable or null if no throwable exists.
- *
- * @return the throwable or null if no throwable exists
- */
- public Throwable getThrowable() {
- return throwable;
- }
-
- /**
- * Returns true if a throwable was thrown and a response was not able to be
- * created; false otherwise.
- *
- * @return true if a throwable was thrown and a response was not able to be
- * created; false otherwise
- */
- public boolean hasThrowable() {
- return getThrowable() != null;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (obj == null) {
- return false;
- }
- if (getClass() != obj.getClass()) {
- return false;
- }
- final NodeResponse other = (NodeResponse) obj;
- if (this.nodeId != other.nodeId && (this.nodeId == null || !this.nodeId.equals(other.nodeId))) {
- return false;
- }
- return true;
- }
-
- @Override
- public int hashCode() {
- int hash = 7;
- hash = 13 * hash + (this.nodeId != null ? this.nodeId.hashCode() : 0);
- return hash;
- }
-
- public long getRequestDuration(final TimeUnit timeUnit) {
- return timeUnit.convert(requestDurationNanos, TimeUnit.NANOSECONDS);
- }
-
- public String getRequestId() {
- return requestId;
- }
-
- private Response createResponse() {
-
- // if no client response was created, then generate a 500 response
- if (hasThrowable()) {
- return Response.status(Status.INTERNAL_SERVER_ERROR).build();
- }
-
- // set the status
- final ResponseBuilder responseBuilder = Response.status(clientResponse.getStatus());
-
- // set the headers
- for (final String key : clientResponse.getHeaders().keySet()) {
- final List<String> values = clientResponse.getHeaders().get(key);
- for (final String value : values) {
-
- if (key.equalsIgnoreCase("transfer-encoding") || key.equalsIgnoreCase("content-length")) {
- /*
- * do not copy the transfer-encoding header (i.e., chunked encoding) or
- * the content-length. Let the outgoing response builder determine it.
- */
- continue;
- } else if (key.equals("X-ClusterContext")) {
- /*
- * do not copy the cluster context to the response because
- * this information is private and should not be sent to
- * the client
- */
- continue;
- }
- responseBuilder.header(key, value);
- }
- }
-
- // head requests must not have a message-body in the response
- if (!HttpMethod.HEAD.equalsIgnoreCase(httpMethod)) {
-
- // set the entity
- if (updatedEntity == null) {
- responseBuilder.entity(new StreamingOutput() {
- @Override
- public void write(final OutputStream output) throws IOException, WebApplicationException {
- BufferedInputStream bis = null;
- try {
- bis = new BufferedInputStream(clientResponse.getEntityInputStream());
- IOUtils.copy(bis, output);
- } finally {
- IOUtils.closeQuietly(bis);
- }
- }
- });
- } else {
- responseBuilder.entity(updatedEntity);
- }
- }
-
- return responseBuilder.build();
- }
-
- @Override
- public String toString() {
- final StringBuilder sb = new StringBuilder();
- sb.append("NodeResponse[nodeUri=").append(nodeId.getApiAddress()).append(":").append(nodeId.getApiPort()).append(",")
- .append("method=").append(httpMethod)
- .append(",URI=").append(requestUri)
- .append(",ResponseCode=").append(getStatus())
- .append(",Duration=").append(TimeUnit.MILLISECONDS.convert(requestDurationNanos, TimeUnit.NANOSECONDS)).append(" ms]");
- return sb.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
deleted file mode 100644
index 49bcd35..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.manager.exception;
-
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-
-/**
- *
- */
-public class BlockedByFirewallException extends ClusterException {
-
- private final NodeIdentifier nodeId;
- private final boolean isExistingNode;
-
- public BlockedByFirewallException(NodeIdentifier nodeId, boolean isExistingNode, String msg, Throwable cause) {
- super(msg, cause);
- this.nodeId = nodeId;
- this.isExistingNode = isExistingNode;
- }
-
- public BlockedByFirewallException(NodeIdentifier nodeId, boolean isExistingNode, Throwable cause) {
- super(cause);
- this.nodeId = nodeId;
- this.isExistingNode = isExistingNode;
- }
-
- public BlockedByFirewallException(NodeIdentifier nodeId, boolean isExistingNode, String msg) {
- super(msg);
- this.nodeId = nodeId;
- this.isExistingNode = isExistingNode;
- }
-
- public BlockedByFirewallException(NodeIdentifier nodeId, boolean isExistingNode) {
- this.nodeId = nodeId;
- this.isExistingNode = isExistingNode;
- }
-
- public NodeIdentifier getNodeId() {
- return nodeId;
- }
-
- public boolean isExistingNode() {
- return isExistingNode;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/300952a9/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/ClusterException.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/ClusterException.java b/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/ClusterException.java
deleted file mode 100644
index 3bf9752..0000000
--- a/nar-bundles/framework-bundle/framework/cluster/src/main/java/org/apache/nifi/cluster/manager/exception/ClusterException.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.manager.exception;
-
-/**
- * The base exception class for cluster related exceptions.
- *
- * @author unattributed
- */
-public class ClusterException extends RuntimeException {
-
- public ClusterException() {
- }
-
- public ClusterException(String msg) {
- super(msg);
- }
-
- public ClusterException(Throwable cause) {
- super(cause);
- }
-
- public ClusterException(String msg, Throwable cause) {
- super(msg, cause);
- }
-}