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 2015/08/31 01:48:43 UTC
[3/5] nifi git commit: NIFI-902: Ensure that if we get an IOException
during rollover of WAL, we are able to recover
NIFI-902: Ensure that if we get an IOException during rollover of WAL, we are able to recover
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/5de37f63
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/5de37f63
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/5de37f63
Branch: refs/heads/master
Commit: 5de37f63d920e63d001eaa479e5f9164923ec78a
Parents: 76b5b38
Author: Mark Payne <ma...@hotmail.com>
Authored: Thu Aug 27 16:25:24 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Aug 28 10:04:58 2015 -0400
----------------------------------------------------------------------
.../org/wali/MinimalLockingWriteAheadLog.java | 40 +++++++++++++++-----
.../tasks/ContinuallyRunProcessorTask.java | 13 ++++++-
.../nifi/controller/tasks/ExpireFlowFiles.java | 9 ++++-
3 files changed, 51 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/5de37f63/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java b/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
index 5a7656d..f354f69 100644
--- a/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
+++ b/nifi-commons/nifi-write-ahead-log/src/main/java/org/wali/MinimalLockingWriteAheadLog.java
@@ -722,19 +722,41 @@ public final class MinimalLockingWriteAheadLog<T> implements WriteAheadRepositor
try {
final DataOutputStream out = dataOut;
if (out != null) {
- out.close();
+ try {
+ out.close();
+ } catch (final IOException ioe) {
+ dataOut = null;
+ fileOut = null;
+
+ blackList();
+ throw ioe;
+ }
}
final Path editPath = getNewEditPath();
final FileOutputStream fos = new FileOutputStream(editPath.toFile());
- final DataOutputStream outStream = new DataOutputStream(new BufferedOutputStream(fos));
- outStream.writeUTF(MinimalLockingWriteAheadLog.class.getName());
- outStream.writeInt(writeAheadLogVersion);
- outStream.writeUTF(serde.getClass().getName());
- outStream.writeInt(serde.getVersion());
- outStream.flush();
- dataOut = outStream;
- fileOut = fos;
+ try {
+ final DataOutputStream outStream = new DataOutputStream(new BufferedOutputStream(fos));
+ outStream.writeUTF(MinimalLockingWriteAheadLog.class.getName());
+ outStream.writeInt(writeAheadLogVersion);
+ outStream.writeUTF(serde.getClass().getName());
+ outStream.writeInt(serde.getVersion());
+ outStream.flush();
+ dataOut = outStream;
+ fileOut = fos;
+ } catch (final IOException ioe) {
+ logger.error("Failed to create new journal for {} due to {}", new Object[] {this, ioe.toString()}, ioe);
+ try {
+ fos.close();
+ } catch (final IOException innerIOE) {
+ }
+
+ dataOut = null;
+ fileOut = null;
+ blackList();
+
+ throw ioe;
+ }
currentJournalFilename = editPath.toFile().getName();
http://git-wip-us.apache.org/repos/asf/nifi/blob/5de37f63/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
index 7a85274..d4d595b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
@@ -159,7 +159,18 @@ public class ContinuallyRunProcessorTask implements Callable<Boolean> {
} finally {
try {
if (batch) {
- rawSession.commit();
+ try {
+ rawSession.commit();
+ } catch (final Exception e) {
+ final ProcessorLog procLog = new SimpleProcessLogger(procNode.getIdentifier(), procNode.getProcessor());
+ procLog.error("Failed to commit session {} due to {}; rolling back", new Object[] { rawSession, e.toString() }, e);
+
+ try {
+ rawSession.rollback(true);
+ } catch (final Exception e1) {
+ procLog.error("Failed to roll back session {} due to {}", new Object[] { rawSession, e.toString() }, e);
+ }
+ }
}
final long processingNanos = System.nanoTime() - startNanos;
http://git-wip-us.apache.org/repos/asf/nifi/blob/5de37f63/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ExpireFlowFiles.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ExpireFlowFiles.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ExpireFlowFiles.java
index 7d8bcec..159f58f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ExpireFlowFiles.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ExpireFlowFiles.java
@@ -32,11 +32,14 @@ import org.apache.nifi.controller.scheduling.ProcessContextFactory;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.util.FormatUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This task runs through all Connectable Components and goes through its incoming queues, polling for FlowFiles and accepting none. This causes the desired side effect of expiring old FlowFiles.
*/
public class ExpireFlowFiles implements Runnable {
+ private static final Logger logger = LoggerFactory.getLogger(ExpireFlowFiles.class);
private final FlowController flowController;
private final ProcessContextFactory contextFactory;
@@ -49,7 +52,11 @@ public class ExpireFlowFiles implements Runnable {
@Override
public void run() {
final ProcessGroup rootGroup = flowController.getGroup(flowController.getRootGroupId());
- expireFlowFiles(rootGroup);
+ try {
+ expireFlowFiles(rootGroup);
+ } catch (final Exception e) {
+ logger.error("Failed to expire FlowFiles due to {}", e.toString(), e);
+ }
}
private StandardProcessSession createSession(final Connectable connectable) {