http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/io/fs/FileSplitterInputTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/io/fs/FileSplitterInputTest.java b/library/src/test/java/com/datatorrent/lib/io/fs/FileSplitterInputTest.java index e5a2832..acbf93b 100644 --- a/library/src/test/java/com/datatorrent/lib/io/fs/FileSplitterInputTest.java +++ b/library/src/test/java/com/datatorrent/lib/io/fs/FileSplitterInputTest.java @@ -141,7 +141,7 @@ public class FileSplitterInputTest @Rule public TestMeta testMeta = new TestMeta(); - + private void window1TestHelper() throws InterruptedException { testMeta.fileSplitterInput.beginWindow(1); @@ -157,7 +157,7 @@ public class FileSplitterInputTest Assert.assertTrue("path: " + metadata.getFilePath(), testMeta.filePaths.contains(metadata.getFilePath())); Assert.assertNotNull("name: ", metadata.getFileName()); } - + testMeta.fileMetadataSink.collectedTuples.clear(); } @@ -256,7 +256,7 @@ public class FileSplitterInputTest testMeta.fileSplitterInput = KryoCloneUtils.cloneObject(testMeta.fileSplitterInput); testMeta.resetSinks(); - + testMeta.fileSplitterInput.setup(testMeta.context); testMeta.fileSplitterInput.beginWindow(1); Assert.assertEquals("Blocks", 12, testMeta.blockMetadataSink.collectedTuples.size()); @@ -325,22 +325,22 @@ public class FileSplitterInputTest Assert.assertEquals("window 2: blocks", 1, testMeta.blockMetadataSink.collectedTuples.size()); testMeta.fileSplitterInput.teardown(); } - + private void blocksTestHelper() throws InterruptedException { testMeta.fileSplitterInput.beginWindow(1); testMeta.scanner.semaphore.acquire(); testMeta.fileSplitterInput.emitTuples(); testMeta.fileSplitterInput.endWindow(); - + Assert.assertEquals("Blocks", 10, testMeta.blockMetadataSink.collectedTuples.size()); - + for (int window = 2; window < 8; window++) { testMeta.fileSplitterInput.beginWindow(window); testMeta.fileSplitterInput.emitTuples(); testMeta.fileSplitterInput.endWindow(); } - + int noOfBlocks = 0; for (int i = 0; i < 12; i++) { File testFile = new File(testMeta.dataDirectory, "file" + i + ".txt"); @@ -384,7 +384,7 @@ public class FileSplitterInputTest { FSWindowDataManager fsWindowDataManager = new FSWindowDataManager(); testMeta.updateConfig(fsWindowDataManager, 500, 2L, 10); - + testMeta.fileSplitterInput.setup(testMeta.context); recoveryTestHelper(); testMeta.fileSplitterInput.teardown(); @@ -398,7 +398,7 @@ public class FileSplitterInputTest testMeta.fileSplitterInput.setup(testMeta.context); recoveryTestHelper(); - + Thread.sleep(1000); HashSet<String> lines = Sets.newHashSet(); for (int line = 2; line < 4; line++) { @@ -535,7 +535,7 @@ public class FileSplitterInputTest { FSWindowDataManager fsWindowDataManager = new FSWindowDataManager(); testMeta.updateConfig(fsWindowDataManager, 500L, 2L, 2); - + testMeta.fileSplitterInput.setup(testMeta.context); testMeta.fileSplitterInput.beginWindow(1); @@ -613,7 +613,7 @@ public class FileSplitterInputTest Assert.assertEquals("window 2: files", 0, testMeta.fileMetadataSink.collectedTuples.size()); Assert.assertEquals("window 2: blocks", 0, testMeta.blockMetadataSink.collectedTuples.size()); - + testMeta.fileSplitterInput.teardown(); } @@ -694,6 +694,6 @@ public class FileSplitterInputTest super.scanIterationComplete(); } } - + private static final Logger LOG = LoggerFactory.getLogger(FileSplitterInputTest.class); }
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/io/fs/SynchronizerTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/io/fs/SynchronizerTest.java b/library/src/test/java/com/datatorrent/lib/io/fs/SynchronizerTest.java index b017f20..7bdee4f 100644 --- a/library/src/test/java/com/datatorrent/lib/io/fs/SynchronizerTest.java +++ b/library/src/test/java/com/datatorrent/lib/io/fs/SynchronizerTest.java @@ -39,7 +39,7 @@ public class SynchronizerTest public static final String[] FILE_NAMES = {"a.txt", "b.txt", "c.txt", "d.txt", "e.txt" }; public static final long[][] BLOCK_IDS = { - //Block ids for file1 (a.txt) + //Block ids for file1 (a.txt) {1001, 1002, 1003 }, //Block ids for file2 (b.txt) {1004, 1005, 1006, 1007 }, http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/io/fs/TailFsInputOperatorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/io/fs/TailFsInputOperatorTest.java b/library/src/test/java/com/datatorrent/lib/io/fs/TailFsInputOperatorTest.java index 34ffecd..f273e4d 100644 --- a/library/src/test/java/com/datatorrent/lib/io/fs/TailFsInputOperatorTest.java +++ b/library/src/test/java/com/datatorrent/lib/io/fs/TailFsInputOperatorTest.java @@ -192,7 +192,7 @@ public class TailFsInputOperatorTest /** * This tests the case when the file is rotated and new file has same size as old file - * + * * @throws Exception */ @@ -231,7 +231,7 @@ public class TailFsInputOperatorTest oper.deactivate(); file = new File(filePath); if (file.exists()) { - file.delete(); + file.delete(); } file = new File(filePath + ".bk"); if (file.exists()) { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/logs/FilteredLineToTokenHashMapTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/logs/FilteredLineToTokenHashMapTest.java b/library/src/test/java/com/datatorrent/lib/logs/FilteredLineToTokenHashMapTest.java index 532aef5..dfd9627 100644 --- a/library/src/test/java/com/datatorrent/lib/logs/FilteredLineToTokenHashMapTest.java +++ b/library/src/test/java/com/datatorrent/lib/logs/FilteredLineToTokenHashMapTest.java @@ -28,7 +28,7 @@ import org.junit.Test; import com.datatorrent.lib.testbench.HashTestSink; /** - * + * * Functional tests for * {@link com.datatorrent.lib.logs.FilteredLineToTokenHashMap} */ http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/logs/LineToTokenArrayListTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/logs/LineToTokenArrayListTest.java b/library/src/test/java/com/datatorrent/lib/logs/LineToTokenArrayListTest.java index 92575b7..ef38f51 100644 --- a/library/src/test/java/com/datatorrent/lib/logs/LineToTokenArrayListTest.java +++ b/library/src/test/java/com/datatorrent/lib/logs/LineToTokenArrayListTest.java @@ -29,10 +29,10 @@ import org.junit.Test; import com.datatorrent.lib.testbench.ArrayListTestSink; /** - * + * * Functional tests for {@link com.datatorrent.lib.logs.LineToTokenArrayList} * <p> - * + * */ public class LineToTokenArrayListTest { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/logs/LineToTokenHashMapTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/logs/LineToTokenHashMapTest.java b/library/src/test/java/com/datatorrent/lib/logs/LineToTokenHashMapTest.java index 20a799e..e9ab8fb 100644 --- a/library/src/test/java/com/datatorrent/lib/logs/LineToTokenHashMapTest.java +++ b/library/src/test/java/com/datatorrent/lib/logs/LineToTokenHashMapTest.java @@ -28,7 +28,7 @@ import org.junit.Test; import com.datatorrent.lib.testbench.HashTestSink; /** - * + * * Functional tests for {@link com.datatorrent.lib.logs.LineToTokenHashMap}. */ public class LineToTokenHashMapTest http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/logs/LineTokenizerKeyValTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/logs/LineTokenizerKeyValTest.java b/library/src/test/java/com/datatorrent/lib/logs/LineTokenizerKeyValTest.java index 7b6613b..1382c55 100644 --- a/library/src/test/java/com/datatorrent/lib/logs/LineTokenizerKeyValTest.java +++ b/library/src/test/java/com/datatorrent/lib/logs/LineTokenizerKeyValTest.java @@ -27,7 +27,7 @@ import org.junit.Test; import com.datatorrent.lib.testbench.HashTestSink; /** - * + * * Functional tests for {@link com.datatorrent.lib.logs.LineTokenizerKeyVal}. */ public class LineTokenizerKeyValTest http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/logs/MultiWindowDimensionAggregationTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/logs/MultiWindowDimensionAggregationTest.java b/library/src/test/java/com/datatorrent/lib/logs/MultiWindowDimensionAggregationTest.java index 94845a7..11d5622 100644 --- a/library/src/test/java/com/datatorrent/lib/logs/MultiWindowDimensionAggregationTest.java +++ b/library/src/test/java/com/datatorrent/lib/logs/MultiWindowDimensionAggregationTest.java @@ -37,7 +37,7 @@ import com.datatorrent.lib.testbench.CollectorTestSink; * <p> * MultiWindowDimensionAggregationTest class. * </p> - * + * */ public class MultiWindowDimensionAggregationTest { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/math/MarginMapTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/math/MarginMapTest.java b/library/src/test/java/com/datatorrent/lib/math/MarginMapTest.java index 07a378c..28866a8 100644 --- a/library/src/test/java/com/datatorrent/lib/math/MarginMapTest.java +++ b/library/src/test/java/com/datatorrent/lib/math/MarginMapTest.java @@ -29,10 +29,10 @@ import org.slf4j.LoggerFactory; import com.datatorrent.lib.testbench.CountAndLastTupleTestSink; /** - * + * * Functional tests for {@link com.datatorrent.lib.math.MarginMap} * <p> - * + * */ public class MarginMapTest { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/math/SigmaTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/math/SigmaTest.java b/library/src/test/java/com/datatorrent/lib/math/SigmaTest.java index f74e0c9..8608a96 100644 --- a/library/src/test/java/com/datatorrent/lib/math/SigmaTest.java +++ b/library/src/test/java/com/datatorrent/lib/math/SigmaTest.java @@ -29,7 +29,7 @@ import com.datatorrent.lib.testbench.SumTestSink; * * Functional tests for {@link com.datatorrent.lib.math.Sigma} * <p> - * + * */ public class SigmaTest http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/multiwindow/SortedMovingWindowTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/multiwindow/SortedMovingWindowTest.java b/library/src/test/java/com/datatorrent/lib/multiwindow/SortedMovingWindowTest.java index 10e0f5e..8030d97 100644 --- a/library/src/test/java/com/datatorrent/lib/multiwindow/SortedMovingWindowTest.java +++ b/library/src/test/java/com/datatorrent/lib/multiwindow/SortedMovingWindowTest.java @@ -38,7 +38,7 @@ import com.datatorrent.lib.testbench.CollectorTestSink; * 1. sort simple comparable tuples * 2. sort tuples by given order (Comparator) * 3. group tuples into different category and sort the category by a given order - * + * */ public class SortedMovingWindowTest { @@ -53,20 +53,20 @@ public class SortedMovingWindowTest CollectorTestSink<Object> testSink = new CollectorTestSink<Object>(); smw.outputPort.setSink(testSink); smw.setup(null); - + smw.setWindowSize(2); - // The incoming 6 integer tuples are disordered among 4 windows + // The incoming 6 integer tuples are disordered among 4 windows emitObjects(smw, new Integer[][]{{1,3}, {2,5}, {4}, {6}}); smw.beginWindow(4); smw.endWindow(); smw.beginWindow(5); smw.endWindow(); - + // The outcome is sorted Assert.assertEquals(Lists.newArrayList(1, 2, 3, 4, 5, 6), testSink.collectedTuples); - + } - + /** * Given sorting key, sorting function, test sorting the map tuples within the sliding window */ @@ -92,7 +92,7 @@ public class SortedMovingWindowTest smw.setup(null); smw.setWindowSize(2); - // The incoming 6 simple map tuples are disordered among 4 windows + // The incoming 6 simple map tuples are disordered among 4 windows emitObjects(smw, new Map[][]{createHashMapTuples(keys, new Integer[][]{{1}, {3}}), createHashMapTuples(keys, new Integer[][]{{2}, {5}}), createHashMapTuples(keys, new Integer[][]{{4}}), createHashMapTuples(keys, new Integer[][]{{6}})}); @@ -105,8 +105,8 @@ public class SortedMovingWindowTest Assert.assertEquals(Arrays.asList(createHashMapTuples(keys, new Integer[][]{{1}, {2}, {3}, {4}, {5}, {6}})), testSink.collectedTuples); } - - + + /** * Given grouping key, sorting key and sorting function, test sorting the map tuples within the sliding window */ @@ -127,7 +127,7 @@ public class SortedMovingWindowTest return (Integer)o1.get(keys[1]) - (Integer)o2.get(keys[1]); } }); - + smw.setFunction(new Function<Map<String,Object>, String>() { @Override @@ -142,7 +142,7 @@ public class SortedMovingWindowTest smw.setup(null); smw.setWindowSize(2); - // The incoming 9 complex map tuples are disordered with same name among 4 windows + // The incoming 9 complex map tuples are disordered with same name among 4 windows emitObjects(smw, new Map[][]{createHashMapTuples(keys, new Object[][]{{"bob", 1}, {"jim", 1}}), createHashMapTuples(keys, new Object[][]{{"jim", 2}, {"bob", 3}}), createHashMapTuples(keys, new Object[][]{{"bob", 2}, {"jim", 4}}), @@ -156,7 +156,7 @@ public class SortedMovingWindowTest Assert.assertEquals(Arrays.asList(createHashMapTuples(keys, new Object[][]{{"bob", 1}, {"jim", 1}, {"jim", 2}, {"bob", 2}, {"bob", 3}, {"jim", 3}, {"jim", 4}, {"bob", 4}, {"bob", 5}})), testSink.collectedTuples); } - + @SuppressWarnings({"rawtypes", "unchecked"}) private void emitObjects(SortedMovingWindow win, Object[][] obj) { @@ -168,11 +168,11 @@ public class SortedMovingWindowTest win.endWindow(); } } - + @SuppressWarnings({"rawtypes", "unchecked"}) private Map[] createHashMapTuples(String[] cols, Object[][] values) { - + HashMap[] maps = new HashMap[values.length]; int index = -1; for (Object[] vs : values) { @@ -182,8 +182,8 @@ public class SortedMovingWindowTest maps[index].put(cols[colIndex++], value); } } - + return maps; } - + } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/parser/XmlParserTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/parser/XmlParserTest.java b/library/src/test/java/com/datatorrent/lib/parser/XmlParserTest.java index 5ce07f5..670d5b6 100644 --- a/library/src/test/java/com/datatorrent/lib/parser/XmlParserTest.java +++ b/library/src/test/java/com/datatorrent/lib/parser/XmlParserTest.java @@ -209,7 +209,7 @@ public class XmlParserTest @Test public void testXmlToPojoIncorrectXML() { - String tuple = "<EmployeeBean>" + "<firstname>john</firstname>" //incorrect field name is ignored by JAXB + String tuple = "<EmployeeBean>" + "<firstname>john</firstname>" //incorrect field name is ignored by JAXB + "<dept>cs</dept>" + "<eid>1</eid>" + "<dateOfJoining>2015-01-01 00:00:00.00 IST</dateOfJoining>"; // + "</EmployeeBean>"; // Incorrect XML format http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/statistics/MeridianOperatorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/statistics/MeridianOperatorTest.java b/library/src/test/java/com/datatorrent/lib/statistics/MeridianOperatorTest.java index 47fa2c2..b7979c6 100644 --- a/library/src/test/java/com/datatorrent/lib/statistics/MeridianOperatorTest.java +++ b/library/src/test/java/com/datatorrent/lib/statistics/MeridianOperatorTest.java @@ -43,7 +43,7 @@ public class MeridianOperatorTest oper.data.process(3.0); oper.data.process(9.0); oper.endWindow(); - + Assert.assertEquals("Must be one tuple in sink", sink.collectedTuples.size(), 1); Assert.assertTrue("Median value", sink.collectedTuples.get(0).doubleValue() == 5.0); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/statistics/ModeOperatorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/statistics/ModeOperatorTest.java b/library/src/test/java/com/datatorrent/lib/statistics/ModeOperatorTest.java index 26e94c6..d38766e 100644 --- a/library/src/test/java/com/datatorrent/lib/statistics/ModeOperatorTest.java +++ b/library/src/test/java/com/datatorrent/lib/statistics/ModeOperatorTest.java @@ -45,7 +45,7 @@ public class ModeOperatorTest oper.data.process(7.0); oper.data.process(7.0); oper.endWindow(); - + Assert.assertEquals("Must be one tuple in sink", sink.collectedTuples.size(), 1); Assert.assertTrue("Median value", sink.collectedTuples.get(0) == 7.0); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/statistics/StandardDeviationOperatorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/statistics/StandardDeviationOperatorTest.java b/library/src/test/java/com/datatorrent/lib/statistics/StandardDeviationOperatorTest.java index 3d2d99c..b32f202 100644 --- a/library/src/test/java/com/datatorrent/lib/statistics/StandardDeviationOperatorTest.java +++ b/library/src/test/java/com/datatorrent/lib/statistics/StandardDeviationOperatorTest.java @@ -36,7 +36,7 @@ public class StandardDeviationOperatorTest oper.variance.setSink(variance); CollectorTestSink<Object> deviation = new CollectorTestSink<Object>(); oper.standardDeviation.setSink(deviation); - + oper.setup(null); oper.beginWindow(0); oper.data.process(1.0); @@ -44,7 +44,7 @@ public class StandardDeviationOperatorTest oper.data.process(3.0); oper.data.process(9.0); oper.endWindow(); - + Assert.assertEquals("Must be one tuple in sink", variance.collectedTuples.size(), 1); Assert.assertEquals("Must be one tuple in sink", deviation.collectedTuples.size(), 1); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/statistics/WeightedMeanOperatorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/statistics/WeightedMeanOperatorTest.java b/library/src/test/java/com/datatorrent/lib/statistics/WeightedMeanOperatorTest.java index f9589db..90066a6 100644 --- a/library/src/test/java/com/datatorrent/lib/statistics/WeightedMeanOperatorTest.java +++ b/library/src/test/java/com/datatorrent/lib/statistics/WeightedMeanOperatorTest.java @@ -45,7 +45,7 @@ public class WeightedMeanOperatorTest oper.data.process(2.0); oper.data.process(4.0); oper.endWindow(); - + Assert.assertEquals("Must be one tuple in sink", sink.collectedTuples.size(), 1); Assert.assertTrue("Expected mean value", sink.collectedTuples.get(0) == 3.0); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/stream/DevNullCounterTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/stream/DevNullCounterTest.java b/library/src/test/java/com/datatorrent/lib/stream/DevNullCounterTest.java index 6266787..c027a21 100644 --- a/library/src/test/java/com/datatorrent/lib/stream/DevNullCounterTest.java +++ b/library/src/test/java/com/datatorrent/lib/stream/DevNullCounterTest.java @@ -25,7 +25,7 @@ import org.slf4j.LoggerFactory; import com.datatorrent.lib.testbench.EventGenerator; /** - * + * * Functional tests for {@link com.datatorrent.lib.testbench.DevNullCounter}. * <p> * <br> @@ -35,7 +35,7 @@ import com.datatorrent.lib.testbench.EventGenerator; * Benchmarks:<br> * Object payload benchmarked at over 125 Million/sec <br> * DRC checks are validated<br> - * + * */ public class DevNullCounterTest { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/stream/DevNullTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/stream/DevNullTest.java b/library/src/test/java/com/datatorrent/lib/stream/DevNullTest.java index 3bd9c11..a9563ca 100644 --- a/library/src/test/java/com/datatorrent/lib/stream/DevNullTest.java +++ b/library/src/test/java/com/datatorrent/lib/stream/DevNullTest.java @@ -25,7 +25,7 @@ import org.slf4j.LoggerFactory; import com.datatorrent.lib.testbench.EventGenerator; /** - * Functional tests for {@link com.datatorrent.lib.testbench.DevNull}. + * Functional tests for {@link com.datatorrent.lib.testbench.DevNull}. */ public class DevNullTest { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/testbench/ActiveMQMessageGenerator.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/testbench/ActiveMQMessageGenerator.java b/library/src/test/java/com/datatorrent/lib/testbench/ActiveMQMessageGenerator.java index 4f231cb..c569bf8 100644 --- a/library/src/test/java/com/datatorrent/lib/testbench/ActiveMQMessageGenerator.java +++ b/library/src/test/java/com/datatorrent/lib/testbench/ActiveMQMessageGenerator.java @@ -36,7 +36,7 @@ import org.apache.activemq.ActiveMQConnectionFactory; * This is the message generator outside of Malhar/Hadoop. This generates data * and send to ActiveMQ message bus so that Malhar input adapter for ActiveMQ * can receive it. - * + * */ public class ActiveMQMessageGenerator { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/testbench/RandomEventGeneratorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/testbench/RandomEventGeneratorTest.java b/library/src/test/java/com/datatorrent/lib/testbench/RandomEventGeneratorTest.java index 6876525..b9210f7 100644 --- a/library/src/test/java/com/datatorrent/lib/testbench/RandomEventGeneratorTest.java +++ b/library/src/test/java/com/datatorrent/lib/testbench/RandomEventGeneratorTest.java @@ -76,7 +76,7 @@ public class RandomEventGeneratorTest node.integer_data.setSink(integer_data); CollectorTestSink string_data = new CollectorTestSink(); node.string_data.setSink(string_data); - + node.setup(null); node.beginWindow(1); node.emitTuples(); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/com/datatorrent/lib/util/TestUtils.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/com/datatorrent/lib/util/TestUtils.java b/library/src/test/java/com/datatorrent/lib/util/TestUtils.java index 673054b..403072d 100644 --- a/library/src/test/java/com/datatorrent/lib/util/TestUtils.java +++ b/library/src/test/java/com/datatorrent/lib/util/TestUtils.java @@ -82,7 +82,7 @@ public class TestUtils { FileUtils.deleteQuietly(new File("target/" + description.getClassName())); } - + @SuppressWarnings({"unchecked", "rawtypes"}) public static <S extends Sink, T> S setSink(OutputPort<T> port, S sink) { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/dedup/DeduperPartitioningTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/dedup/DeduperPartitioningTest.java b/library/src/test/java/org/apache/apex/malhar/lib/dedup/DeduperPartitioningTest.java index 479fbe8..fb03f2e 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/dedup/DeduperPartitioningTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/dedup/DeduperPartitioningTest.java @@ -71,13 +71,13 @@ public class DeduperPartitioningTest dedup.setTimeExpression("eventTime.getTime()"); dedup.setBucketSpan(60); dedup.setExpireBefore(600); - + ConsoleOutputOperator console = dag.addOperator("Console", new ConsoleOutputOperator()); dag.addStream("Generator to Dedup", gen.output, dedup.input); dag.addStream("Dedup to Console", dedup.unique, console.input); dag.setInputPortAttribute(dedup.input, Context.PortContext.TUPLE_CLASS, TestEvent.class); dag.setOutputPortAttribute(dedup.unique, Context.PortContext.TUPLE_CLASS, TestEvent.class); - dag.setAttribute(dedup, Context.OperatorContext.PARTITIONER, + dag.setAttribute(dedup, Context.OperatorContext.PARTITIONER, new StatelessPartitioner<TimeBasedDedupOperator>(NUM_DEDUP_PARTITIONS)); } } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/fs/FSRecordReaderTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/fs/FSRecordReaderTest.java b/library/src/test/java/org/apache/apex/malhar/lib/fs/FSRecordReaderTest.java index ecaff70..8560228 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/fs/FSRecordReaderTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/fs/FSRecordReaderTest.java @@ -90,10 +90,10 @@ public class FSRecordReaderTest LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(true); lc.runAsync(); - + Set<String> expectedRecords = new HashSet<String>(Arrays.asList(FILE_1_DATA.split("\n"))); expectedRecords.addAll(Arrays.asList(FILE_2_DATA.split("\n"))); - + while (DelimitedValidator.records.size() != expectedRecords.size()) { LOG.debug("Waiting for app to finish"); Thread.sleep(1000); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/fs/GenericFileOutputOperatorTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/fs/GenericFileOutputOperatorTest.java b/library/src/test/java/org/apache/apex/malhar/lib/fs/GenericFileOutputOperatorTest.java index 52d5c5a..6082f57 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/fs/GenericFileOutputOperatorTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/fs/GenericFileOutputOperatorTest.java @@ -38,7 +38,7 @@ public class GenericFileOutputOperatorTest extends AbstractFileOutputOperatorTes /** * Test file rollover in case of idle windows - * + * * @throws IOException */ @Test @@ -84,7 +84,7 @@ public class GenericFileOutputOperatorTest extends AbstractFileOutputOperatorTes /** * Test file rollover for tuple count - * + * * @throws IOException */ @Test @@ -120,7 +120,7 @@ public class GenericFileOutputOperatorTest extends AbstractFileOutputOperatorTes checkOutput(i, testMeta.getDir() + "/output.txt_0", expected[i], true); } } - + public static void checkOutput(int fileCount, String baseFilePath, String expectedOutput, boolean checkTmp) { if (fileCount >= 0) { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableByteArrayListMultimapImplTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableByteArrayListMultimapImplTest.java b/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableByteArrayListMultimapImplTest.java index 2c9d7eb..22d317d 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableByteArrayListMultimapImplTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableByteArrayListMultimapImplTest.java @@ -339,7 +339,7 @@ public class SpillableByteArrayListMultimapImplTest map.teardown(); store.teardown(); } - + @Test public void testLoad() { @@ -349,10 +349,10 @@ public class SpillableByteArrayListMultimapImplTest final int numOfEntry = 100000; SpillableStateStore store = testMeta.store; - + SpillableByteArrayListMultimapImpl<String, String> multimap = new SpillableByteArrayListMultimapImpl<>( this.testMeta.store, ID1, 0L, new SerdeStringSlice(), new SerdeStringSlice()); - + Attribute.AttributeMap.DefaultAttributeMap attributes = new Attribute.AttributeMap.DefaultAttributeMap(); attributes.put(DAG.APPLICATION_PATH, testMeta.applicationPath); Context.OperatorContext context = http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableComplexComponentImplTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableComplexComponentImplTest.java b/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableComplexComponentImplTest.java index 67db6ba..96855e0 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableComplexComponentImplTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableComplexComponentImplTest.java @@ -42,7 +42,7 @@ public class SpillableComplexComponentImplTest { simpleIntegrationTestHelper(testMeta.store); } - + public void simpleIntegrationTestHelper(SpillableStateStore store) { SpillableComplexComponentImpl sccImpl = new SpillableComplexComponentImpl(store); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableSetMultimapImplTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableSetMultimapImplTest.java b/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableSetMultimapImplTest.java index e9903ec..6b188e4 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableSetMultimapImplTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/state/spillable/SpillableSetMultimapImplTest.java @@ -266,7 +266,7 @@ public class SpillableSetMultimapImplTest map.teardown(); store.teardown(); } - + @Test public void testLoad() { @@ -276,10 +276,10 @@ public class SpillableSetMultimapImplTest final int numOfEntry = 100000; SpillableStateStore store = testMeta.store; - + SpillableByteArrayListMultimapImpl<String, String> multimap = new SpillableByteArrayListMultimapImpl<>( this.testMeta.store, ID1, 0L, new SerdeStringSlice(), new SerdeStringSlice()); - + Attribute.AttributeMap.DefaultAttributeMap attributes = new Attribute.AttributeMap.DefaultAttributeMap(); attributes.put(DAG.APPLICATION_PATH, testMeta.applicationPath); Context.OperatorContext context = http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/wal/FSWindowDataManagerTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/wal/FSWindowDataManagerTest.java b/library/src/test/java/org/apache/apex/malhar/lib/wal/FSWindowDataManagerTest.java index 9939bb9..bef1ae4 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/wal/FSWindowDataManagerTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/wal/FSWindowDataManagerTest.java @@ -206,33 +206,33 @@ public class FSWindowDataManagerTest Assert.assertEquals("largest recovery window", 1, manager.getLargestCompletedWindow()); manager.teardown(); } - + @Test public void testDelete() throws IOException { Pair<Context.OperatorContext, FSWindowDataManager> pair1 = createManagerAndContextFor(1); pair1.second.getWal().setMaxLength(2); pair1.second.setup(pair1.first); - + Map<Integer, String> dataOf1 = Maps.newHashMap(); dataOf1.put(1, "one"); dataOf1.put(2, "two"); dataOf1.put(3, "three"); - + for (int i = 1; i <= 9; ++i) { pair1.second.save(dataOf1, i); } - + pair1.second.committed(3); pair1.second.teardown(); - + Pair<Context.OperatorContext, FSWindowDataManager> pair1AfterRecovery = createManagerAndContextFor(1); testMeta.attributes.put(Context.OperatorContext.ACTIVATION_WINDOW_ID, 1L); pair1AfterRecovery.second.setup(pair1AfterRecovery.first); - + Assert.assertEquals("window 1 deleted", null, pair1AfterRecovery.second.retrieve(1)); Assert.assertEquals("window 3 deleted", null, pair1AfterRecovery.second.retrieve(3)); - + Assert.assertEquals("window 4 exists", dataOf1, pair1AfterRecovery.second.retrieve(4)); pair1.second.teardown(); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/AverageTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/AverageTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/AverageTest.java index e5fd541..bfae223 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/AverageTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/AverageTest.java @@ -32,7 +32,7 @@ public class AverageTest { Average ave = new Average(); MutablePair<Double, Long> accu = ave.defaultAccumulatedValue(); - + for (int i = 1; i <= 10; i++) { accu = ave.accumulate(accu, (double)i); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/FoldFnTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/FoldFnTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/FoldFnTest.java index cda6bf8..34a1812 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/FoldFnTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/FoldFnTest.java @@ -34,16 +34,16 @@ public class FoldFnTest public static class NumGen extends BaseOperator implements InputOperator { public transient DefaultOutputPort<Integer> output = new DefaultOutputPort<>(); - + public static int count = 0; private int i = 0; - + public NumGen() { count = 0; i = 0; } - + @Override public void emitTuples() { @@ -61,11 +61,11 @@ public class FoldFnTest i = -1; } } - + public static class Collector extends BaseOperator { private static int result; - + public transient DefaultInputPort<Tuple.WindowedTuple<Integer>> input = new DefaultInputPort<Tuple.WindowedTuple<Integer>>() { @Override @@ -74,13 +74,13 @@ public class FoldFnTest result = tuple.getValue(); } }; - + public int getResult() { return result; } } - + public static class Plus extends FoldFn<Integer, Integer> { @Override @@ -88,7 +88,7 @@ public class FoldFnTest { return fold(accumulatedValue1, accumulatedValue2); } - + @Override public Integer fold(Integer input1, Integer input2) { @@ -98,11 +98,11 @@ public class FoldFnTest return input1 + input2; } } - + @Test public void FoldFnTest() { - + FoldFn<String, String> concat = new FoldFn<String, String>() { @Override @@ -110,17 +110,17 @@ public class FoldFnTest { return fold(accumulatedValue1, accumulatedValue2); } - + @Override public String fold(String input1, String input2) { return input1 + ", " + input2; } }; - + String[] ss = new String[]{"b", "c", "d", "e"}; String base = "a"; - + for (String s : ss) { base = concat.accumulate(base, s); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/GroupTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/GroupTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/GroupTest.java index 891a824..d0455f8 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/GroupTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/GroupTest.java @@ -31,7 +31,7 @@ public class GroupTest public void GroupTest() { Group<Integer> group = new Group<>(); - + List<Integer> accu = group.defaultAccumulatedValue(); Assert.assertEquals(0, accu.size()); Assert.assertEquals(1, group.accumulate(accu, 10).size()); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MaxTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MaxTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MaxTest.java index fe87d9e..5c5a761 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MaxTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MaxTest.java @@ -31,11 +31,11 @@ public class MaxTest public void MaxTest() { Max<Integer> max = new Max<>(); - + Assert.assertEquals((Integer)5, max.accumulate(5, 3)); Assert.assertEquals((Integer)6, max.accumulate(4, 6)); Assert.assertEquals((Integer)5, max.merge(5, 2)); - + Comparator<Integer> com = new Comparator<Integer>() { @Override @@ -44,7 +44,7 @@ public class MaxTest return -(o1.compareTo(o2)); } }; - + max.setComparator(com); Assert.assertEquals((Integer)3, max.accumulate(5, 3)); Assert.assertEquals((Integer)4, max.accumulate(4, 6)); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MinTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MinTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MinTest.java index 3589735..7bf5c39 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MinTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/MinTest.java @@ -31,11 +31,11 @@ public class MinTest public void MinTest() { Min<Integer> min = new Min<>(); - + Assert.assertEquals((Integer)3, min.accumulate(5, 3)); Assert.assertEquals((Integer)4, min.accumulate(4, 6)); Assert.assertEquals((Integer)2, min.merge(5, 2)); - + Comparator<Integer> com = new Comparator<Integer>() { @Override @@ -44,7 +44,7 @@ public class MinTest return -(o1.compareTo(o2)); } }; - + min.setComparator(com); Assert.assertEquals((Integer)5, min.accumulate(5, 3)); Assert.assertEquals((Integer)6, min.accumulate(4, 6)); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/ReduceFnTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/ReduceFnTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/ReduceFnTest.java index 26d73a7..798c16c 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/ReduceFnTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/ReduceFnTest.java @@ -26,7 +26,7 @@ import org.junit.Test; */ public class ReduceFnTest { - + @Test public void ReduceFnTest() { @@ -38,10 +38,10 @@ public class ReduceFnTest return input1 + ", " + input2; } }; - + String[] ss = new String[]{"b", "c", "d", "e"}; String base = "a"; - + for (String s : ss) { base = concat.accumulate(base, s); } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/RemoveDuplicatesTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/RemoveDuplicatesTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/RemoveDuplicatesTest.java index 674f871..37bfec8 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/RemoveDuplicatesTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/RemoveDuplicatesTest.java @@ -31,7 +31,7 @@ public class RemoveDuplicatesTest public void RemoveDuplicatesTest() { RemoveDuplicates<Integer> rd = new RemoveDuplicates<>(); - + Set<Integer> accu = rd.defaultAccumulatedValue(); Assert.assertEquals(0, accu.size()); Assert.assertEquals(1, rd.accumulate(accu, 10).size()); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/SumTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/SumTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/SumTest.java index 4c55612..4587a91 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/SumTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/SumTest.java @@ -37,19 +37,19 @@ public class SumTest SumLong sl = new SumLong(); SumFloat sf = new SumFloat(); SumDouble sd = new SumDouble(); - + Assert.assertEquals(new MutableInt(10), si.accumulate(si.defaultAccumulatedValue(), 10)); Assert.assertEquals(new MutableInt(11), si.accumulate(new MutableInt(1), 10)); Assert.assertEquals(new MutableInt(22), si.merge(new MutableInt(1), new MutableInt(21))); - + Assert.assertEquals(new MutableLong(10L), sl.accumulate(sl.defaultAccumulatedValue(), 10L)); Assert.assertEquals(new MutableLong(22L), sl.accumulate(new MutableLong(2L), 20L)); Assert.assertEquals(new MutableLong(41L), sl.merge(new MutableLong(32L), new MutableLong(9L))); - + Assert.assertEquals(new MutableFloat(9.0F), sf.accumulate(sf.defaultAccumulatedValue(), 9.0F)); Assert.assertEquals(new MutableFloat(22.5F), sf.accumulate(new MutableFloat(2.5F), 20F)); Assert.assertEquals(new MutableFloat(41.0F), sf.merge(new MutableFloat(33.1F), new MutableFloat(7.9F))); - + Assert.assertEquals(new MutableDouble(9.0), sd.accumulate(sd.defaultAccumulatedValue(), 9.0)); Assert.assertEquals(new MutableDouble(22.5), sd.accumulate(new MutableDouble(2.5), 20.0)); Assert.assertEquals(new MutableDouble(41.0), sd.merge(new MutableDouble(33.1), new MutableDouble(7.9))); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/TopNByKeyTest.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/TopNByKeyTest.java b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/TopNByKeyTest.java index 5bf2207..eec86e0 100644 --- a/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/TopNByKeyTest.java +++ b/library/src/test/java/org/apache/apex/malhar/lib/window/accumulation/TopNByKeyTest.java @@ -37,38 +37,38 @@ public class TopNByKeyTest TopNByKey<String, Integer> topNByKey = new TopNByKey<>(); topNByKey.setN(3); Map<String, Integer> accu = topNByKey.defaultAccumulatedValue(); - + Assert.assertEquals(0, accu.size()); - + accu = topNByKey.accumulate(accu, new KeyValPair<String, Integer>("1", 1)); accu = topNByKey.accumulate(accu, new KeyValPair<String, Integer>("3", 3)); - + List<KeyValPair<String, Integer>> result1 = new ArrayList<>(); - + result1.add(new KeyValPair<String, Integer>("3", 3)); result1.add(new KeyValPair<String, Integer>("1", 1)); - + Assert.assertEquals(result1, topNByKey.getOutput(accu)); - + accu = topNByKey.accumulate(accu, new KeyValPair<String, Integer>("2", 2)); - + List<KeyValPair<String, Integer>> result2 = new ArrayList<>(); - + result2.add(new KeyValPair<String, Integer>("3", 3)); result2.add(new KeyValPair<String, Integer>("2", 2)); result2.add(new KeyValPair<String, Integer>("1", 1)); - + Assert.assertEquals(result2, topNByKey.getOutput(accu)); - + accu = topNByKey.accumulate(accu, new KeyValPair<String, Integer>("5", 5)); accu = topNByKey.accumulate(accu, new KeyValPair<String, Integer>("4", 4)); - + List<KeyValPair<String, Integer>> result3 = new ArrayList<>(); - + result3.add(new KeyValPair<String, Integer>("5", 5)); result3.add(new KeyValPair<String, Integer>("4", 4)); result3.add(new KeyValPair<String, Integer>("3", 3)); - + Assert.assertEquals(result3, topNByKey.getOutput(accu)); } } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFile.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFile.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFile.java index f92d9aa..005d510 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFile.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFile.java @@ -36,7 +36,7 @@ import junit.framework.TestCase; /** * test tfile features. - * + * */ public class TestDTFile extends TestCase { @@ -323,7 +323,7 @@ public class TestDTFile extends TestCase scanner.seekTo(new byte[0]); byte[] val2 = readValue(scanner); assertTrue(Arrays.equals(val1, val2)); - + // check for lowerBound scanner.lowerBound(getSomeKey(50)); assertTrue("locaton lookup failed", scanner.currentLocation http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFileByteArrays.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFileByteArrays.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFileByteArrays.java index f3479de..7e3bfc1 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFileByteArrays.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestDTFileByteArrays.java @@ -84,7 +84,7 @@ public class TestDTFileByteArrays this.records1stBlock = numRecords1stBlock; this.records2ndBlock = numRecords2ndBlock; } - + public void init(String compression, String comparator) { this.compression = compression; @@ -252,7 +252,7 @@ public class TestDTFileByteArrays } return scanner.endLocation; } - + @Test public void testLocate() throws IOException { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparator2.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparator2.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparator2.java index 5129d3c..23ddfbe 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparator2.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparator2.java @@ -46,11 +46,11 @@ public class TestTFileComparator2 { private static long cube(long n) { return n*n*n; } - + private static String buildValue(long i) { return String.format("%s-%d", VALUE, i); } - + @Test public void testSortedLongWritable() throws IOException { Configuration conf = new Configuration(); @@ -79,11 +79,11 @@ public class TestTFileComparator2 { } } finally { writer.close(); - } + } } finally { out.close(); } - + FSDataInputStream in = fs.open(path); try { DTFile.Reader reader = new DTFile.Reader(in, fs.getFileStatus(path) http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparators.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparators.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparators.java index d588484..f5e3405 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparators.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileComparators.java @@ -30,10 +30,10 @@ import org.apache.hadoop.io.file.tfile.DTFile.Writer; import org.junit.Assert; /** - * + * * Byte arrays test case class using GZ compression codec, base class of none * and LZO compression classes. - * + * */ public class TestTFileComparators extends TestCase { private static String ROOT = http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileJClassComparatorByteArrays.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileJClassComparatorByteArrays.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileJClassComparatorByteArrays.java index 301cffc..a1d6960 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileJClassComparatorByteArrays.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileJClassComparatorByteArrays.java @@ -25,10 +25,10 @@ import org.apache.hadoop.io.RawComparator; import org.apache.hadoop.io.WritableComparator; /** - * + * * Byte arrays test case class using GZ compression codec, base class of none * and LZO compression classes. - * + * */ public class TestTFileJClassComparatorByteArrays extends TestDTFileByteArrays { @@ -54,6 +54,6 @@ class MyComparator implements RawComparator<byte[]>, Serializable { public int compare(byte[] o1, byte[] o2) { return WritableComparator.compareBytes(o1, 0, o1.length, o2, 0, o2.length); } - + } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileLzoCodecsStreams.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileLzoCodecsStreams.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileLzoCodecsStreams.java index 7c6581d..9fb0e0b 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileLzoCodecsStreams.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileLzoCodecsStreams.java @@ -33,7 +33,7 @@ public class TestTFileLzoCodecsStreams extends TestTFileStreams { System.out.println("Skipped"); } init(Compression.Algorithm.LZO.getName(), "memcmp"); - if (!skip) + if (!skip) super.setUp(); } } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileNoneCodecsJClassComparatorByteArrays.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileNoneCodecsJClassComparatorByteArrays.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileNoneCodecsJClassComparatorByteArrays.java index 31e3cad..00f08de 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileNoneCodecsJClassComparatorByteArrays.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileNoneCodecsJClassComparatorByteArrays.java @@ -21,10 +21,10 @@ package org.apache.hadoop.io.file.tfile; import java.io.IOException; /** - * + * * Byte arrays test case class using GZ compression codec, base class of none * and LZO compression classes. - * + * */ public class TestTFileNoneCodecsJClassComparatorByteArrays extends TestDTFileByteArrays { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeek.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeek.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeek.java index 6a3e086..5a9e706 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeek.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeek.java @@ -47,7 +47,7 @@ import org.apache.hadoop.io.file.tfile.DTFile.Reader.Scanner; * test the performance for seek. * */ -public class TestTFileSeek extends TestCase { +public class TestTFileSeek extends TestCase { private MyOptions options; private Configuration conf; private Path path; @@ -83,12 +83,12 @@ public class TestTFileSeek extends TestCase { new KVGenerator(rng, true, keyLenGen, valLenGen, wordLenGen, options.dictSize); } - + @Override public void tearDown() throws IOException { fs.delete(path, true); } - + private static FSDataOutputStream createFSOutput(Path name, FileSystem fs) throws IOException { if (fs.exists(name)) { @@ -140,7 +140,7 @@ public class TestTFileSeek extends TestCase { System.out.printf("time: %s...file size: %.2fMB...disk thrpt: %.2fMB/s\n", timer.toString(), (double) fsize / 1024 / 1024, fsize / duration); } - + public void seekTFile() throws IOException { int miss = 0; long totalBytes = 0; @@ -176,7 +176,7 @@ public class TestTFileSeek extends TestCase { (double) totalBytes / 1024 / (options.seekCount - miss)); } - + public void testSeeks() throws IOException { String[] supported = TFile.getSupportedCompressionAlgorithms(); boolean proceed = false; @@ -200,7 +200,7 @@ public class TestTFileSeek extends TestCase { seekTFile(); } } - + private static class IntegerRange { private final int from, to; @@ -241,7 +241,7 @@ public class TestTFileSeek extends TestCase { int fsOutputBufferSizeNone = 1; int fsOutputBufferSizeLzo = 1; int fsOutputBufferSizeGz = 1; - + String rootDir = System.getProperty("test.build.data", "target/tfile-test"); String file = "TestTFileSeek"; @@ -401,7 +401,7 @@ public class TestTFileSeek extends TestCase { if (line.hasOption('o')) { fsOutputBufferSize = Integer.parseInt(line.getOptionValue('o')); } - + if (line.hasOption('n')) { seekCount = Integer.parseInt(line.getOptionValue('n')); } @@ -425,7 +425,7 @@ public class TestTFileSeek extends TestCase { if (line.hasOption('r')) { rootDir = line.getOptionValue('r'); } - + if (line.hasOption('f')) { file = line.getOptionValue('f'); } @@ -488,11 +488,11 @@ public class TestTFileSeek extends TestCase { return (op & OP_READ) != 0; } } - + public static void main(String[] argv) throws IOException { TestTFileSeek testCase = new TestTFileSeek(); MyOptions options = new MyOptions(argv); - + if (options.proceed == false) { return; } http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeqFileComparison.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeqFileComparison.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeqFileComparison.java index a68ae45..4a01b74 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeqFileComparison.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSeqFileComparison.java @@ -469,7 +469,7 @@ public class TestTFileSeqFileComparison extends TestCase { System.out.println("Skipped for " + compress); return; } - + options.compress = compress; String parameters = parameters2String(options); createSeqFile(parameters, compress); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSplit.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSplit.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSplit.java index 2d28ae7..181be09 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSplit.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileSplit.java @@ -95,14 +95,14 @@ public class TestTFileSplit extends TestCase { reader.close(); } - /* Similar to readFile(), tests the scanner created + /* Similar to readFile(), tests the scanner created * by record numbers rather than the offsets. */ void readRowSplits(int numSplits) throws IOException { Reader reader = new Reader(fs.open(path), fs.getFileStatus(path).getLen(), conf); - + long totalRecords = reader.getEntryCount(); for (int i=0; i<numSplits; i++) { long startRec = i*totalRecords/numSplits; @@ -130,11 +130,11 @@ public class TestTFileSplit extends TestCase { Scanner scanner = reader.createScannerByRecordNum(totalRecords, -1); Assert.assertTrue(scanner.atEnd()); } - + static String composeSortedKey(String prefix, int total, int value) { return String.format("%s%010d", prefix, value); } - + void checkRecNums() throws IOException { long fileLen = fs.getFileStatus(path).getLen(); Reader reader = new Reader(fs.open(path), fileLen, conf); @@ -177,11 +177,11 @@ public class TestTFileSplit extends TestCase { .getRecordNumByLocation(reader.getLocationByRecordNum(x))); } } - + public void testSplit() throws IOException { System.out.println("testSplit"); createFile(100000, Compression.Algorithm.NONE.getName()); - checkRecNums(); + checkRecNums(); readFile(); readRowSplits(10); fs.delete(path, true); http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileStreams.java ---------------------------------------------------------------------- diff --git a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileStreams.java b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileStreams.java index 860a1de..a4a3a25 100644 --- a/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileStreams.java +++ b/library/src/test/java/org/apache/hadoop/io/file/tfile/TestTFileStreams.java @@ -36,10 +36,10 @@ import org.apache.hadoop.io.file.tfile.DTFile.Reader.Scanner; import org.junit.Assert; /** - * + * * Streaming interfaces test case class using GZ compression codec, base class * of none and LZO compression classes. - * + * */ public class TestTFileStreams extends TestCase { @@ -356,7 +356,7 @@ public class TestTFileStreams extends TestCase { /** * Verify that the compressed data size is less than raw data size. - * + * * @throws IOException */ public void testFailureCompressionNotWorking() throws IOException { http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/763d14fc/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java ---------------------------------------------------------------------- diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java b/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java index 9d03f2a..a5da669 100644 --- a/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java +++ b/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java @@ -169,7 +169,7 @@ public class FunctionOperatorTest // Create local cluster LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(false); - + ((StramLocalCluster)lc).setExitCondition(new Callable<Boolean>() { @Override @@ -178,9 +178,9 @@ public class FunctionOperatorTest return TupleCount == NumTuples; } }); - + lc.run(5000); - + Assert.assertEquals(sum, 285); } @@ -224,7 +224,7 @@ public class FunctionOperatorTest // Create local cluster LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(false); - + ((StramLocalCluster)lc).setExitCondition(new Callable<Boolean>() { @Override @@ -235,7 +235,7 @@ public class FunctionOperatorTest }); lc.run(5000); - + Assert.assertEquals(sum, 39555); } @@ -285,7 +285,7 @@ public class FunctionOperatorTest // Create local cluster LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(false); - + ((StramLocalCluster)lc).setExitCondition(new Callable<Boolean>() { @Override
