Fix Slf4j formatting errors in sdks/java/io/google-cloud-platform

Author: stankiewicz

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java

@@ -296,7 +296,7 @@ void logBigQueryError(@Nullable Job job) {
               ? String.format(" due to: %s", jobErrors.get(jobErrors.size() - 1).getMessage())
               : "";
 
-      LOG.error(String.format("BigQuery Error : %s %s", finalError, causativeError));
+      LOG.error("BigQuery Error : {} {}", finalError, causativeError);
     }
   }
 

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

@@ -781,8 +781,7 @@ public static <T> TypedRead<T> read(SerializableFunction<SchemaAndRecord, T> par
                         (writer, reader) ->
                             new GenericDatumTransformer<>(parseFn, jsonTableSchema, writer);
                   } catch (IOException e) {
-                    LOG.warn(
-                        String.format("Error while converting table schema %s to JSON!", input), e);
+                    LOG.warn("Error while converting table schema {} to JSON!", input, e);
                     return null;
                   }
                 })

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java

@@ -383,7 +383,7 @@ static void startJob(
           return; // SUCCEEDED
         } catch (IOException e) {
           if (errorExtractor.itemAlreadyExists(e)) {
-            LOG.info("BigQuery job " + jobRef + " already exists, will not retry inserting it:", e);
+            LOG.info("BigQuery job {} already exists, will not retry inserting it:", jobRef, e);
             return; // SUCCEEDED
           }
 
@@ -395,7 +395,7 @@ static void startJob(
                       .create()
                   : null) {
             // ignore and retry
-            LOG.info("Failed to insert job " + jobRef + ", will retry:", e);
+            LOG.info("Failed to insert job {}, will retry:", jobRef, e);
           }
           lastException = e;
         }
@@ -431,12 +431,11 @@ static void startJobStream(
         } catch (IOException e) {
           if (errorExtractor.itemAlreadyExists(e)) {
             LOG.info(
-                "BigQuery job " + jobReference + " already exists, will not retry inserting it:",
-                e);
+                "BigQuery job {} already exists, will not retry inserting it:", jobReference, e);
             return; // SUCCEEDED
           }
           // ignore and retry
-          LOG.info("Failed to insert job " + jobReference + ", will retry:", e);
+          LOG.info("Failed to insert job {}, will retry:", jobReference, e);
           exception = e;
         }
       } while (nextBackOff(sleeper, backOff));
@@ -1065,9 +1064,8 @@ public List<TableDataInsertAllResponse.InsertErrors> call() throws Exception {
                     .withFullResourceName(BigQueryHelpers.toTableFullResourceName(ref))
                     .create()) {
               LOG.info(
-                  String.format(
-                      "BigQuery insertAll error, retrying: %s",
-                      ApiErrorExtractor.INSTANCE.getErrorMessage(e)));
+                  "BigQuery insertAll error, retrying: {}",
+                  ApiErrorExtractor.INSTANCE.getErrorMessage(e));
             }
             try {
               long nextBackOffMillis = backoff1.nextBackOffMillis();

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/DynamicDestinationsHelpers.java

@@ -451,7 +451,7 @@ private MatchTableDynamicDestinations(
             try (DatasetService datasetService = bqServices.getDatasetService(bqOptions)) {
               return datasetService.getTable(tableReference);
             } catch (InterruptedException | IOException e) {
-              LOG.info("Failed to get BigQuery table " + tableReference);
+              LOG.info("Failed to get BigQuery table {}", tableReference);
             }
           } catch (Exception e) {
             throw new RuntimeException(e);

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFinalizeWritesDoFn.java

@@ -120,7 +120,7 @@ public void process(PipelineOptions pipelineOptions, @Element KV<String, String>
         contexts -> {
           RetryManager.Operation.Context<FinalizeWriteStreamResponse> firstContext =
               Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));
-          LOG.error("Finalize of stream " + streamId + " failed with " + firstContext.getError());
+          LOG.error("Finalize of stream {} failed", streamId, firstContext.getError());
           finalizeOperationsFailed.inc();
           BigQuerySinkMetrics.reportFailedRPCMetrics(
               firstContext, BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM);
@@ -132,7 +132,7 @@ public void process(PipelineOptions pipelineOptions, @Element KV<String, String>
               Preconditions.checkArgumentNotNull(
                   c.getResult(),
                   "Finalize of write stream " + streamId + " finished, but with null result");
-          LOG.debug("Finalize of stream " + streamId + " finished with " + response);
+          LOG.debug("Finalize of stream {} finished with {}", streamId, response);
           rowsFinalized.inc(response.getRowCount());
 
           finalizeOperationsSucceeded.inc();
@@ -169,17 +169,15 @@ public void finishBundle(PipelineOptions pipelineOptions) throws Exception {
             RetryManager.Operation.Context<BatchCommitWriteStreamsResponse> firstContext =
                 Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));
             LOG.error(
-                "BatchCommit failed. tableId "
-                    + tableId
-                    + " streamNames "
-                    + streamNames
-                    + " error: "
-                    + firstContext.getError());
+                "BatchCommit failed. tableId {} streamNames {}",
+                tableId,
+                streamNames,
+                firstContext.getError());
             batchCommitOperationsFailed.inc();
             return RetryType.RETRY_ALL_OPERATIONS;
           },
           c -> {
-            LOG.info("BatchCommit succeeded for tableId " + tableId + " response " + c.getResult());
+            LOG.info("BatchCommit succeeded for tableId {} response {}", tableId, c.getResult());
             batchCommitOperationsSucceeded.inc();
           },
           response -> {

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java

@@ -212,8 +212,7 @@ public void process(PipelineOptions pipelineOptions, @Element KV<String, Operati
             Context<FlushRowsResponse> failedContext =
                 Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));
             Throwable error = failedContext.getError();
