NIFI-5557: handling expired ticket by rollback and penalization
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/0f55cbfb Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/0f55cbfb Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/0f55cbfb Branch: refs/heads/master Commit: 0f55cbfb9f49087492a333c59b63e146a1444d55 Parents: 2e1005e Author: Endre Zoltan Kovacs <[email protected]> Authored: Tue Aug 28 10:47:59 2018 +0200 Committer: Jeff Storck <[email protected]> Committed: Thu Sep 27 16:24:31 2018 -0400 ---------------------------------------------------------------------- .../apache/nifi/processors/hadoop/PutHDFS.java | 33 ++++++++++++++++++++ 1 file changed, 33 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/0f55cbfb/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java index 91fd204..6e71331 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java @@ -49,6 +49,9 @@ import org.apache.nifi.processor.io.InputStreamCallback; import org.apache.nifi.processor.util.StandardValidators; import org.apache.nifi.stream.io.StreamUtils; import org.apache.nifi.util.StopWatch; +import org.ietf.jgss.GSSException; + +import com.google.common.base.Throwables; import java.io.BufferedInputStream; import java.io.FileNotFoundException; @@ -60,8 +63,11 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.stream.Stream; /** * This processor copies FlowFiles to HDFS. @@ -373,6 +379,17 @@ public class PutHDFS extends AbstractHadoopProcessor { session.transfer(putFlowFile, REL_SUCCESS); + } catch (final IOException e) { + Optional<GSSException> causeOptional = findCause(e, GSSException.class, gsse -> GSSException.NO_CRED == gsse.getMajor()); + if (causeOptional.isPresent()) { + getLogger().warn("An error occurred while connecting to HDFS. " + + "Rolling back session, and penalizing flow file {}", + new Object[] {putFlowFile.getAttribute(CoreAttributes.UUID.key()), causeOptional.get()}); + session.rollback(true); + } else { + getLogger().error("Failed to access HDFS due to {}", new Object[]{e}); + session.transfer(putFlowFile, REL_FAILURE); + } } catch (final Throwable t) { if (tempDotCopyFile != null) { try { @@ -391,6 +408,22 @@ public class PutHDFS extends AbstractHadoopProcessor { }); } + + /** + * Returns an optional with the first throwable in the causal chain that is assignable to the provided cause type, + * and satisfies the provided cause predicate, {@link Optional#empty()} otherwise. + * @param t The throwable to inspect for the cause. + * @return + */ + private <T extends Throwable> Optional<T> findCause(Throwable t, Class<T> expectedCauseType, Predicate<T> causePredicate) { + Stream<Throwable> causalChain = Throwables.getCausalChain(t).stream(); + return causalChain + .filter(expectedCauseType::isInstance) + .map(expectedCauseType::cast) + .filter(causePredicate) + .findFirst(); + } + protected void changeOwner(final ProcessContext context, final FileSystem hdfs, final Path name, final FlowFile flowFile) { try { // Change owner and group of file if configured to do so
