S3Reconciler extends AbstractReconciler<FSRecordCompactionOperator.OutputMetaData, FSRecordCompactionOperator.OutputMetaData> { @Override public void endWindow() { while (doneTuples.peek() != null) { FSRecordCompactionOperator.OutputMetaData metaData = doneTuples.poll(); removeIntermediateFiles(metaData); if (outputPort.isConnected()) { metaData.setPath(getDirectoryName() + Path.SEPARATOR + metaData.getFileName()); outputPort.emit(metaData); } } }  @Override void setup(Context.OperatorContext context); @Override void endWindow(); String getAccessKey(); void setAccessKey(@NotNull String accessKey); String getSecretKey(); void setSecretKey(@NotNull String secretKey); String getBucketName(); void setBucketName(@NotNull String bucketName); String getDirectoryName(); void setDirectoryName(@NotNull String directoryName); String getRegion(); void setRegion(String region); final transient DefaultOutputPort<FSRecordCompactionOperator.OutputMetaData> outputPort; }
@Test public void verifyS3ReconclierOutputTuple() throws Exception { String fileName = "s3-compaction_1.0"; String path = testMeta.outputPath + Path.SEPARATOR + fileName; long size = 80; File file = new File(path); File tmpFile = new File(path + "." + System.currentTimeMillis() + ".tmp"); StringBuffer sb = new StringBuffer(); for (int i = 0; i < 10; i++) { sb.append("Record" + i + "\n"); if (i == 5) { FileUtils.write(tmpFile, sb.toString()); } } FileUtils.write(file, sb.toString()); testMeta.sink = new CollectorTestSink<Object>(); testMeta.underTest.outputPort.setSink(testMeta.sink); FSRecordCompactionOperator.OutputMetaData outputMetaData = new FSRecordCompactionOperator.OutputMetaData(path, fileName, size); testMeta.underTest.beginWindow(0); testMeta.underTest.input.process(outputMetaData); testMeta.underTest.endWindow(); for (int i = 1; i < 60; i++) { testMeta.underTest.beginWindow(i); testMeta.underTest.endWindow(); } testMeta.underTest.committed(59); testMeta.sink.waitForResultCount(1, 12000); for (int i = 60; i < 70; i++) { testMeta.underTest.beginWindow(i); Thread.sleep(10); testMeta.underTest.endWindow(); } Assert.assertEquals(1, testMeta.sink.getCount(false)); }
@Test public void testFileClearing() throws Exception { String fileName = "s3-compaction_1.0"; String path = testMeta.outputPath + Path.SEPARATOR + fileName; long size = 80; File file = new File(path); File tmpFile = new File(path + "." + System.currentTimeMillis() + ".tmp"); StringBuffer sb = new StringBuffer(); for (int i = 0; i < 10; i++) { sb.append("Record" + i + "\n"); if (i == 5) { FileUtils.write(tmpFile, sb.toString()); } } FileUtils.write(file, sb.toString()); FSRecordCompactionOperator.OutputMetaData outputMetaData = new FSRecordCompactionOperator.OutputMetaData(path, fileName, size); testMeta.underTest.beginWindow(0); testMeta.underTest.input.process(outputMetaData); testMeta.underTest.endWindow(); for (int i = 1; i < 60; i++) { testMeta.underTest.beginWindow(i); testMeta.underTest.endWindow(); } testMeta.underTest.committed(59); for (int i = 60; i < 70; i++) { testMeta.underTest.beginWindow(i); Thread.sleep(10); testMeta.underTest.endWindow(); } Collection<File> files = FileUtils.listFiles(new File(testMeta.outputPath), null, true); Assert.assertEquals(0, files.size()); }