This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 98598d6e885f54afa65e81285022ca0f49aba0a8 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. (cherry picked from commit fc91325b63812851d42b5fc66b3738564c0225f2) --- .../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 fad1f7e9272b..37d8b8294c09 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 8cb5ef9fdeb4..27f9b375de31 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 @@ -144,7 +144,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, rowType, preserveHoodieMetadata, isAppendMode && !populateMetaFields); @@ -154,7 +154,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); @@ -172,7 +172,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 34ac7e801570..2ef5d261cd1e 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 b5fe77b8e082..5167c1f62342 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 @@ -467,7 +467,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) {
