This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new fc91325b6381 refactor(flink): use SLF4J parameterized logging instead
of string concatenation (#19156)
fc91325b6381 is described below
commit fc91325b63812851d42b5fc66b3738564c0225f2
Author: voonhous <[email protected]>
AuthorDate: Sat Jul 4 11:44:30 2026 +0800
refactor(flink): use SLF4J parameterized logging instead of string
concatenation (#19156)
* refactor(flink): use SLF4J parameterized logging instead of string
concatenation
Convert `+`-concatenated log messages to SLF4J `{}` placeholders across
hudi-flink-datasource (hudi-flink and the flink1.18.x-2.1.x version
modules).
The rewrite is behaviour-preserving: each concatenated expression becomes a
placeholder argument in order, and a trailing throwable is kept as the last
argument so its stack trace is still recorded. This also avoids building the
message string eagerly when the log level is disabled.
* refactor(flink): reword "Error ?" shutdown log to "Error:"
With `{}` templating, "Error ?{}" reads oddly (the `?` looks like a stray
ternary fragment). Reword to "Error: {}" in both the clustering job and
compactor graceful-shutdown logs. Addresses review nit on #19156.
---
.../main/java/org/apache/hudi/configuration/OptionsInference.java | 3 +--
.../org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java | 2 +-
.../java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java | 6 +++---
.../org/apache/hudi/sink/clustering/ClusteringCommitSink.java | 2 +-
.../org/apache/hudi/sink/clustering/ClusteringPlanOperator.java | 6 +++---
.../org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java | 6 +++---
.../java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java | 8 ++++----
.../java/org/apache/hudi/sink/partitioner/BucketAssigner.java | 2 +-
.../apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java | 2 +-
.../org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java | 2 +-
.../main/java/org/apache/hudi/source/IncrementalInputSplits.java | 7 +++----
.../src/main/java/org/apache/hudi/table/HoodieTableSource.java | 2 +-
.../java/org/apache/hudi/table/lookup/HoodieLookupFunction.java | 2 +-
.../hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java | 2 +-
.../src/main/java/org/apache/hudi/util/ClusteringUtil.java | 2 +-
.../src/main/java/org/apache/hudi/util/CompactionUtil.java | 4 ++--
.../src/main/java/org/apache/hudi/util/ViewStorageProperties.java | 2 +-
.../format/cow/vector/reader/BaseVectorizedColumnReader.java | 7 +------
.../format/cow/vector/reader/BaseVectorizedColumnReader.java | 7 +------
.../format/cow/vector/reader/BaseVectorizedColumnReader.java | 7 +------
.../format/cow/vector/reader/BaseVectorizedColumnReader.java | 7 +------
.../format/cow/vector/reader/BaseVectorizedColumnReader.java | 7 +------
22 files changed, 34 insertions(+), 61 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsInference.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsInference.java
index 6b5fddfacfab..70a44b1df7c2 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsInference.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsInference.java
@@ -149,8 +149,7 @@ public class OptionsInference {
conf.set(FlinkOptions.BUCKET_INDEX_PARTITION_EXPRESSIONS,
hashingConfig.getExpressions());
conf.set(FlinkOptions.BUCKET_INDEX_PARTITION_RULE,
hashingConfig.getRule());
conf.set(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS,
hashingConfig.getDefaultBucketNumber());
- log.info("Loaded Latest Hashing Config " + hashingConfig
- + ". Reset hoodie.bucket.index.num.buckets to " +
hashingConfig.getDefaultBucketNumber());
+ log.info("Loaded Latest Hashing Config {}. Reset
hoodie.bucket.index.num.buckets to {}", hashingConfig,
hashingConfig.getDefaultBucketNumber());
}
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
index a6eee1caf517..358686f9f6ff 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketBulkInsertWriterHelper.java
@@ -64,7 +64,7 @@ public class BucketBulkInsertWriterHelper extends
BulkInsertWriterHelper {
String partitionPath = keyGen.getPartitionPath(record);
String fileId = tuple.getString(0).toString();
if ((lastFileId == null) || !lastFileId.equals(fileId)) {
- log.info("Creating new file for partition path " + partitionPath);
+ log.info("Creating new file for partition path {}", partitionPath);
handle = getRowCreateHandle(partitionPath, fileId);
lastFileId = fileId;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
index 14e7ea59191c..30a19e1423aa 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriterHelper.java
@@ -151,7 +151,7 @@ public class BulkInsertWriterHelper {
close();
}
- log.info("Creating new file for partition path " + partitionPath);
+ log.info("Creating new file for partition path {}", partitionPath);
writeMetrics.ifPresent(FlinkStreamWriteMetrics::startHandleCreation);
HoodieRowDataCreateHandle rowCreateHandle = new
HoodieRowDataCreateHandle(hoodieTable, writeConfig, partitionPath,
getNextFileId(),
instantTime, taskPartitionId, totalSubtaskNum, taskEpochId,
writerSchema, preserveHoodieMetadata, isAppendMode && !populateMetaFields);
@@ -161,7 +161,7 @@ public class BulkInsertWriterHelper {
} else if (!handles.get(partitionPath).canWrite()) {
// even if there is a handle to the partition path, it could have
reached its max size threshold. So, we close the handle here and
// create a new one.
- log.info("Rolling max-size file for partition path " + partitionPath);
+ log.info("Rolling max-size file for partition path {}", partitionPath);
writeStatusList.add(closeWriteHandle(handles.remove(partitionPath)));
HoodieRowDataCreateHandle rowCreateHandle =
createWriteHandle(partitionPath);
handles.put(partitionPath, rowCreateHandle);
@@ -179,7 +179,7 @@ public class BulkInsertWriterHelper {
allOf(handles.values().stream()
.map(rowCreateHandle -> CompletableFuture.supplyAsync(() -> {
try {
- log.info("Closing bulk insert file " +
rowCreateHandle.getFileName());
+ log.info("Closing bulk insert file {}",
rowCreateHandle.getFileName());
return rowCreateHandle.close();
} catch (IOException e) {
throw new HoodieIOException("IOE during rowCreateHandle.close()",
e);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
index c4c1efc21c0a..2bc668d2afa7 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
@@ -175,7 +175,7 @@ public class ClusteringCommitSink extends
CleanFunction<ClusteringCommitEvent> {
doCommit(instant, clusteringPlan, events);
} catch (Throwable throwable) {
// make it fail-safe
- log.error("Error while committing clustering instant: " + instant,
throwable);
+ log.error("Error while committing clustering instant: {}", instant,
throwable);
} finally {
// reset the status
reset(instant);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
index 609abbdaf862..49340fc5f465 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
@@ -117,7 +117,7 @@ public class ClusteringPlanOperator extends
AbstractStreamOperator<ClusteringPla
scheduleClustering(table, checkpointId);
} catch (Throwable throwable) {
// make it fail-safe
- log.error("Error while scheduling clustering plan for checkpoint: " +
checkpointId, throwable);
+ log.error("Error while scheduling clustering plan for checkpoint: {}",
checkpointId, throwable);
}
}
@@ -135,7 +135,7 @@ public class ClusteringPlanOperator extends
AbstractStreamOperator<ClusteringPla
if (!firstRequested.isPresent()) {
// do nothing.
- log.info("No clustering plan for checkpoint " + checkpointId);
+ log.info("No clustering plan for checkpoint {}", checkpointId);
return;
}
@@ -158,7 +158,7 @@ public class ClusteringPlanOperator extends
AbstractStreamOperator<ClusteringPla
if (clusteringPlan == null || (clusteringPlan.getInputGroups() == null)
|| (clusteringPlan.getInputGroups().isEmpty())) {
// do nothing.
- log.info("Empty clustering plan for instant " + clusteringInstantTime);
+ log.info("Empty clustering plan for instant {}", clusteringInstantTime);
} else {
// Mark instant as clustering inflight
ClusteringUtils.transitionClusteringOrReplaceRequestedToInflight(clusteringInstant,
Option.empty(), table.getActiveTimeline());
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
index d2f6419515ac..af9f011715b0 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
@@ -339,7 +339,7 @@ public class HoodieFlinkClusteringJob {
Option<HoodieInstant> inflightInstantOpt =
ClusteringUtils.getInflightClusteringInstant(clusteringInstant.requestedTime(),
table.getActiveTimeline(), table.getInstantGenerator());
if (inflightInstantOpt.isPresent()) {
- LOG.info("Rollback inflight clustering instant: [" + clusteringInstant
+ "]");
+ LOG.info("Rollback inflight clustering instant: [{}]",
clusteringInstant);
table.rollbackInflightClustering(inflightInstantOpt.get(),
commitToRollback ->
writeClient.getTableServiceClient().getPendingRollbackInfo(table.getMetaClient(),
commitToRollback, false),
writeClient.getTransactionManager());
@@ -362,7 +362,7 @@ public class HoodieFlinkClusteringJob {
if (clusteringPlan == null || (clusteringPlan.getInputGroups() == null)
|| (clusteringPlan.getInputGroups().isEmpty())) {
// no clustering plan, do nothing and return.
- LOG.info("No clustering plan for instant " +
clusteringInstant.requestedTime());
+ LOG.info("No clustering plan for instant {}",
clusteringInstant.requestedTime());
return;
}
@@ -418,7 +418,7 @@ public class HoodieFlinkClusteringJob {
* Shutdown async services like compaction/clustering as DeltaSync is
shutdown.
*/
public void shutdownAsyncService(boolean error) {
- LOG.info("Gracefully shutting down clustering job. Error ?" + error);
+ LOG.info("Gracefully shutting down clustering job. Error: {}", error);
executor.shutdown();
writeClient.close();
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
index d6567c8c08ff..324dfd6917be 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
@@ -300,7 +300,7 @@ public class HoodieFlinkCompactor {
compactionInstantTimes.forEach(timestamp -> {
HoodieInstant inflightInstant =
table.getInstantGenerator().getCompactionInflightInstant(timestamp);
if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
- LOG.info("Rollback inflight compaction instant: [" + timestamp +
"]");
+ LOG.info("Rollback inflight compaction instant: [{}]", timestamp);
table.rollbackInflightCompaction(inflightInstant,
writeClient.getTransactionManager());
table.getMetaClient().reloadActiveTimeline();
}
@@ -322,7 +322,7 @@ public class HoodieFlinkCompactor {
if (compactionPlans.isEmpty()) {
// No compaction plan, do nothing and return.
- LOG.info("No compaction plan for instant " + String.join(",",
compactionInstantTimes));
+ LOG.info("No compaction plan for instant {}", String.join(",",
compactionInstantTimes));
return;
}
@@ -336,7 +336,7 @@ public class HoodieFlinkCompactor {
? totalOperations
: Math.min(conf.get(FlinkOptions.COMPACTION_TASKS), totalOperations);
- LOG.info("Start to compaction for instant " + compactionInstantTimes);
+ LOG.info("Start to compaction for instant {}", compactionInstantTimes);
// Mark instant as compaction inflight
for (HoodieInstant instant : instants) {
@@ -367,7 +367,7 @@ public class HoodieFlinkCompactor {
* Shutdown async services like compaction/clustering as DeltaSync is
shutdown.
*/
public void shutdownAsyncService(boolean error) {
- LOG.info("Gracefully shutting down compactor. Error ?" + error);
+ LOG.info("Gracefully shutting down compactor. Error: {}", error);
executor.shutdown();
writeClient.close();
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
index fe848d7bcc20..0544f700d15d 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
@@ -177,7 +177,7 @@ public class BucketAssigner implements AutoCloseable {
}
List<SmallFile> smallFiles =
smallFilesOfThisTask(writeProfile.getSmallFiles(partitionPath));
if (smallFiles.size() > 0) {
- log.info("For partitionPath : " + partitionPath + " Small Files => " +
smallFiles);
+ log.info("For partitionPath : {} Small Files => {}", partitionPath,
smallFiles);
SmallFileAssignState[] states = smallFiles.stream()
.map(smallFile -> new
SmallFileAssignState(config.getParquetMaxFileSize(), smallFile,
writeProfile.getAvgSize()))
.toArray(SmallFileAssignState[]::new);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
index b6b087c62a1f..4f31ee877f7a 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/clustering/ClusteringCommitSinkV2.java
@@ -185,7 +185,7 @@ public class ClusteringCommitSinkV2 extends
CleanFunctionV2<ClusteringCommitEven
doCommit(instant, clusteringPlan, events);
} catch (Throwable throwable) {
// make it fail-safe
- log.error("Error while committing clustering instant: " + instant,
throwable);
+ log.error("Error while committing clustering instant: {}", instant,
throwable);
} finally {
// reset the status
reset(instant);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
index 848cb10d0c77..7e81112bb0b6 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/v2/compact/CompactionCommitSinkV2.java
@@ -175,7 +175,7 @@ public class CompactionCommitSinkV2 extends
CleanFunctionV2<CompactionCommitEven
doCommit(instant, events);
} catch (Throwable throwable) {
// make it fail-safe
- log.error("Error while committing compaction instant: " + instant,
throwable);
+ log.error("Error while committing compaction instant: {}", instant,
throwable);
this.compactionMetrics.markCompactionRolledBack();
} finally {
// reset the status
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
index 344b24bc10c4..68e81271d6c8 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
@@ -147,7 +147,7 @@ public class IncrementalInputSplits implements Serializable
{
IncrementalQueryAnalyzer.QueryContext analyzingResult = analyzer.analyze();
if (analyzingResult.isEmpty()) {
- log.info("No new instant found for the table under path " + path + ",
skip reading");
+ log.info("No new instant found for the table under path {}, skip
reading", path);
return Result.EMPTY;
}
final HoodieTimeline commitTimeline = analyzingResult.getActiveTimeline();
@@ -276,7 +276,7 @@ public class IncrementalInputSplits implements Serializable
{
IncrementalQueryAnalyzer.QueryContext queryContext = analyzer.analyze();
if (queryContext.isEmpty()) {
- log.info("No new instant found for the table under path " + path + ",
skip reading");
+ log.info("No new instant found for the table under path {}, skip
reading", path);
return Result.EMPTY;
}
@@ -501,8 +501,7 @@ public class IncrementalInputSplits implements Serializable
{
double total = partitions.size();
double selectedNum = selectedPartitions.size();
double percentPruned = total == 0 ? 0 : (1 - selectedNum / total) * 100;
- log.info("Selected " + selectedNum + " partitions out of " + total
- + ", pruned " + percentPruned + "% partitions.");
+ log.info("Selected {} partitions out of {}, pruned {}% partitions.",
selectedNum, total, percentPruned);
return selectedPartitions;
}
return partitions;
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
index 76d4bae29a88..c3ba129eaa27 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
@@ -471,7 +471,7 @@ public class HoodieTableSource extends FileIndexReader
implements
}
StringJoiner joiner = new StringJoiner(" and ");
partitionFilters.forEach(f -> joiner.add(f.asSummaryString()));
- log.info("Partition pruner for hoodie source, condition is:\n" + joiner);
+ log.info("Partition pruner for hoodie source, condition is:\n{}", joiner);
List<ExpressionEvaluators.Evaluator> evaluators =
ExpressionEvaluators.fromExpression(partitionFilters);
List<DataType> partitionTypes = this.partitionKeys.stream().map(name ->
this.schema.getColumn(name).orElseThrow(() -> new
HoodieValidationException("Field " + name + " does not exist")))
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
index 876822246397..10c47b2fe1c9 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
@@ -145,7 +145,7 @@ public class HoodieLookupFunction extends LookupFunction
implements Serializable
// Determine whether to reload data by comparing instant
if (latestCommitInstant.get().equals(currentCommit)) {
scheduleNextLoad();
- log.info("Ignore loading data because the commit instant " +
currentCommit + " has not changed.");
+ log.info("Ignore loading data because the commit instant {} has not
changed.", currentCommit);
return;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
index 0a18ac029e92..c2b1c1d3a14f 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
@@ -138,7 +138,7 @@ public class ClientIds implements AutoCloseable,
Serializable {
}
} catch (IOException e) {
// if any exception happens, just return false.
- log.error("Check heartbeat file existence error: " + path);
+ log.error("Check heartbeat file existence error: {}", path);
}
return false;
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
index 41dee1cd1b07..e0d3a9d50018 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
@@ -87,7 +87,7 @@ public class ClusteringUtil {
.filter(instant -> instant.getState() == HoodieInstant.State.INFLIGHT)
.collect(Collectors.toList());
inflightInstants.forEach(inflightInstant -> {
- log.info("Rollback the inflight clustering instant: " + inflightInstant
+ " for failover");
+ log.info("Rollback the inflight clustering instant: {} for failover",
inflightInstant);
table.rollbackInflightClustering(inflightInstant,
commitToRollback ->
writeClient.getTableServiceClient().getPendingRollbackInfo(table.getMetaClient(),
commitToRollback, false),
writeClient.getTransactionManager());
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/CompactionUtil.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/CompactionUtil.java
index 8e5456b8f1a6..39b54fefb2e2 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/CompactionUtil.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/CompactionUtil.java
@@ -245,7 +245,7 @@ public class CompactionUtil {
.filter(instant ->
instant.getState() == HoodieInstant.State.INFLIGHT);
inflightCompactionTimeline.getInstants().forEach(inflightInstant -> {
- log.info("Rollback the inflight compaction instant: " + inflightInstant
+ " for failover");
+ log.info("Rollback the inflight compaction instant: {} for failover",
inflightInstant);
table.rollbackInflightCompaction(inflightInstant, commitToRollback ->
writeClient.getTableServiceClient().getPendingRollbackInfo(table.getMetaClient(),
commitToRollback, false),
writeClient.getTransactionManager());
table.getMetaClient().reloadActiveTimeline();
@@ -269,7 +269,7 @@ public class CompactionUtil {
String currentTime =
HoodieInstantTimeGenerator.getCurrentInstantTimeStr();
int timeout = conf.get(FlinkOptions.COMPACTION_TIMEOUT_SECONDS);
if (StreamerUtil.instantTimeDiffSeconds(currentTime,
instant.requestedTime()) >= timeout) {
- log.info("Rollback the inflight compaction instant: " + instant + "
for timeout(" + timeout + "s)");
+ log.info("Rollback the inflight compaction instant: {} for
timeout({}s)", instant, timeout);
try (TransactionManager transactionManager = new
TransactionManager(table.getConfig(), table.getStorage())) {
table.rollbackInflightCompaction(instant, transactionManager);
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
index e0f715c1c85a..94c2104d44d1 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
@@ -68,7 +68,7 @@ public class ViewStorageProperties {
*/
public static FileSystemViewStorageConfig loadFromProperties(String
basePath, Configuration conf) {
Path propertyPath = getPropertiesFilePath(basePath,
conf.get(FlinkOptions.WRITE_CLIENT_ID));
- log.info("Loading filesystem view storage properties from " +
propertyPath);
+ log.info("Loading filesystem view storage properties from {}",
propertyPath);
FileSystem fs = HadoopFSUtils.getFs(basePath,
HadoopConfigurations.getHadoopConf(conf));
Properties props = new Properties();
try {
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
index 7c9fd994a0c2..700d7505fbc7 100644
---
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
@@ -226,12 +226,7 @@ public abstract class BaseVectorizedColumnReader
implements ColumnReader<Writabl
this.definitionLevelColumn =
newRLEIterator(descriptor.getMaxDefinitionLevel(),
page.getDefinitionLevels());
try {
- log.debug(
- "page data size "
- + page.getData().size()
- + " bytes and "
- + pageValueCount
- + " records");
+ log.debug("page data size {} bytes and {} records",
page.getData().size(), pageValueCount);
initDataReader(
page.getDataEncoding(), page.getData().toInputStream(),
page.getValueCount());
} catch (IOException e) {
diff --git
a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
index 7c9fd994a0c2..700d7505fbc7 100644
---
a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
+++
b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
@@ -226,12 +226,7 @@ public abstract class BaseVectorizedColumnReader
implements ColumnReader<Writabl
this.definitionLevelColumn =
newRLEIterator(descriptor.getMaxDefinitionLevel(),
page.getDefinitionLevels());
try {
- log.debug(
- "page data size "
- + page.getData().size()
- + " bytes and "
- + pageValueCount
- + " records");
+ log.debug("page data size {} bytes and {} records",
page.getData().size(), pageValueCount);
initDataReader(
page.getDataEncoding(), page.getData().toInputStream(),
page.getValueCount());
} catch (IOException e) {
diff --git
a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
index 7c9fd994a0c2..700d7505fbc7 100644
---
a/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
+++
b/hudi-flink-datasource/hudi-flink1.20.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
@@ -226,12 +226,7 @@ public abstract class BaseVectorizedColumnReader
implements ColumnReader<Writabl
this.definitionLevelColumn =
newRLEIterator(descriptor.getMaxDefinitionLevel(),
page.getDefinitionLevels());
try {
- log.debug(
- "page data size "
- + page.getData().size()
- + " bytes and "
- + pageValueCount
- + " records");
+ log.debug("page data size {} bytes and {} records",
page.getData().size(), pageValueCount);
initDataReader(
page.getDataEncoding(), page.getData().toInputStream(),
page.getValueCount());
} catch (IOException e) {
diff --git
a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
index 7c9fd994a0c2..700d7505fbc7 100644
---
a/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
+++
b/hudi-flink-datasource/hudi-flink2.0.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
@@ -226,12 +226,7 @@ public abstract class BaseVectorizedColumnReader
implements ColumnReader<Writabl
this.definitionLevelColumn =
newRLEIterator(descriptor.getMaxDefinitionLevel(),
page.getDefinitionLevels());
try {
- log.debug(
- "page data size "
- + page.getData().size()
- + " bytes and "
- + pageValueCount
- + " records");
+ log.debug("page data size {} bytes and {} records",
page.getData().size(), pageValueCount);
initDataReader(
page.getDataEncoding(), page.getData().toInputStream(),
page.getValueCount());
} catch (IOException e) {
diff --git
a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
index fbb09823e9b9..c5a170c33e10 100644
---
a/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
+++
b/hudi-flink-datasource/hudi-flink2.1.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/BaseVectorizedColumnReader.java
@@ -228,12 +228,7 @@ public abstract class BaseVectorizedColumnReader
implements ColumnReader<Writabl
this.definitionLevelColumn =
newRLEIterator(descriptor.getMaxDefinitionLevel(),
page.getDefinitionLevels());
try {
- LOG.debug(
- "page data size "
- + page.getData().size()
- + " bytes and "
- + pageValueCount
- + " records");
+ LOG.debug("page data size {} bytes and {} records",
page.getData().size(), pageValueCount);
initDataReader(
page.getDataEncoding(), page.getData().toInputStream(),
page.getValueCount());
} catch (IOException e) {
