Repository: apache/kafka
File: core/src/main/scala/kafka/cluster/Partition.scala (Scala)

Code before:
  def markIsrShrink(): Unit
  def markFailed(): Unit
}

class DelayedOperations(topicId: Option[Uuid],
                        topicPartition: TopicPartition,
                        produce: DelayedOperationPurgatory[DelayedProduce],
                        fetch: DelayedOperationPurgatory[DelayedFetch],
                        deleteRecords: DelayedOperationPurgatory[DelayedDeleteRecords],
                        shareFetch: DelayedOperationPurgatory[DelayedShareFetch]) extends Logging {

  def checkAndCompleteAll(): Unit = {
    val requestKey = new TopicPartitionOperationKey(topicPartition)
    CoreUtils.swallow(() -> fetch.checkAndComplete(requestKey), this, Level.ERROR)
    CoreUtils.swallow(() -> produce.checkAndComplete(requestKey), this, Level.ERROR)
    CoreUtils.swallow(() -> deleteRecords.checkAndComplete(requestKey), this, Level.ERROR)
    if (topicId.isDefined) CoreUtils.swallow(() -> shareFetch.checkAndComplete(new DelayedShareFetchPartitionKey(
      topicId.get, topicPartition.partition())), this, Level.ERROR)
  }

  def numDelayedDelete: Int = deleteRecords.numDelayed()
}

object Partition {
  private val metricsPackage = "kafka.cluster"
  private val metricsClassName = "Partition"
  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)

  def apply(topicIdPartition: TopicIdPartition,
            time: Time,
            replicaManager: ReplicaManager): Partition = {
    Partition(
      topicPartition = topicIdPartition.topicPartition,
      topicId = Some(topicIdPartition.topicId),
      time = time,
      replicaManager = replicaManager)
  }
  def apply(topicPartition: TopicPartition,
            time: Time,
            replicaManager: ReplicaManager,
            topicId: Option[Uuid] = None): Partition = {

    val isrChangeListener = new AlterPartitionListener {
      override def markIsrExpand(): Unit = {
        replicaManager.isrExpandRate.mark()
      }

      override def markIsrShrink(): Unit = {
        replicaManager.isrShrinkRate.mark()
      }


Review comment (suggestion):
Could you please add specific comments, for example: “Changing the package or class name will break compatibility.”

Code after:
  def markIsrShrink(): Unit
  def markFailed(): Unit
}

class DelayedOperations(topicId: Option[Uuid],
                        topicPartition: TopicPartition,
                        produce: DelayedOperationPurgatory[DelayedProduce],
                        fetch: DelayedOperationPurgatory[DelayedFetch],
                        deleteRecords: DelayedOperationPurgatory[DelayedDeleteRecords],
                        shareFetch: DelayedOperationPurgatory[DelayedShareFetch]) extends Logging {

  def checkAndCompleteAll(): Unit = {
    val requestKey = new TopicPartitionOperationKey(topicPartition)
    CoreUtils.swallow(() -> fetch.checkAndComplete(requestKey), this, Level.ERROR)
    CoreUtils.swallow(() -> produce.checkAndComplete(requestKey), this, Level.ERROR)
    CoreUtils.swallow(() -> deleteRecords.checkAndComplete(requestKey), this, Level.ERROR)
    if (topicId.isDefined) CoreUtils.swallow(() -> shareFetch.checkAndComplete(new DelayedShareFetchPartitionKey(
      topicId.get, topicPartition.partition())), this, Level.ERROR)
  }

  def numDelayedDelete: Int = deleteRecords.numDelayed()
}

object Partition {
  // Changing the package or class name may cause incompatibility with existing code and metrics configuration
  private val metricsPackage = "kafka.cluster"
  private val metricsClassName = "Partition"
  private val metricsGroup = new KafkaMetricsGroup(metricsPackage, metricsClassName)

  def apply(topicIdPartition: TopicIdPartition,
            time: Time,
            replicaManager: ReplicaManager): Partition = {
    Partition(
      topicPartition = topicIdPartition.topicPartition,
      topicId = Some(topicIdPartition.topicId),
      time = time,
      replicaManager = replicaManager)
  }
  def apply(topicPartition: TopicPartition,
            time: Time,
            replicaManager: ReplicaManager,
            topicId: Option[Uuid] = None): Partition = {

    val isrChangeListener = new AlterPartitionListener {
      override def markIsrExpand(): Unit = {
        replicaManager.isrExpandRate.mark()
      }

      override def markIsrShrink(): Unit = {
        replicaManager.isrShrinkRate.mark()
      }