[hotfix][connector/kafka] Reduce the offset commit logging verbosity …

…from INFO to DEBUG.
deepfield · Mar 29, 2021 · b5b9682 · b5b9682
1 parent b0077c4
commit b5b9682
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/...kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java b/...kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
@@ -109,11 +109,13 @@ public List<KafkaPartitionSplit> snapshotState(long checkpointId) {
 
  @Override
  public void notifyCheckpointComplete(long checkpointId) throws Exception {
- LOG.info("Committing offsets for checkpoint {}", checkpointId);
+ LOG.debug("Committing offsets for checkpoint {}", checkpointId);
  ((KafkaSourceFetcherManager<T>) splitFetcherManager)
  .commitOffsets(
  offsetsToCommit.get(checkpointId),
  (ignored, e) -> {
+ // The offset commit here is needed by the external monitoring. It won't
+ // break Flink job's correctness if we fail to commit the offset here.
  if (e != null) {
  LOG.warn(
  "Failed to commit consumer offsets for checkpoint {}",
@@ -124,7 +126,7 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
  "Successfully committed offsets for checkpoint {}",
  checkpointId);
  // If the finished topic partition has been committed, we remove it
- // from the offsets of finsihed splits map.
+ // from the offsets of the finished splits map.
  Map<TopicPartition, OffsetAndMetadata> committedPartitions =
  offsetsToCommit.get(checkpointId);
  offsetsOfFinishedSplits