Skip to content

Commit fee094e

Browse files
voonhousvoonhous
committed
refactor: Add Lombok Builder annotation to HoodieLogFormat
1 parent c177e2b commit fee094e

File tree

24 files changed

+616
-503
lines changed

24 files changed

+616
-503
lines changed

‎hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestHoodieLogFileCommand.java‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.apache.hudi.common.schema.HoodieSchema;
3737
import org.apache.hudi.common.schema.HoodieSchemaUtils;
3838
import org.apache.hudi.common.table.log.HoodieLogFormat;
39+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3940
import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
4041
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
4142
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
@@ -109,11 +110,14 @@ public void init() throws IOException, InterruptedException, URISyntaxException
109110
Files.createDirectories(Paths.get(partitionPath));
110111
storage = HoodieStorageUtils.getStorage(tablePath, storageConf());
111112

112-
try (HoodieLogFormat.Writer writer = HoodieLogFormat.newWriterBuilder()
113-
.onParentPath(new StoragePath(partitionPath))
113+
try (HoodieLogFormat.Writer writer = HoodieLogFormatWriter.builder()
114+
.withParentPath(new StoragePath(partitionPath))
114115
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
115-
.withFileId("test-log-fileid1").withInstantTime("100").withStorage(storage)
116-
.withSizeThreshold(1).build()) {
116+
.withLogFileId("test-log-fileid1")
117+
.withInstantTime("100")
118+
.withStorage(storage)
119+
.withSizeThreshold(1L)
120+
.build()) {
117121

118122
// write data to file
119123
List<HoodieRecord> records = SchemaTestUtil.generateTestRecords(0, 100).stream().map(HoodieAvroIndexedRecord::new).collect(Collectors.toList());
@@ -203,27 +207,21 @@ public void testShowLogFileRecordsWithMerge() throws IOException, InterruptedExc
203207
partitionPath = tablePath + StoragePath.SEPARATOR + HoodieTestCommitMetadataGenerator.DEFAULT_SECOND_PARTITION_PATH;
204208
Files.createDirectories(Paths.get(partitionPath));
205209

206-
HoodieLogFormat.Writer writer = null;
207-
try {
208-
// set little threshold to split file.
209-
writer =
210-
HoodieLogFormat.newWriterBuilder().onParentPath(new StoragePath(partitionPath))
211-
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
212-
.withFileId("test-log-fileid1").withInstantTime(INSTANT_TIME).withStorage(
213-
storage)
214-
.withSizeThreshold(500).build();
215-
210+
try (HoodieLogFormat.Writer writer = HoodieLogFormatWriter.builder()
211+
.withParentPath(new StoragePath(partitionPath))
212+
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
213+
.withLogFileId("test-log-fileid1")
214+
.withInstantTime(INSTANT_TIME)
215+
.withStorage(storage)
216+
.withSizeThreshold(500L) // set little threshold to split file.
217+
.build()) {
216218
SchemaTestUtil testUtil = new SchemaTestUtil();
217219
List<HoodieRecord> records1 = testUtil.generateHoodieTestRecords(0, 100).stream().map(HoodieAvroIndexedRecord::new).collect(Collectors.toList());
218220
Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
219221
header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, INSTANT_TIME);
220222
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
221223
HoodieAvroDataBlock dataBlock = new HoodieAvroDataBlock(records1, header, HoodieRecord.RECORD_KEY_METADATA_FIELD);
222224
writer.appendBlock(dataBlock);
223-
} finally {
224-
if (writer != null) {
225-
writer.close();
226-
}
227225
}
228226

229227
Object result = shell.evaluate(() -> "show logfile records --logFilePathPattern "

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v1/TimelineArchiverV1.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,8 @@
2929
import org.apache.hudi.common.model.HoodieRecord;
3030
import org.apache.hudi.common.model.HoodieTableType;
3131
import org.apache.hudi.common.table.HoodieTableMetaClient;
32-
import org.apache.hudi.common.table.log.HoodieLogFormat;
3332
import org.apache.hudi.common.table.log.HoodieLogFormat.Writer;
33+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3434
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
3535
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
3636
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
@@ -111,9 +111,12 @@ public TimelineArchiverV1(HoodieWriteConfig config, HoodieTable<T, I, K, O> tabl
111111
private Writer openWriter(StoragePath archivePath) {
112112
try {
113113
if (this.writer == null) {
114-
return HoodieLogFormat.newWriterBuilder().onParentPath(archivePath).withInstantTime("")
115-
.withFileId(archiveFilePath.getName()).withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION)
116-
.withStorage(metaClient.getStorage()).build();
114+
return HoodieLogFormatWriter.builder()
115+
.withParentPath(archivePath).withInstantTime("")
116+
.withLogFileId(archiveFilePath.getName())
117+
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION)
118+
.withStorage(metaClient.getStorage())
119+
.build();
117120
} else {
118121
return this.writer;
119122
}

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
import org.apache.hudi.common.schema.HoodieSchemaUtils;
4040
import org.apache.hudi.common.table.HoodieTableVersion;
4141
import org.apache.hudi.common.table.log.AppendResult;
42-
import org.apache.hudi.common.table.log.HoodieLogFormat.Writer;
42+
import org.apache.hudi.common.table.log.HoodieLogFormat;
4343
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
4444
import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
4545
import org.apache.hudi.common.table.log.block.HoodieHFileDataBlock;
@@ -105,7 +105,7 @@ public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O
105105
// Incoming records to be written to logs.
106106
protected Iterator<HoodieRecord<T>> recordItr;
107107
// Writer to log into the file group's latest slice.
108-
protected Writer writer;
108+
protected HoodieLogFormat.Writer writer;
109109

110110
protected final List<WriteStatus> statuses;
111111
// Total number of records written during appending

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.apache.hudi.common.table.HoodieTableMetaClient;
3838
import org.apache.hudi.common.table.HoodieTableVersion;
3939
import org.apache.hudi.common.table.log.HoodieLogFormat;
40+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
4041
import org.apache.hudi.common.table.log.LogFileCreationCallback;
4142
import org.apache.hudi.common.table.read.DeleteContext;
4243
import org.apache.hudi.common.util.ConfigUtils;
@@ -288,11 +289,11 @@ protected HoodieLogFormat.Writer createLogWriter(String instantTime, Option<File
288289
protected HoodieLogFormat.Writer createLogWriter(String instantTime, String fileSuffix, Option<FileSlice> fileSliceOpt) {
289290
try {
290291
if (config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)) {
291-
return HoodieLogFormat.newWriterBuilder()
292-
.onParentPath(FSUtils.constructAbsolutePath(hoodieTable.getMetaClient().getBasePath(), partitionPath))
293-
.withFileId(fileId)
292+
return HoodieLogFormatWriter.builder()
293+
.withParentPath(FSUtils.constructAbsolutePath(hoodieTable.getMetaClient().getBasePath(), partitionPath))
294+
.withLogFileId(fileId)
294295
.withInstantTime(instantTime)
295-
.withFileSize(0L)
296+
.withFileLen(0L)
296297
.withSizeThreshold(config.getLogFileMaxSize())
297298
.withStorage(storage)
298299
.withLogWriteToken(writeToken)
@@ -305,12 +306,12 @@ protected HoodieLogFormat.Writer createLogWriter(String instantTime, String file
305306
Option<HoodieLogFile> latestLogFile = fileSliceOpt.isPresent()
306307
? fileSliceOpt.get().getLatestLogFile()
307308
: Option.empty();
308-
return HoodieLogFormat.newWriterBuilder()
309-
.onParentPath(FSUtils.constructAbsolutePath(hoodieTable.getMetaClient().getBasePath(), partitionPath))
310-
.withFileId(fileId)
309+
return HoodieLogFormatWriter.builder()
310+
.withParentPath(FSUtils.constructAbsolutePath(hoodieTable.getMetaClient().getBasePath(), partitionPath))
311+
.withLogFileId(fileId)
311312
.withInstantTime(instantTime)
312313
.withLogVersion(latestLogFile.map(HoodieLogFile::getLogVersion).orElse(HoodieLogFile.LOGFILE_BASE_VERSION))
313-
.withFileSize(latestLogFile.map(HoodieLogFile::getFileSize).orElse(0L))
314+
.withFileLen(latestLogFile.map(HoodieLogFile::getFileSize).orElse(0L))
314315
.withSizeThreshold(config.getLogFileMaxSize())
315316
.withStorage(storage)
316317
.withLogWriteToken(latestLogFile.map(HoodieLogFile::getLogWriteToken).orElse(writeToken))

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@
5555
import org.apache.hudi.common.schema.HoodieSchemaUtils;
5656
import org.apache.hudi.common.table.HoodieTableMetaClient;
5757
import org.apache.hudi.common.table.log.HoodieLogFormat;
58+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
5859
import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
5960
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
6061
import org.apache.hudi.common.table.read.HoodieFileGroupReader;
@@ -1073,12 +1074,12 @@ private void initializeFileGroups(HoodieTableMetaClient dataMetaClient, Metadata
10731074

10741075
final HoodieDeleteBlock block = new HoodieDeleteBlock(Collections.emptyList(), blockHeader);
10751076

1076-
try (HoodieLogFormat.Writer writer = HoodieLogFormat.newWriterBuilder()
1077-
.onParentPath(FSUtils.constructAbsolutePath(metadataWriteConfig.getBasePath(), relativePartitionPath))
1078-
.withFileId(fileGroupFileId)
1077+
try (HoodieLogFormat.Writer writer = HoodieLogFormatWriter.builder()
1078+
.withParentPath(FSUtils.constructAbsolutePath(metadataWriteConfig.getBasePath(), relativePartitionPath))
1079+
.withLogFileId(fileGroupFileId)
10791080
.withInstantTime(instantTime)
10801081
.withLogVersion(HoodieLogFile.LOGFILE_BASE_VERSION)
1081-
.withFileSize(0L)
1082+
.withFileLen(0L)
10821083
.withSizeThreshold(metadataWriteConfig.getLogFileMaxSize())
10831084
.withStorage(dataMetaClient.getStorage())
10841085
.withLogWriteToken(HoodieLogFormat.DEFAULT_WRITE_TOKEN)

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelper.java‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.hudi.common.table.HoodieTableMetaClient;
2929
import org.apache.hudi.common.table.HoodieTableVersion;
3030
import org.apache.hudi.common.table.log.HoodieLogFormat;
31+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3132
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
3233
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
3334
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -137,16 +138,17 @@ List<Pair<String, HoodieRollbackStat>> maybeDeleteAndCollectStats(HoodieEngineCo
137138
String fileId = rollbackRequest.getFileId();
138139
HoodieTableVersion tableVersion = metaClient.getTableConfig().getTableVersion();
139140

140-
writer = HoodieLogFormat.newWriterBuilder()
141-
.onParentPath(FSUtils.constructAbsolutePath(metaClient.getBasePath(), rollbackRequest.getPartitionPath()))
142-
.withFileId(fileId)
141+
writer = HoodieLogFormatWriter.builder()
142+
.withParentPath(FSUtils.constructAbsolutePath(metaClient.getBasePath(), rollbackRequest.getPartitionPath()))
143+
.withLogFileId(fileId)
143144
.withLogWriteToken(CommonClientUtils.generateWriteToken(taskContextSupplier))
144145
.withInstantTime(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT)
145-
? instantToRollback.requestedTime() : rollbackRequest.getLatestBaseInstant()
146-
)
146+
? instantToRollback.requestedTime() : rollbackRequest.getLatestBaseInstant()
147+
)
147148
.withStorage(metaClient.getStorage())
148149
.withTableVersion(tableVersion)
149-
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).build();
150+
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
151+
.build();
150152

151153
// generate metadata
152154
if (doDelete) {

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackHelperV1.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.hudi.common.model.IOType;
3131
import org.apache.hudi.common.table.HoodieTableVersion;
3232
import org.apache.hudi.common.table.log.HoodieLogFormat;
33+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3334
import org.apache.hudi.common.table.log.LogFileCreationCallback;
3435
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
3536
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
@@ -206,9 +207,9 @@ List<Pair<String, HoodieRollbackStat>> maybeDeleteAndCollectStats(HoodieEngineCo
206207
// Let's emit markers for rollback as well. markers are emitted under rollback instant time.
207208
WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), table, instantTime);
208209

209-
writer = HoodieLogFormat.newWriterBuilder()
210-
.onParentPath(FSUtils.constructAbsolutePath(metaClient.getBasePath(), partitionPath))
211-
.withFileId(fileId)
210+
writer = HoodieLogFormatWriter.builder()
211+
.withParentPath(FSUtils.constructAbsolutePath(metaClient.getBasePath(), partitionPath))
212+
.withLogFileId(fileId)
212213
.withLogWriteToken(CommonClientUtils.generateWriteToken(taskContextSupplier))
213214
.withInstantTime(tableVersion.greaterThanOrEquals(HoodieTableVersion.EIGHT)
214215
? instantToRollback.requestedTime() : rollbackRequest.getLatestBaseInstant()

‎hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestLegacyArchivedMetaEntryReader.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.hudi.common.model.WriteOperationType;
2828
import org.apache.hudi.common.table.HoodieTableMetaClient;
2929
import org.apache.hudi.common.table.log.HoodieLogFormat;
30+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3031
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
3132
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
3233
import org.apache.hudi.common.table.timeline.ActiveAction;
@@ -101,10 +102,13 @@ private void prepareLegacyArchivedTimeline(HoodieTableMetaClient metaClient) thr
101102

102103
private HoodieLogFormat.Writer openWriter(HoodieTableMetaClient metaClient) {
103104
try {
104-
return HoodieLogFormat.newWriterBuilder()
105-
.onParentPath(metaClient.getArchivePath())
106-
.withFileId("commits").withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION)
107-
.withStorage(metaClient.getStorage()).withInstantTime("").build();
105+
return HoodieLogFormatWriter.builder()
106+
.withParentPath(metaClient.getArchivePath())
107+
.withLogFileId("commits")
108+
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION)
109+
.withStorage(metaClient.getStorage())
110+
.withInstantTime("")
111+
.build();
108112
} catch (IOException e) {
109113
throw new HoodieException("Unable to initialize HoodieLogFormat writer", e);
110114
}

‎hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/HoodieWriteableTestTable.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.hudi.common.table.HoodieTableConfig;
3333
import org.apache.hudi.common.table.HoodieTableMetaClient;
3434
import org.apache.hudi.common.table.log.HoodieLogFormat;
35+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3536
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
3637
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
3738
import org.apache.hudi.common.testutils.FileCreateUtilsLegacy;
@@ -168,10 +169,13 @@ public Map<String, List<HoodieLogFile>> withLogAppends(String partition, String
168169
}
169170

170171
private Pair<String, HoodieLogFile> appendRecordsToLogFile(String partitionPath, String fileId, List<HoodieRecord> records) throws Exception {
171-
try (HoodieLogFormat.Writer logWriter = HoodieLogFormat.newWriterBuilder()
172-
.onParentPath(new StoragePath(basePath, partitionPath))
173-
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).withFileId(fileId)
174-
.withInstantTime(currentInstantTime).withStorage(storage).build()) {
172+
try (HoodieLogFormat.Writer logWriter = HoodieLogFormatWriter.builder()
173+
.withParentPath(new StoragePath(basePath, partitionPath))
174+
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
175+
.withLogFileId(fileId)
176+
.withInstantTime(currentInstantTime)
177+
.withStorage(storage)
178+
.build()) {
175179
Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
176180
header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, currentInstantTime);
177181
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());

‎hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/testutils/HoodieFlinkWriteableTestTable.java‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.apache.hudi.common.model.HoodieRecordPayload;
3232
import org.apache.hudi.common.table.HoodieTableMetaClient;
3333
import org.apache.hudi.common.table.log.HoodieLogFormat;
34+
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
3435
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
3536
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
3637
import org.apache.hudi.common.util.collection.Pair;
@@ -136,10 +137,13 @@ public Map<String, List<HoodieLogFile>> withLogAppends(List<HoodieRecord> record
136137
private Pair<String, HoodieLogFile> appendRecordsToLogFile(List<HoodieRecord> groupedRecords) throws Exception {
137138
String partitionPath = groupedRecords.get(0).getPartitionPath();
138139
HoodieRecordLocation location = groupedRecords.get(0).getCurrentLocation();
139-
try (HoodieLogFormat.Writer logWriter = HoodieLogFormat.newWriterBuilder()
140-
.onParentPath(new StoragePath(basePath, partitionPath))
141-
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).withFileId(location.getFileId())
142-
.withInstantTime(location.getInstantTime()).withStorage(storage).build()) {
140+
try (HoodieLogFormat.Writer logWriter = HoodieLogFormatWriter.builder()
141+
.withParentPath(new StoragePath(basePath, partitionPath))
142+
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
143+
.withLogFileId(location.getFileId())
144+
.withInstantTime(location.getInstantTime())
145+
.withStorage(storage)
146+
.build()) {
143147
Map<HeaderMetadataType, String> header = new java.util.HashMap<>();
144148
header.put(HeaderMetadataType.INSTANT_TIME, location.getInstantTime());
145149
header.put(HeaderMetadataType.SCHEMA, schema.toString());
@@ -150,7 +154,7 @@ private Pair<String, HoodieLogFile> appendRecordsToLogFile(List<HoodieRecord> gr
150154
HoodieAvroUtils.addHoodieKeyToRecord(val, r.getRecordKey(), r.getPartitionPath(), "");
151155
return (IndexedRecord) val;
152156
} catch (IOException e) {
153-
log.warn("Failed to convert record " + r.toString(), e);
157+
log.warn("Failed to convert record {}", r, e);
154158
return null;
155159
}
156160
}).map(HoodieAvroIndexedRecord::new).collect(Collectors.toList()), header, HoodieRecord.RECORD_KEY_METADATA_FIELD));

0 commit comments

Comments
 (0)