From e1663995328055760866bb9f5e9a1fc0bd498796 Mon Sep 17 00:00:00 2001 From: Koji Kawamura Date: Fri, 2 Dec 2016 11:23:33 +0900 Subject: [PATCH] NIFI-3141: Fixed TailFile ArrayIndexOutOfBounds. - Added unit test cases to simulate NiFi version update which fails without this fix. - Added state object migration code, add file.0. prefix to state keys, and add length from stored position. --- .../nifi/processors/standard/TailFile.java | 28 ++++++ .../processors/standard/TestTailFile.java | 94 +++++++++++++++++++ 2 files changed, 122 insertions(+) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java index 3553ce8fcb8c..c5fcefbc2a40 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java @@ -345,6 +345,25 @@ public void recoverState(final ProcessContext context) throws IOException { Map statesMap = stateMap.toMap(); + if (statesMap.containsKey(TailFileState.StateKeys.FILENAME) + && !statesMap.keySet().stream().anyMatch(key -> key.startsWith(MAP_PREFIX))) { + // If statesMap contains "filename" key without "file.0." prefix, + // and there's no key with "file." prefix, then + // it indicates that the statesMap is created with earlier version of NiFi. + // In this case, we need to migrate the state by adding prefix indexed with 0. + final Map migratedStatesMap = new HashMap<>(statesMap.size()); + for (String key : statesMap.keySet()) { + migratedStatesMap.put(MAP_PREFIX + "0." + key, statesMap.get(key)); + } + + // LENGTH is added from NiFi 1.1.0. Set the value with using the last position so that we can use existing state + // to avoid sending duplicated log data after updating NiFi. + migratedStatesMap.put(MAP_PREFIX + "0." + TailFileState.StateKeys.LENGTH, statesMap.get(TailFileState.StateKeys.POSITION)); + statesMap = Collections.unmodifiableMap(migratedStatesMap); + + getLogger().info("statesMap has been migrated. {}", new Object[]{migratedStatesMap}); + } + initStates(filesToTail, statesMap, false); recoverState(context, filesToTail, statesMap); } @@ -931,6 +950,15 @@ private void persistState(final Map state, final ProcessContext Map updatedState = new HashMap(); for(String key : oldState.toMap().keySet()) { + // These states are stored by older version of NiFi, and won't be used anymore. + // New states have 'file..' prefix. + if (TailFileState.StateKeys.CHECKSUM.equals(key) + || TailFileState.StateKeys.FILENAME.equals(key) + || TailFileState.StateKeys.POSITION.equals(key) + || TailFileState.StateKeys.TIMESTAMP.equals(key)) { + getLogger().info("Removed state {}={} stored by older version of NiFi.", new Object[]{key, oldState.get(key)}); + continue; + } updatedState.put(key, oldState.get(key)); } diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestTailFile.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestTailFile.java index 0cda3f0142f4..efd314c91038 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestTailFile.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestTailFile.java @@ -18,16 +18,26 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import java.io.BufferedWriter; import java.io.File; +import java.io.FileOutputStream; import java.io.FilenameFilter; import java.io.IOException; +import java.io.OutputStreamWriter; import java.io.RandomAccessFile; +import java.util.HashMap; +import java.util.Map; import java.util.Optional; import java.util.regex.Pattern; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; import org.apache.nifi.processors.standard.TailFile.TailFileState; +import org.apache.nifi.state.MockStateManager; +import org.apache.nifi.stream.io.ByteArrayOutputStream; import org.apache.nifi.util.MockFlowFile; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; @@ -39,6 +49,7 @@ public class TestTailFile { private File file; + private File existingFile; private File otherFile; private RandomAccessFile raf; @@ -56,6 +67,19 @@ public void setup() throws IOException { file.delete(); assertTrue(file.createNewFile()); + existingFile = new File("target/existing-log.txt"); + existingFile.delete(); + assertTrue(existingFile.createNewFile()); + try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(existingFile)))) { + writer.write("Line 1"); + writer.newLine(); + writer.write("Line 2"); + writer.newLine(); + writer.write("Line 3"); + writer.newLine(); + writer.flush(); + } + File directory = new File("target/testDir"); if(!directory.exists()) { assertTrue(directory.mkdirs()); @@ -812,6 +836,76 @@ public void testMultipleFilesChangingNameStrategy() throws IOException, Interrup runner.clearTransferState(); } + @Test + public void testMigrateFrom100To110() throws IOException { + + runner.setProperty(TailFile.FILENAME, "target/existing-log.txt"); + + final MockStateManager stateManager = runner.getStateManager(); + + // Before NiFi 1.1.0, TailFile only handles single file + // and state key doesn't have index in it. + final Map state = new HashMap<>(); + state.put("filename", "target/existing-log.txt"); + // Simulate that it has been tailed up to the 2nd line. + state.put("checksum", "2279929157"); + state.put("position", "14"); + state.put("timestamp", "1480639134000"); + stateManager.setState(state, Scope.LOCAL); + + runner.run(); + + runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 1); + final MockFlowFile flowFile = runner.getFlowFilesForRelationship(TailFile.REL_SUCCESS).iterator().next(); + + final ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (final BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(bos))) { + writer.write("Line 3"); + writer.newLine(); + } + + flowFile.assertContentEquals(bos.toByteArray()); + + // The old states should be replaced with new ones. + final StateMap updatedState = stateManager.getState(Scope.LOCAL); + assertNull(updatedState.get("filename")); + assertNull(updatedState.get("checksum")); + assertNull(updatedState.get("position")); + assertNull(updatedState.get("timestamp")); + assertEquals("target/existing-log.txt", updatedState.get("file.0.filename")); + assertEquals("3380848603", updatedState.get("file.0.checksum")); + assertEquals("21", updatedState.get("file.0.position")); + assertNotNull(updatedState.get("file.0.timestamp")); + + // When it runs again, the state is already migrated, so it shouldn't emit any flow files. + runner.clearTransferState(); + runner.run(); + runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 0); + } + + + @Test + public void testMigrateFrom100To110FileNotFound() throws IOException { + + runner.setProperty(TailFile.FILENAME, "target/not-existing-log.txt"); + + final MockStateManager stateManager = runner.getStateManager(); + + // Before NiFi 1.1.0, TailFile only handles single file + // and state key doesn't have index in it. + final Map state = new HashMap<>(); + state.put("filename", "target/not-existing-log.txt"); + // Simulate that it has been tailed up to the 2nd line. + state.put("checksum", "2279929157"); + state.put("position", "14"); + state.put("timestamp", "1480639134000"); + stateManager.setState(state, Scope.LOCAL); + + runner.run(); + + runner.assertTransferCount(TailFile.REL_SUCCESS, 0); + } + private void cleanFiles(String directory) { final File targetDir = new File(directory); if(targetDir.exists()) {