[FLINK-36278] Decrease log levels for hotpath logs

We should never use INFO for tracking records on the hotpath. Ideally, we would use trace but for this commit I just decreased to DEBUG to minimize the impact on production settings (is it even possible to leave production on INFO currently?).

Author: AHeise

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java

@@ -138,7 +138,7 @@ public InputStatus pollNext(ReaderOutput<T> readerOutput) throws Exception {
         }
 
         if (restartingReaders.get()) {
-            logger.info("Poll next invoked while restarting readers");
+            logger.debug("Poll next invoked while restarting readers");
             return logAndReturnInputStatus(InputStatus.NOTHING_AVAILABLE);
         }
 

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/DynamicKafkaSourceExternalContext.java

@@ -240,7 +240,7 @@ public void writeRecords(List<String> records) {
                         }
                     }
 
-                    logger.info("Writing producer records: {}", producerRecords);
+                    logger.debug("Writing producer records: {}", producerRecords);
 
                     DynamicKafkaSourceTestHelper.produceToKafka(
                             clusterPropertiesMap.get(cluster),

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

@@ -2097,7 +2097,7 @@ public void flatMap(
 
                                 count++;
 
-                                LOG.info("Received message {}, total {} messages", value, count);
+                                LOG.debug("Received message {}, total {} messages", value, count);
 
                                 // verify if we've seen everything
                                 if (count == finalCount) {

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

@@ -167,7 +167,7 @@ public static void startClusters(KafkaTestEnvironment.Config environmentConfig)
             throws Exception {
         kafkaServer = constructKafkaTestEnvironment();
 
-        LOG.info("Starting KafkaTestBase.prepare() for Kafka " + kafkaServer.getVersion());
+        LOG.info("Starting KafkaTestBase.prepare() for Kafka {}", kafkaServer.getVersion());
 
         kafkaServer.prepare(environmentConfig);