Repository: apache/kafka
File: core/src/main/scala/kafka/server/ReplicaManager.scala (Scala)

Code before:
        result.info.leaderHwChange match {
          case LeaderHwChange.INCREASED =>
            // some delayed operations may be unblocked after HW changed
            delayedProducePurgatory.checkAndComplete(requestKey)
            delayedFetchPurgatory.checkAndComplete(requestKey)
            delayedDeleteRecordsPurgatory.checkAndComplete(requestKey)
            if (topicIdPartition.topicId != Uuid.ZERO_UUID) delayedShareFetchPurgatory.checkAndComplete(new DelayedShareFetchPartitionKey(
              topicIdPartition.topicId, topicIdPartition.partition))
          case LeaderHwChange.SAME =>
            // probably unblock some follower fetch requests since log end offset has been updated
            delayedFetchPurgatory.checkAndComplete(requestKey)
          case LeaderHwChange.NONE =>
          // nothing
        }
      }
    }
  }

  private def maybeAddDelayedProduce(
    requiredAcks: Short,
    delayedProduceLock: Option[Lock],
    timeoutMs: Long,
    entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
    initialAppendResults: Map[TopicIdPartition, LogAppendResult],
    initialProduceStatus: Map[TopicIdPartition, ProducePartitionStatus],
    responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit,
  ): Unit = {
    if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, initialAppendResults)) {
      // create delayed produce operation
      val produceMetadata = ProduceMetadata(requiredAcks, initialProduceStatus)
      val delayedProduce = new DelayedProduce(timeoutMs, produceMetadata, this, responseCallback)

      // create a list of (topic, partition) pairs to use as keys for this delayed produce operation
      val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toList

      // try to complete the request immediately, otherwise put it into the purgatory
      // this is because while the delayed produce operation is being created, new
      // requests may arrive and hence make this operation completable.
      delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys.asJava)
    } else {
      // we can respond immediately
      val produceResponseStatus = initialProduceStatus.map { case (k, status) => k -> status.responseStatus }
      responseCallback(produceResponseStatus)
    }
  }

  private def sendInvalidRequiredAcksResponse(
    entries: Map[TopicIdPartition, MemoryRecords],
    responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit): Unit = {
    // If required.acks is outside accepted range, something is wrong with the client
    // Just return an error and don't handle the request at all

Review comment (suggestion):
Please remove `delayedProduceLock` from method signature too 

Code after:
            delayedFetchPurgatory.checkAndComplete(requestKey)
            delayedDeleteRecordsPurgatory.checkAndComplete(requestKey)
            if (topicIdPartition.topicId != Uuid.ZERO_UUID) delayedShareFetchPurgatory.checkAndComplete(new DelayedShareFetchPartitionKey(
              topicIdPartition.topicId, topicIdPartition.partition))
          case LeaderHwChange.SAME =>
            // probably unblock some follower fetch requests since log end offset has been updated
            delayedFetchPurgatory.checkAndComplete(requestKey)
          case LeaderHwChange.NONE =>
          // nothing
        }
      }
    }
  }

  private def maybeAddDelayedProduce(
    requiredAcks: Short,
    timeoutMs: Long,
    entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
    initialAppendResults: Map[TopicIdPartition, LogAppendResult],
    initialProduceStatus: Map[TopicIdPartition, ProducePartitionStatus],
    responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit,
  ): Unit = {
    if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, initialAppendResults)) {
      // create delayed produce operation
      val produceMetadata = ProduceMetadata(requiredAcks, initialProduceStatus)
      val delayedProduce = new DelayedProduce(timeoutMs, produceMetadata, this, responseCallback)

      // create a list of (topic, partition) pairs to use as keys for this delayed produce operation
      val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toList

      // try to complete the request immediately, otherwise put it into the purgatory
      // this is because while the delayed produce operation is being created, new
      // requests may arrive and hence make this operation completable.
      delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys.asJava)
    } else {
      // we can respond immediately
      val produceResponseStatus = initialProduceStatus.map { case (k, status) => k -> status.responseStatus }
      responseCallback(produceResponseStatus)
    }
  }

  private def sendInvalidRequiredAcksResponse(
    entries: Map[TopicIdPartition, MemoryRecords],
    responseCallback: Map[TopicIdPartition, PartitionResponse] => Unit): Unit = {
    // If required.acks is outside accepted range, something is wrong with the client
    // Just return an error and don't handle the request at all
    val responseStatus = entries.map { case (topicIdPartition, _) =>
      topicIdPartition -> new PartitionResponse(
        Errors.INVALID_REQUIRED_ACKS,
        LogAppendInfo.UNKNOWN_LOG_APPEND_INFO.firstOffset,
        RecordBatch.NO_TIMESTAMP,