<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.samza.checkpoint.kafka

import java.nio.ByteBuffer
import java.util
import java.util.{Collections, Properties}

import kafka.utils.ZkUtils
import org.apache.kafka.common.utils.Utils
import org.apache.samza.checkpoint.{Checkpoint, CheckpointManager}
import org.apache.samza.container.TaskName
import org.apache.samza.serializers.CheckpointSerde
import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadata
import org.apache.samza.system.kafka.{KafkaSystemAdmin, KafkaStreamSpec}
import org.apache.samza.system.{StreamSpec, SystemAdmin, _}
import org.apache.samza.util._
import org.apache.samza.{Partition, SamzaException}

import scala.collection.JavaConversions._
import scala.collection.mutable

/**
 * Kafka checkpoint manager is used to store checkpoints in a Kafka topic.
 * To read a checkpoint for a specific taskName, we find the newest message
 * keyed to that taskName. If there is no such message, no checkpoint data
 * exists.  The underlying log has a single partition into which all
 * checkpoints and TaskName to changelog partition mappings are written.
 */
class KafkaCheckpointManager(
                              clientId: String,
                              checkpointTopic: String,
                              val systemName: String,
                              replicationFactor: Int,
                              socketTimeout: Int,
                              bufferSize: Int,
                              fetchSize: Int,
                              getSystemConsumer: () => SystemConsumer,
                              getSystemAdmin: () => SystemAdmin,
                              val metadataStore: TopicMetadataStore,
                              getSystemProducer: () => SystemProducer,
                              val connectZk: () => ZkUtils,
                              systemStreamPartitionGrouperFactoryString: String,
                              failOnCheckpointValidation: Boolean,
                              val retryBackoff: ExponentialSleepStrategy = new ExponentialSleepStrategy,
                              serde: CheckpointSerde = new CheckpointSerde,
                              checkpointTopicProperties: Properties = new Properties) extends CheckpointManager with Logging {

  var taskNames = Set[TaskName]()
  @volatile var  systemProducer: SystemProducer = null
  var taskNamesToOffsets: Map[TaskName, Checkpoint] = null
  val systemAdmin = getSystemAdmin()

  val kafkaUtil: KafkaUtil = new KafkaUtil(retryBackoff, connectZk)



  KafkaCheckpointLogKey.setSystemStreamPartitionGrouperFactoryString(systemStreamPartitionGrouperFactoryString)

  info("Creating KafkaCheckpointManager with: clientId=%s, checkpointTopic=%s, systemName=%s" format(clientId, checkpointTopic, systemName))

  /**
   * Write Checkpoint for specified taskName to log
   *
   * @param taskName Specific Samza taskName of which to write a checkpoint of.
   * @param checkpoint Reference to a Checkpoint object to store offset data in.
   **/
  override def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
    val key = KafkaCheckpointLogKey.getCheckpointKey(taskName)
    val keyBytes = key.toBytes()
    val msgBytes = serde.toBytes(checkpoint)
    val systemStream = new SystemStream(systemName, checkpointTopic)
    val envelope = new OutgoingMessageEnvelope(systemStream, keyBytes, msgBytes)

    retryBackoff.run(
      loop => {
        if (systemProducer == null) {
          synchronized {
            if (systemProducer == null) {
              systemProducer = getSystemProducer()
              systemProducer.register(taskName.getTaskName)
              systemProducer.start
            }
          }
        }

        systemProducer.send(taskName.getTaskName, envelope)
        systemProducer.flush(taskName.getTaskName) // make sure it is written
        info("Completed writing checkpoint=%s into %s topic for system %s." format(checkpoint, checkpointTopic, systemName) )
        loop.done
      },

      (exception, loop) => {
        warn("Failed to write checkpoint log partition entry %s: %s. Retrying." format(key, exception))
        debug("Exception detail:", exception)
      }
    )
  }<commit_msg>reduce debugging from info to debug in KafkaCheckpointManager.java<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.samza.checkpoint.kafka

import java.nio.ByteBuffer
import java.util
import java.util.{Collections, Properties}

