NIFI-1037 Ported processor for HDFS' inotify events to 0.x.

nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/inotify/EventAttributes.java

@@ -0,0 +1,23 @@
+/*
+ * 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.processors.hadoop.inotify;
+
+final class EventAttributes {
+    static final String EVENT_PATH = "hdfs.inotify.event.path";
+    static final String EVENT_TYPE = "hdfs.inotify.event.type";
+    static final String MIME_TYPE = "mime.type";
+}

nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/inotify/EventTypeValidator.java

@@ -0,0 +1,57 @@
+/*
+ * 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.processors.hadoop.inotify;
+
+import org.apache.hadoop.hdfs.inotify.Event;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.Validator;
+
+import java.util.ArrayList;
+import java.util.List;
+
+class EventTypeValidator implements Validator {
+
+    @Override
+    public ValidationResult validate(String subject, String input, ValidationContext context) {
+        final String explanation = isValidEventType(input);
+        return new ValidationResult.Builder()
+                .subject(subject)
+                .input(input)
+                .valid(explanation == null)
+                .explanation(explanation)
+                .build();
+    }
+
+    private String isValidEventType(String input) {
+        if (input != null && !"".equals(input.trim())) {
+            final String[] events = input.split(",");
+            final List<String> invalid = new ArrayList<>();
+            for (String event : events) {
+                try {
+                    Event.EventType.valueOf(event.trim().toUpperCase());
+                } catch (IllegalArgumentException e) {
+                    invalid.add(event.trim());
+                }
+            }
+
+            return invalid.isEmpty() ? null : "The following are not valid event types: " + invalid;
+        }
+
+        return "Empty event types are not allowed.";
+    }
+}

nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/inotify/NotificationEventPathFilter.java

@@ -0,0 +1,38 @@
+/*
+ * 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.processors.hadoop.inotify;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+
+import java.util.regex.Pattern;
+
+class NotificationEventPathFilter implements PathFilter {
+    private final Pattern watchDirectory;
+    private final boolean toIgnoreHiddenFiles;
+
+    NotificationEventPathFilter(Pattern watchDirectory, boolean toIgnoreHiddenFiles) {
+        this.watchDirectory = watchDirectory;
+        this.toIgnoreHiddenFiles = toIgnoreHiddenFiles;
+    }
+
+    @Override
+    public boolean accept(Path path) {
+        return !(path == null || (toIgnoreHiddenFiles && path.getName().startsWith(".")))
+                && watchDirectory.matcher(path.toString()).matches();
+    }
+}

nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor

@@ -16,5 +16,6 @@ org.apache.nifi.processors.hadoop.CreateHadoopSequenceFile
 org.apache.nifi.processors.hadoop.FetchHDFS
 org.apache.nifi.processors.hadoop.GetHDFS
 org.apache.nifi.processors.hadoop.GetHDFSSequenceFile
+org.apache.nifi.processors.hadoop.inotify.GetHDFSEvents
 org.apache.nifi.processors.hadoop.ListHDFS
 org.apache.nifi.processors.hadoop.PutHDFS

nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/inotify/TestEventTypeValidator.java

@@ -0,0 +1,98 @@
+/*
+ * 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.processors.hadoop.inotify;
+
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestEventTypeValidator {
+    ValidationContext context;
+    EventTypeValidator eventTypeValidator;
+
+    @Before
+    public void setUp() throws Exception {
+        context = Mockito.mock(ValidationContext.class);
+        eventTypeValidator = new EventTypeValidator();
+    }
+
+    @Test
+    public void nullInputShouldProperlyFail() throws Exception {
+        String subject = "subject";
+        String input = null;
+        ValidationResult result = eventTypeValidator.validate(subject, input, context);
+
+        assertEquals("subject", result.getSubject());
+        assertEquals(null, result.getInput());
+        assertEquals("Empty event types are not allowed.", result.getExplanation());
+        assertFalse(result.isValid());
+    }
+
+    @Test
+    public void emptyInputShouldProperlyFail() throws Exception {
+        String subject = "subject";
+        String input = "";
+        ValidationResult result = eventTypeValidator.validate(subject, input, context);
+
+        assertEquals("subject", result.getSubject());
+        assertEquals("", result.getInput());
+        assertEquals("Empty event types are not allowed.", result.getExplanation());
+        assertFalse(result.isValid());
+    }
+
+    @Test
+    public void validEventTypesShouldProperlyValidate() throws Exception {
+        String input = "  append, Create, CLOSE";
+        String subject = "subject";
+        ValidationResult result = eventTypeValidator.validate(subject, input, context);
+
+        assertEquals("subject", result.getSubject());
+        assertEquals("  append, Create, CLOSE", result.getInput());
+        assertEquals("", result.getExplanation());
+        assertTrue(result.isValid());
+    }
+
+    @Test
+    public void inputWithInvalidEventTypeShouldProperlyDisplayEventsInExplanation() throws Exception {
+        String subject = "subject";
+        String input = "append, CREATE, cllose, rename, metadata, unlink";
+        ValidationResult result = eventTypeValidator.validate(subject, input, context);
+
+        assertEquals("subject", result.getSubject());
+        assertEquals("append, CREATE, cllose, rename, metadata, unlink", result.getInput());
+        assertEquals("The following are not valid event types: [cllose]", result.getExplanation());
+        assertFalse(result.isValid());
+    }
+
+    @Test
+    public void inputWithMultipleInvalidEventTypeShouldProperlyDisplayEventsInExplanation() throws Exception {
+        String subject = "subject";
+        String input = "append, CREATE, cllose, rename, metadata, unlink, unllink";
+        ValidationResult result = eventTypeValidator.validate(subject, input, context);
+
+        assertEquals("subject", result.getSubject());
+        assertEquals("append, CREATE, cllose, rename, metadata, unlink, unllink", result.getInput());
+        assertEquals("The following are not valid event types: [cllose, unllink]", result.getExplanation());
+        assertFalse(result.isValid());
+    }
+}