-            LOG.warn(
-                "Flush of stream " + streamId + " to offset " + offset + " failed with " + error);
+            LOG.warn("Flush of stream {} to offset {} failed", streamId, offset, error);
             flushOperationsFailed.inc();
             BigQuerySinkMetrics.reportFailedRPCMetrics(
                 failedContext, BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS);
@@ -286,11 +285,9 @@ public void process(PipelineOptions pipelineOptions, @Element KV<String, Operati
           },
           contexts -> {
             LOG.warn(
-                "Finalize of stream "
-                    + streamId
-                    + " failed with "
-                    + Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null))
-                        .getError());
+                "Finalize of stream {} failed",
+                streamId,
+                Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null)).getError());
             finalizeOperationsFailed.inc();
             @Nullable
             Context<FinalizeWriteStreamResponse> firstContext = Iterables.getFirst(contexts, null);

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java

@@ -135,7 +135,7 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
           .expireAfterAccess(15, TimeUnit.MINUTES)
           .removalListener(
               (RemovalNotification<String, AppendClientInfo> removal) -> {
-                LOG.info("Expiring append client for " + removal.getKey());
+                LOG.info("Expiring append client for {}", removal.getKey());
                 final @Nullable AppendClientInfo appendClientInfo = removal.getValue();
                 if (appendClientInfo != null) {
                   appendClientInfo.close();

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java

@@ -799,7 +799,7 @@ public void process(
               // Don't log errors for expected offset mismatch. These will be logged as warnings
               // below.
               LOG.error(
-                  "Got error " + failedContext.getError() + " closing " + failedContext.streamName);
+                  "Got error {} closing {}", failedContext.getError(), failedContext.streamName);
             }
 
             try {
@@ -833,11 +833,9 @@ public void process(
             if (offsetMismatch || streamDoesNotExist) {
               appendOffsetFailures.inc();
               LOG.warn(
-                  "Append to "
-                      + failedContext
-                      + " failed with "
-                      + failedContext.getError()
-                      + " Will retry with a new stream");
+                  "Append to {} failed. Will retry with a new stream",
+                  failedContext,
+                  failedContext.getError());
               // Finalize the stream and clear streamName so a new stream will be created.
               o.get(flushTag)
                   .output(
@@ -901,9 +899,8 @@ public void process(
             // the ProtoRows iterable at 2MB and the max request size is 10MB, this scenario seems
             // nearly impossible.
             LOG.error(
-                "A request containing more than one row is over the request size limit of "
-                    + maxRequestSize
-                    + ". This is unexpected. All rows in the request will be sent to the failed-rows PCollection.");
+                "A request containing more than one row is over the request size limit of {}. This is unexpected. All rows in the request will be sent to the failed-rows PCollection.",
+                maxRequestSize);
           }
           for (int i = 0; i < splitValue.getProtoRows().getSerializedRowsCount(); ++i) {
             org.joda.time.Instant timestamp = splitValue.getTimestamps().get(i);

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableSchemaCache.java

@@ -294,7 +294,7 @@ public void refreshThread() {
       // Since this is a daemon thread, don't exit until it is explicitly shut down. Exiting early
       // can cause the
       // pipeline to stall.
-      LOG.error("Caught exception in BigQuery's table schema cache refresh thread: " + e);
+      LOG.error("Caught exception in BigQuery's table schema cache refresh thread", e);
     }
     this.refreshExecutor.submit(this::refreshThread);
   }
@@ -313,7 +313,7 @@ private Map<String, TableSchema> refreshAll(Map<String, Refresh> tables)
       if (table == null) {
         throw new RuntimeException("Did not get value for table " + tableReference);
       }
-      LOG.info("Refreshed BigQuery schema for " + entry.getKey());
+      LOG.info("Refreshed BigQuery schema for {}", entry.getKey());
       schemas.put(entry.getKey(), table.getSchema());
     }
     return schemas;

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java

@@ -1545,7 +1545,7 @@ private void checkForFailures() throws IOException {
           String.format(
               "At least %d errors occurred writing to Bigtable. First %d errors: %s",
               i + failures.size(), i, logEntry.toString());
-      LOG.error(message);
+      LOG.error("{}", message);
       IOException exception = new IOException(message);
       for (BigtableWriteException e : suppressed) {
         exception.addSuppressed(e);
@@ -2462,7 +2462,7 @@ private void createOrUpdateMetadataTable(
       // Only try to create or update metadata table if option is set to true. Otherwise, just
       // check if the table exists.
       if (shouldCreateOrUpdateMetadataTable && metadataTableAdminDao.createMetadataTable()) {
-        LOG.info("Created metadata table: " + metadataTableId);
+        LOG.info("Created metadata table: {}", metadataTableId);
       }
     }
 
@@ -2645,7 +2645,7 @@ public static boolean createOrUpdateReadChangeStreamMetadataTable(
       // Only try to create or update metadata table if option is set to true. Otherwise, just
       // check if the table exists.
       if (metadataTableAdminDao.createMetadataTable()) {
-        LOG.info("Created metadata table: " + metadataTableAdminDao.getTableId());
+        LOG.info("Created metadata table: {}", metadataTableAdminDao.getTableId());
       }
       return metadataTableAdminDao.doesMetadataTableExist();
     } finally {