import kafka.utils.ZkUtils
import org.apache.kafka.common.utils.Utils
import org.apache.samza.checkpoint.{Checkpoint, CheckpointManager}
import org.apache.samza.container.TaskName
import org.apache.samza.serializers.CheckpointSerde
import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadata
import org.apache.samza.system.kafka.{KafkaSystemAdmin, KafkaStreamSpec}
import org.apache.samza.system.{StreamSpec, SystemAdmin, _}
import org.apache.samza.util._
import org.apache.samza.{Partition, SamzaException}

import scala.collection.JavaConversions._
import scala.collection.mutable

/**
 * Kafka checkpoint manager is used to store checkpoints in a Kafka topic.
 * To read a checkpoint for a specific taskName, we find the newest message
 * keyed to that taskName. If there is no such message, no checkpoint data
 * exists.  The underlying log has a single partition into which all
 * checkpoints and TaskName to changelog partition mappings are written.
 */
class KafkaCheckpointManager(
                              clientId: String,
                              checkpointTopic: String,
                              val systemName: String,
                              replicationFactor: Int,
                              socketTimeout: Int,
                              bufferSize: Int,
                              fetchSize: Int,
                              getSystemConsumer: () => SystemConsumer,
                              getSystemAdmin: () => SystemAdmin,
                              val metadataStore: TopicMetadataStore,
                              getSystemProducer: () => SystemProducer,
                              val connectZk: () => ZkUtils,
                              systemStreamPartitionGrouperFactoryString: String,
                              failOnCheckpointValidation: Boolean,
                              val retryBackoff: ExponentialSleepStrategy = new ExponentialSleepStrategy,
                              serde: CheckpointSerde = new CheckpointSerde,
                              checkpointTopicProperties: Properties = new Properties) extends CheckpointManager with Logging {

  var taskNames = Set[TaskName]()
  @volatile var  systemProducer: SystemProducer = null
  var taskNamesToOffsets: Map[TaskName, Checkpoint] = null
  val systemAdmin = getSystemAdmin()

  val kafkaUtil: KafkaUtil = new KafkaUtil(retryBackoff, connectZk)



  KafkaCheckpointLogKey.setSystemStreamPartitionGrouperFactoryString(systemStreamPartitionGrouperFactoryString)

  info("Creating KafkaCheckpointManager with: clientId=%s, checkpointTopic=%s, systemName=%s" format(clientId, checkpointTopic, systemName))

  /**
   * Write Checkpoint for specified taskName to log
   *
   * @param taskName Specific Samza taskName of which to write a checkpoint of.
   * @param checkpoint Reference to a Checkpoint object to store offset data in.
   **/
  override def writeCheckpoint(taskName: TaskName, checkpoint: Checkpoint) {
    val key = KafkaCheckpointLogKey.getCheckpointKey(taskName)
    val keyBytes = key.toBytes()
    val msgBytes = serde.toBytes(checkpoint)
    val systemStream = new SystemStream(systemName, checkpointTopic)
    val envelope = new OutgoingMessageEnvelope(systemStream, keyBytes, msgBytes)

    retryBackoff.run(
      loop => {
        if (systemProducer == null) {
          synchronized {
            if (systemProducer == null) {
              systemProducer = getSystemProducer()
              systemProducer.register(taskName.getTaskName)
              systemProducer.start
            }
          }
        }

        systemProducer.send(taskName.getTaskName, envelope)
        systemProducer.flush(taskName.getTaskName) // make sure it is written
        debug("Completed writing checkpoint=%s into %s topic for system %s." format(checkpoint, checkpointTopic, systemName) )
        loop.done
      },

      (exception, loop) => {
        warn("Failed to write checkpoint log partition entry %s: %s. Retrying." format(key, exception))
        debug("Exception detail:", exception)
      }
    )
  }
{'fraction_non_alphanumeric': 0.06999409332545777, 'fraction_numerical': 0.0007875566056310297, 'mean_word_length': 2.7198828268033686, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4944824', 'n_tokens_mistral': 2553, 'n_tokens_neox': 2506, 'n_words': 803}