[FLINK-14654][logging] Fix various placeholder issues

flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -161,7 +161,7 @@ private void tryDelete(AdminClient adminClient, String topic)
  try {
  adminClient.deleteTopics(Collections.singleton(topic)).all().get(DELETE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
  } catch (TimeoutException e) {
- LOG.info("Did not receive delete topic response within %d seconds. Checking if it succeeded",
+ LOG.info("Did not receive delete topic response within {} seconds. Checking if it succeeded",
  DELETE_TIMEOUT_SECONDS);
  if (adminClient.listTopics().names().get(DELETE_TIMEOUT_SECONDS, TimeUnit.SECONDS).contains(topic)) {
  throw new Exception("Topic still exists after timeout");

flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisProducer.java

@@ -361,7 +361,7 @@ private void checkAndPropagateAsyncError() throws Exception {
  if (failOnError) {
  throw new RuntimeException("An exception was thrown while processing a record: " + errorMessages, thrownException);
  } else {
- LOG.warn("An exception was thrown while processing a record: {}", thrownException, errorMessages);
+ LOG.warn("An exception was thrown while processing a record: {}.", errorMessages, thrownException);
 
  // reset, prevent double throwing
  thrownException = null;

flink-contrib/flink-connector-wikiedits/src/main/java/org/apache/flink/streaming/connectors/wikiedits/WikipediaEditEventIrcStream.java

@@ -166,7 +166,7 @@ public void onNotice(String target, IRCUser user, String msg) {
 
  @Override
  public void onPart(String chan, IRCUser user, String msg) {
- LOG.debug("[{}] {} parts.", chan, user.getNick(), msg);
+ LOG.debug("[{}] {} parts {}.", chan, user.getNick(), msg);
  }
 
  @Override

flink-end-to-end-tests/flink-connector-gcp-pubsub-emulator-tests/src/test/java/org/apache/flink/streaming/connectors/gcp/pubsub/emulator/GCloudEmulatorManager.java

@@ -207,7 +207,7 @@ private static boolean waitForOkStatus(String label, String port) {
  private static String getPort(Map<String, List<PortBinding>> ports, String internalTCPPort, String label) {
  List<PortBinding> portMappings = ports.get(internalTCPPort + "/tcp");
  if (portMappings == null || portMappings.isEmpty()) {
- LOG.info("| {} Emulator {} --> NOTHING CONNECTED TO {}", label, internalTCPPort + "/tcp");
+ LOG.info("| {} Emulator --> NOTHING CONNECTED TO {}/tcp", label, internalTCPPort);
  return null;
  }
 

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetPojoInputFormat.java

@@ -61,7 +61,7 @@ public ParquetPojoInputFormat(Path filePath, MessageType messageType, PojoTypeIn
  public void open(FileInputSplit split) throws IOException {
  super.open(split);
  pojoFields = new Field[getFieldNames().length];
- LOG.error("Fields number is %d", getFieldNames().length);
+ LOG.error("Fields number is {}.", getFieldNames().length);
  final Map<String, Field> fieldMap = new HashMap<>();
  findAllFields(pojoTypeClass, fieldMap);
 

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java

@@ -321,7 +321,7 @@ private FilterPredicate toParquetPredicate(Expression exp) {
  }
  } else if (exp instanceof BinaryExpression) {
  if (exp instanceof And) {
- LOG.debug("All of the predicates should be in CNF. Found an AND expression.", exp);
+ LOG.debug("All of the predicates should be in CNF. Found an AND expression: {}.", exp);
  } else if (exp instanceof Or) {
  FilterPredicate c1 = toParquetPredicate(((Or) exp).left());
  FilterPredicate c2 = toParquetPredicate(((Or) exp).right());

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java

@@ -158,7 +158,7 @@ public void report() {
  client.send(request);
  LOGGER.debug("Reported series with size {}.", request.getSeries().getSeries().size());
  } catch (SocketTimeoutException e) {
- LOGGER.warn("Failed reporting metrics to Datadog because of socket timeout.", e.getMessage());
+ LOGGER.warn("Failed reporting metrics to Datadog because of socket timeout: {}.", e.getMessage());
  } catch (Exception e) {
  LOGGER.warn("Failed reporting metrics to Datadog.", e);
  }

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/PipelineErrorHandler.java

@@ -55,7 +55,7 @@ protected void channelRead0(ChannelHandlerContext ctx, Object message) {
 
  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- logger.debug("Unhandled exception: {}", cause);
+ logger.debug("Unhandled exception.", cause);
  sendError(ctx, ExceptionUtils.stringifyException(cause));
  }
 

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/ReporterSetup.java

@@ -223,7 +223,7 @@ private static Optional<MetricReporter> loadViaFactory(
  MetricReporterFactory factory = reporterFactories.get(factoryClassName);
 
  if (factory == null) {
- LOG.warn("The reporter factory ({}) could not be found for reporter {}. Available factories: ", factoryClassName, reporterName, reporterFactories.keySet());
+ LOG.warn("The reporter factory ({}) could not be found for reporter {}. Available factories: {}.", factoryClassName, reporterName, reporterFactories.keySet());
  return Optional.empty();
  } else {
  final MetricConfig metricConfig = new MetricConfig();

flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/HadoopModule.java

@@ -115,7 +115,7 @@ public void install() throws SecurityInstallException {
  Credentials.class);
  addCredentialsMethod.invoke(loginUser, credentials);
  } catch (NoSuchMethodException e) {
- LOG.warn("Could not find method implementations in the shaded jar. Exception: {}", e);
+ LOG.warn("Could not find method implementations in the shaded jar.", e);
  } catch (InvocationTargetException e) {
  throw e.getTargetException();
  }
@@ -129,7 +129,7 @@ public void install() throws SecurityInstallException {
  Method loginUserFromSubjectMethod = UserGroupInformation.class.getMethod("loginUserFromSubject", Subject.class);
  loginUserFromSubjectMethod.invoke(null, (Subject) null);
  } catch (NoSuchMethodException e) {
- LOG.warn("Could not find method implementations in the shaded jar. Exception: {}", e);
+ LOG.warn("Could not find method implementations in the shaded jar.", e);
  } catch (InvocationTargetException e) {
  throw e.getTargetException();
  }

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitor.java

@@ -101,7 +101,7 @@ private void setProperty(ColumnFamilyHandle handle, String property, RocksDBNati
  }
  } catch (RocksDBException e) {
  metricView.close();
- LOG.warn("Failed to read native metric %s from RocksDB", property, e);
+ LOG.warn("Failed to read native metric {} from RocksDB.", property, e);
  }
  }
 

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java

@@ -571,11 +571,11 @@ public static int divideInt(int a, int b) {
 
  public static String subString(String str, long start, long len) {
  if (len < 0) {
- LOG.error("len of 'substring(str, start, len)' must be >= 0 and Int type, but len = {0}", len);
+ LOG.error("len of 'substring(str, start, len)' must be >= 0 and Int type, but len = {}", len);
  return null;
  }
  if (len > Integer.MAX_VALUE || start > Integer.MAX_VALUE) {
- LOG.error("len or start of 'substring(str, start, len)' must be Int type, but len = {0}, start = {0}", len, start);
+ LOG.error("len or start of 'substring(str, start, len)' must be Int type, but len = {}, start = {}", len, start);
  return null;
  }
  int length = (int) len;

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesInputFormat.java

@@ -50,7 +50,7 @@ public ValuesInputFormat(GeneratedInput<GenericInputFormat<BaseRow>> generatedIn
 
  @Override
  public void open(GenericInputSplit split) {
- LOG.debug("Compiling GenericInputFormat: $name \n\n Code:\n$code",
+ LOG.debug("Compiling GenericInputFormat: {} \n\n Code:\n{}",
  generatedInput.getClassName(), generatedInput.getCode());
  LOG.debug("Instantiating GenericInputFormat.");