You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2016/06/03 19:30:13 UTC
[12/18] nifi git commit: NIFI-1897: - Refactoring to allow requests
to be replicated from a node to other nodes - Renaming cluster node
connection/read timeout properties. - Renaming NCM DN to Cluster Coordinator
DN. - Fixing default values in properties
http://git-wip-us.apache.org/repos/asf/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java
deleted file mode 100644
index 7c99c93..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowDaoImpl.java
+++ /dev/null
@@ -1,595 +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.
- *
- */
-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 CONTROLLER_SERVICES_FILENAME = "controller-services.xml";
- public static final String REPORTING_TASKS_FILENAME = "reporting-tasks.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();
-
- writeDataFlow(stateFile, new ClusterDataFlow(null, null, new byte[0], new byte[0]), new ClusterMetadata());
-
- return stateFile;
- }
-
- private byte[] getEmptyFlowBytes() throws IOException {
- try {
- final DocumentBuilderFactory docBuilderFactory = DocumentBuilderFactory.newInstance();
- docBuilderFactory.setNamespaceAware(true);
- final DocumentBuilder docBuilder = docBuilderFactory.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];
- byte[] controllerServiceBytes = new byte[0];
- byte[] reportingTaskBytes = 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;
- case CONTROLLER_SERVICES_FILENAME:
- controllerServiceBytes = new byte[(int) tarEntry.getSize()];
- StreamUtils.fillBuffer(tarIn, controllerServiceBytes, true);
- break;
- case REPORTING_TASKS_FILENAME:
- reportingTaskBytes = new byte[(int) tarEntry.getSize()];
- StreamUtils.fillBuffer(tarIn, reportingTaskBytes, 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, snippetBytes);
- dataFlow.setAutoStartProcessors(autoStart);
-
- return new ClusterDataFlow(dataFlow, clusterMetadata == null ? null : clusterMetadata.getPrimaryNodeId(), controllerServiceBytes, reportingTaskBytes);
- }
-
- private void writeDataFlow(final File file, final ClusterDataFlow clusterDataFlow) throws IOException, JAXBException {
- // setup the cluster metadata
- final ClusterMetadata clusterMetadata = new ClusterMetadata();
- clusterMetadata.setPrimaryNodeId(clusterDataFlow.getPrimaryNodeId());
-
- // write to disk
- writeDataFlow(file, clusterDataFlow, 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 ClusterDataFlow clusterDataFlow, final ClusterMetadata clusterMetadata) throws IOException, JAXBException {
-
- try (final OutputStream fos = new FileOutputStream(file);
- final TarArchiveOutputStream tarOut = new TarArchiveOutputStream(new BufferedOutputStream(fos))) {
-
- final DataFlow dataFlow = clusterDataFlow.getDataFlow();
- if (dataFlow == null) {
- writeTarEntry(tarOut, FLOW_XML_FILENAME, getEmptyFlowBytes());
- writeTarEntry(tarOut, SNIPPETS_FILENAME, new byte[0]);
- } else {
- writeTarEntry(tarOut, FLOW_XML_FILENAME, dataFlow.getFlow());
- writeTarEntry(tarOut, SNIPPETS_FILENAME, dataFlow.getSnippets());
- }
- writeTarEntry(tarOut, CONTROLLER_SERVICES_FILENAME, clusterDataFlow.getControllerServices());
- writeTarEntry(tarOut, REPORTING_TASKS_FILENAME, clusterDataFlow.getReportingTasks());
-
- 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/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java
deleted file mode 100644
index 22450da..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/impl/DataFlowManagementServiceImpl.java
+++ /dev/null
@@ -1,403 +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.
- *
- */
-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;
- final byte[] controllerServiceBytes;
- final byte[] reportingTaskBytes;
- if (existingClusterDataFlow == null) {
- dataFlow = null;
- controllerServiceBytes = new byte[0];
- reportingTaskBytes = new byte[0];
- } else {
- dataFlow = existingClusterDataFlow.getDataFlow();
- controllerServiceBytes = existingClusterDataFlow.getControllerServices();
- reportingTaskBytes = existingClusterDataFlow.getReportingTasks();
- }
-
- flowDao.saveDataFlow(new ClusterDataFlow(dataFlow, nodeId, controllerServiceBytes, reportingTaskBytes));
- } finally {
- resourceLock.unlock("updatePrimaryNode");
- }
- }
-
- @Override
- public void updateControllerServices(final byte[] controllerServiceBytes) throws DaoException {
- resourceLock.lock();
- try {
- final ClusterDataFlow existingClusterDataFlow = flowDao.loadDataFlow();
-
- final StandardDataFlow dataFlow;
- final byte[] reportingTaskBytes;
- final NodeIdentifier nodeId;
- if (existingClusterDataFlow == null) {
- dataFlow = null;
- nodeId = null;
- reportingTaskBytes = new byte[0];
- } else {
- dataFlow = existingClusterDataFlow.getDataFlow();
- nodeId = existingClusterDataFlow.getPrimaryNodeId();
- reportingTaskBytes = existingClusterDataFlow.getReportingTasks();
- }
-
- flowDao.saveDataFlow(new ClusterDataFlow(dataFlow, nodeId, controllerServiceBytes, reportingTaskBytes));
- } finally {
- resourceLock.unlock("updateControllerServices");
- }
- }
-
- @Override
- public void updateReportingTasks(final byte[] reportingTaskBytes) throws DaoException {
- resourceLock.lock();
- try {
- final ClusterDataFlow existingClusterDataFlow = flowDao.loadDataFlow();
-
- final StandardDataFlow dataFlow;
- final byte[] controllerServiceBytes;
- final NodeIdentifier nodeId;
- if (existingClusterDataFlow == null) {
- dataFlow = null;
- nodeId = null;
- controllerServiceBytes = null;
- } else {
- dataFlow = existingClusterDataFlow.getDataFlow();
- nodeId = existingClusterDataFlow.getPrimaryNodeId();
- controllerServiceBytes = existingClusterDataFlow.getControllerServices();
- }
-
- flowDao.saveDataFlow(new ClusterDataFlow(dataFlow, nodeId, controllerServiceBytes, reportingTaskBytes));
- } finally {
- resourceLock.unlock("updateControllerServices");
- }
- }
-
- @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, new byte[0], new byte[0]);
- } else {
- currentClusterDataFlow = new ClusterDataFlow(dataFlow, existingClusterDataFlow.getPrimaryNodeId(),
- existingClusterDataFlow.getControllerServices(), existingClusterDataFlow.getReportingTasks());
- }
- 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/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
deleted file mode 100644
index de3c23e..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
+++ /dev/null
@@ -1,142 +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.util.List;
-import java.util.Set;
-
-import org.apache.nifi.cluster.event.Event;
-import org.apache.nifi.cluster.manager.exception.IllegalNodeDeletionException;
-import org.apache.nifi.cluster.manager.exception.IllegalNodeDisconnectionException;
-import org.apache.nifi.cluster.manager.exception.IllegalNodeReconnectionException;
-import org.apache.nifi.cluster.manager.exception.NodeDisconnectionException;
-import org.apache.nifi.cluster.manager.exception.UnknownNodeException;
-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.NodeIdentifier;
-import org.apache.nifi.remote.cluster.NodeInformant;
-import org.apache.nifi.reporting.BulletinRepository;
-
-/**
- * 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.
- *
- */
-public interface ClusterManager extends NodeInformant {
-
- /**
- * @param statuses the statuses of the nodes
- * @return the set of nodes
- */
- Set<Node> getNodes(Status... statuses);
-
- /**
- * @param nodeId node identifier
- * @return returns the node with the given identifier or null if node does not exist
- */
- Node getNode(String nodeId);
-
- /**
- * @param statuses 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);
-
- /**
- * @return the primary node of the cluster or null if no primary node exists
- */
- Node getPrimaryNode();
-
- /**
- * @return the bulletin repository
- */
- BulletinRepository getBulletinRepository();
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java
deleted file mode 100644
index 0f9872f..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpClusterManager.java
+++ /dev/null
@@ -1,129 +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.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.nifi.cluster.manager.exception.ConnectingNodeMutableRequestException;
-import org.apache.nifi.cluster.manager.exception.DisconnectedNodeMutableRequestException;
-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.manager.exception.UriConstructionException;
-import org.apache.nifi.cluster.protocol.NodeIdentifier;
-
-/**
- * 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.
- *
- */
-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/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java
deleted file mode 100644
index 3a3d976..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpRequestReplicator.java
+++ /dev/null
@@ -1,84 +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.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.nifi.cluster.manager.exception.UriConstructionException;
-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.
- *
- */
-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/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java
deleted file mode 100644
index 8f56bbb..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/HttpResponseMapper.java
+++ /dev/null
@@ -1,41 +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.
- *
- */
-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/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
index ec365aa..308652e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/NodeResponse.java
@@ -131,9 +131,6 @@ public class NodeResponse {
return requestUri;
}
- /**
- * @return the HTTP response status code
- */
public int getStatus() {
if (hasThrowable()) {
/*
@@ -151,49 +148,25 @@ public class NodeResponse {
}
}
- /**
- * @return true if the response status is 2xx, false otherwise.
- */
public boolean is2xx() {
final int statusCode = getStatus();
return (200 <= statusCode && statusCode <= 299);
}
- /**
- * @return true if the response status is 5xx, false otherwise.
- */
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;
}
- /**
- * If this node response has been merged returns the updated entity, otherwise null. Also returns null if hasThrowable() is true. The intent of this method is to support getting the response
- * entity when it was already consumed during the merge operation. In this case the client response rom getClientResponse() will not support a getEntity(...) or getEntityInputStream() call.
- *
- * @return If this node response has been merged returns the updated entity, otherwise null. Also returns null if hasThrowable() is true
- */
+
public Entity getUpdatedEntity() {
return updatedEntity;
}
- /**
- * 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()) {
@@ -203,20 +176,11 @@ public class NodeResponse {
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;
}
@@ -313,10 +277,10 @@ public class NodeResponse {
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]");
+ .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/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
index 49bcd35..c13e9d9 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/BlockedByFirewallException.java
@@ -22,7 +22,7 @@ import org.apache.nifi.cluster.protocol.NodeIdentifier;
*
*/
public class BlockedByFirewallException extends ClusterException {
-
+ private static final long serialVersionUID = 1L;
private final NodeIdentifier nodeId;
private final boolean isExistingNode;
http://git-wip-us.apache.org/repos/asf/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IneligiblePrimaryNodeException.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IneligiblePrimaryNodeException.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IneligiblePrimaryNodeException.java
deleted file mode 100644
index 26b22f8..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IneligiblePrimaryNodeException.java
+++ /dev/null
@@ -1,39 +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;
-
-/**
- * Represents the exceptional case when the primary role cannot be assigned to a node because the node is ineligible for the role.
- *
- */
-public class IneligiblePrimaryNodeException extends IllegalClusterStateException {
-
- public IneligiblePrimaryNodeException() {
- }
-
- public IneligiblePrimaryNodeException(String msg) {
- super(msg);
- }
-
- public IneligiblePrimaryNodeException(Throwable cause) {
- super(cause);
- }
-
- public IneligiblePrimaryNodeException(String msg, Throwable cause) {
- super(msg, cause);
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/PrimaryRoleAssignmentException.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/PrimaryRoleAssignmentException.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/PrimaryRoleAssignmentException.java
deleted file mode 100644
index 5750382..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/PrimaryRoleAssignmentException.java
+++ /dev/null
@@ -1,39 +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;
-
-/**
- * Represents the exceptional case when the cluster is unable to update the primary role of a node.
- *
- */
-public class PrimaryRoleAssignmentException extends IllegalClusterStateException {
-
- public PrimaryRoleAssignmentException() {
- }
-
- public PrimaryRoleAssignmentException(String msg) {
- super(msg);
- }
-
- public PrimaryRoleAssignmentException(Throwable cause) {
- super(cause);
- }
-
- public PrimaryRoleAssignmentException(String msg, Throwable cause) {
- super(msg, cause);
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/SafeModeMutableRequestException.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/SafeModeMutableRequestException.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/SafeModeMutableRequestException.java
deleted file mode 100644
index 05d2435..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/SafeModeMutableRequestException.java
+++ /dev/null
@@ -1,39 +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;
-
-/**
- * Represents the exceptional case when a HTTP request that may change a node's dataflow is to be replicated while the cluster is in safe mode.
- *
- */
-public class SafeModeMutableRequestException extends MutableRequestException {
-
- public SafeModeMutableRequestException() {
- }
-
- public SafeModeMutableRequestException(String msg) {
- super(msg);
- }
-
- public SafeModeMutableRequestException(Throwable cause) {
- super(cause);
- }
-
- public SafeModeMutableRequestException(String msg, Throwable cause) {
- super(msg, cause);
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/df0e4e79/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
deleted file mode 100644
index 3966a31..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
+++ /dev/null
@@ -1,146 +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.impl;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.nifi.action.Action;
-import org.apache.nifi.admin.service.AuditService;
-import org.apache.nifi.controller.status.ProcessGroupStatus;
-import org.apache.nifi.events.EventReporter;
-import org.apache.nifi.history.History;
-import org.apache.nifi.provenance.ProvenanceEventBuilder;
-import org.apache.nifi.provenance.ProvenanceEventRecord;
-import org.apache.nifi.provenance.ProvenanceEventRepository;
-import org.apache.nifi.provenance.lineage.ComputeLineageSubmission;
-import org.apache.nifi.provenance.search.Query;
-import org.apache.nifi.provenance.search.QuerySubmission;
-import org.apache.nifi.provenance.search.SearchableField;
-import org.apache.nifi.reporting.EventAccess;
-
-public class ClusteredEventAccess implements EventAccess {
-
- private final WebClusterManager clusterManager;
- private final AuditService auditService;
-
- public ClusteredEventAccess(final WebClusterManager clusterManager, final AuditService auditService) {
- this.clusterManager = clusterManager;
- this.auditService = auditService;
- }
-
- @Override
- public ProcessGroupStatus getControllerStatus() {
- return new ProcessGroupStatus();
- }
-
- @Override
- public List<ProvenanceEventRecord> getProvenanceEvents(long arg0, int arg1) throws IOException {
- return new ArrayList<>();
- }
-
- @Override
- public ProvenanceEventRepository getProvenanceRepository() {
- // NCM doesn't have provenance events, because it doesn't process FlowFiles.
- // So we just use a Provenance Event Repository that does nothing.
- return new ProvenanceEventRepository() {
- @Override
- public void close() throws IOException {
- }
-
- @Override
- public ProvenanceEventRecord getEvent(long eventId) throws IOException {
- return null;
- }
-
- @Override
- public List<ProvenanceEventRecord> getEvents(long startEventId, int maxEvents) throws IOException {
- return new ArrayList<>();
- }
-
- @Override
- public Long getMaxEventId() {
- return null;
- }
-
- @Override
- public List<SearchableField> getSearchableAttributes() {
- return new ArrayList<>();
- }
-
- @Override
- public List<SearchableField> getSearchableFields() {
- return new ArrayList<>();
- }
-
- @Override
- public void registerEvent(final ProvenanceEventRecord event) {
- }
-
- @Override
- public void registerEvents(final Iterable<ProvenanceEventRecord> events) {
- }
-
- @Override
- public ComputeLineageSubmission retrieveLineageSubmission(final String submissionId) {
- return null;
- }
-
- @Override
- public QuerySubmission retrieveQuerySubmission(final String submissionId) {
- return null;
- }
-
- @Override
- public ComputeLineageSubmission submitExpandChildren(final long eventId) {
- return null;
- }
-
- @Override
- public ComputeLineageSubmission submitExpandParents(final long eventId) {
- return null;
- }
-
- @Override
- public ComputeLineageSubmission submitLineageComputation(final String flowFileUuid) {
- return null;
- }
-
- @Override
- public QuerySubmission submitQuery(final Query query) {
- return null;
- }
-
- @Override
- public ProvenanceEventBuilder eventBuilder() {
- return null;
- }
-
- @Override
- public void initialize(EventReporter eventReporter) throws IOException {
-
- }
- };
- }
-
- @Override
- public List<Action> getFlowChanges(int firstActionId, int maxActions) {
- final History history = auditService.getActions(firstActionId, maxActions);
- return new ArrayList<>(history.getActions());
- }
-}