You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/01/21 15:18:19 UTC
nifi git commit: NIFI-259: Update ZooKeeperStateServer so that the
thread starting it doesn't block indefinitely, as this was causing NiFi to
not shut down properly; applied patch for NIFI-1415
Repository: nifi
Updated Branches:
refs/heads/NIFI-259 593f1288d -> d696391f7
NIFI-259: Update ZooKeeperStateServer so that the thread starting it doesn't block indefinitely, as this was causing NiFi to not shut down properly; applied patch for NIFI-1415
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/d696391f
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/d696391f
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/d696391f
Branch: refs/heads/NIFI-259
Commit: d696391f76cb14bf4bdaedb4e88d129aa2e54dc2
Parents: 593f128
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Jan 20 11:52:41 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Jan 20 11:52:41 2016 -0500
----------------------------------------------------------------------
.../state/server/ZooKeeperStateServer.java | 48 +++++++++++++++++---
.../nifi/processors/standard/FetchSFTP.java | 18 --------
2 files changed, 42 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/d696391f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/server/ZooKeeperStateServer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/server/ZooKeeperStateServer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/server/ZooKeeperStateServer.java
index 1a646b0..e65c375 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/server/ZooKeeperStateServer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/server/ZooKeeperStateServer.java
@@ -25,8 +25,11 @@ import java.io.InputStream;
import java.util.Properties;
import org.apache.nifi.util.NiFiProperties;
+import org.apache.zookeeper.server.ServerCnxnFactory;
import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.server.ZooKeeperServerMain;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
import org.slf4j.Logger;
@@ -35,8 +38,12 @@ import org.slf4j.LoggerFactory;
public class ZooKeeperStateServer extends ZooKeeperServerMain {
private static final Logger logger = LoggerFactory.getLogger(ZooKeeperStateServer.class);
- private QuorumPeerConfig quorumPeerConfig;
- private boolean started = false;
+ private final QuorumPeerConfig quorumPeerConfig;
+ private volatile boolean started = false;
+
+ private ServerCnxnFactory connectionFactory;
+ private FileTxnSnapLog transactionLog;
+ private ZooKeeperServer embeddedZkServer;
private ZooKeeperStateServer(final Properties zkProperties) throws IOException, ConfigException {
quorumPeerConfig = new QuorumPeerConfig();
@@ -46,11 +53,25 @@ public class ZooKeeperStateServer extends ZooKeeperServerMain {
public synchronized void start() throws IOException {
logger.info("Starting Embedded ZooKeeper Server");
- final ServerConfig serverConfig = new ServerConfig();
- serverConfig.readFrom(quorumPeerConfig);
+ final ServerConfig config = new ServerConfig();
+ config.readFrom(quorumPeerConfig);
try {
- runFromConfig(serverConfig);
started = true;
+
+ transactionLog = new FileTxnSnapLog(new File(config.getDataLogDir()), new File(config.getDataDir()));
+
+ embeddedZkServer = new ZooKeeperServer();
+ embeddedZkServer.setTxnLogFactory(transactionLog);
+ embeddedZkServer.setTickTime(config.getTickTime());
+ embeddedZkServer.setMinSessionTimeout(config.getMinSessionTimeout());
+ embeddedZkServer.setMaxSessionTimeout(config.getMaxSessionTimeout());
+
+ connectionFactory = ServerCnxnFactory.createFactory();
+ connectionFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns());
+ connectionFactory.startup(embeddedZkServer);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("Embedded ZooKeeper Server interrupted", e);
} catch (final IOException ioe) {
throw new IOException("Failed to start embedded ZooKeeper Server", ioe);
} catch (final Exception e) {
@@ -62,7 +83,22 @@ public class ZooKeeperStateServer extends ZooKeeperServerMain {
public synchronized void shutdown() {
if (started) {
started = false;
- super.shutdown();
+
+ if (transactionLog != null) {
+ try {
+ transactionLog.close();
+ } catch (final IOException ioe) {
+ logger.warn("Failed to close Transaction Log", ioe);
+ }
+ }
+
+ if (connectionFactory != null) {
+ connectionFactory.shutdown();
+ }
+
+ if (embeddedZkServer != null && embeddedZkServer.isRunning()) {
+ embeddedZkServer.shutdown();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/d696391f/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
index 8379987..dd6d880 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
@@ -18,8 +18,6 @@
package org.apache.nifi.processors.standard;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -30,8 +28,6 @@ import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processors.standard.util.FileTransfer;
import org.apache.nifi.processors.standard.util.SFTPTransfer;
@@ -73,20 +69,6 @@ public class FetchSFTP extends FetchFileTransfer {
}
@Override
- protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
- if (!validationContext.getProperty(SFTPTransfer.PASSWORD).isSet() && !(validationContext.getProperty(SFTPTransfer.PRIVATE_KEY_PASSPHRASE).isSet()
- && validationContext.getProperty(SFTPTransfer.PRIVATE_KEY_PATH).isSet())) {
- return Collections.singleton(new ValidationResult.Builder()
- .subject("Password")
- .valid(false)
- .explanation("Must set either password or Private Key Path & Passphrase")
- .build());
- }
-
- return Collections.emptyList();
- }
-
- @Override
protected FileTransfer createFileTransfer(final ProcessContext context) {
return new SFTPTransfer(context, getLogger());
}