ProtoParquetWriterWithOffset implements CloseableWriter<MESSAGE_KIND> { @Override public Path close() throws IOException { if (latestOffset == null) { final String additionalInfo = String.format(" Date = %s, Temp file = %s", dayStartTime.format(DateTimeFormatter.ISO_LOCAL_DATE_TIME), temporaryHdfsPath.toUri()); throw new IOException(String.format("Trying to write a zero-sized file, please fix (%s)", additionalInfo)); } closeWriter(); final Optional<Path> lastestExistingFinalPath = getLastestExistingFinalPath(); long lastIndex = lastestExistingFinalPath.map(path -> fileNamer.getIndex(path.getName()) + 1).orElse(1L); final Path finalPath = new Path(finalHdfsDir, fileNamer.computePath(dayStartTime, lastIndex, latestOffset.getPartition())); FileSystemUtils.ensureDirectoriesExist(Collections.singleton(finalPath.getParent()), fs); if (lastestExistingFinalPath.isPresent()) { long blockSize = fs.getFileStatus(lastestExistingFinalPath.get()).getLen(); if (blockSize > fsBlockSize) { moveToFinalPath(temporaryHdfsPath, finalPath); } else { mergeToFinalPath(lastestExistingFinalPath.get(), finalPath); } } else { moveToFinalPath(temporaryHdfsPath, finalPath); } PrometheusMetrics.latestCommittedOffsetGauge(eventName, latestOffset.getPartition()).set(latestOffset.getOffset()); PrometheusMetrics.latestCommittedTimestampGauge(eventName, latestOffset.getPartition()).set(latestTimestamp); return finalPath; } ProtoParquetWriterWithOffset(ParquetWriter<MESSAGE_KIND> writer, Path temporaryHdfsPath,
                                        Path finalHdfsDir, FileSystem fs, OffsetComputer fileNamer,
                                        LocalDateTime dayStartTime, String eventName,
                                        BiConsumer<String, String> protoMetadataWriter, int partition); @Override Path close(); @Override void write(long timestamp, MESSAGE_KIND msg, Offset offset); ParquetWriter<MESSAGE_KIND> getWriter(); Path getFinalHdfsDir(); LocalDateTime getDayStartTime(); String getEventName(); static final String LATEST_TIMESTAMP_META_KEY; }
@Test(expected = IOException.class) public void closeWithNoEvent() throws IOException { final ProtoParquetWriter<Message> writerMock = mock(ProtoParquetWriter.class); final BiConsumer<String, String> protoMetadataWriter = mock(BiConsumer.class); final ProtoParquetWriterWithOffset parquetWriter = new ProtoParquetWriterWithOffset<>(writerMock, createTmpFile(), finalPath, localFs, new HdfsOffsetComputer(localFs, finalPath, 2), LocalDateTime.MIN, "ignored", protoMetadataWriter, 0); parquetWriter.close(); }