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

Code before:
          markerResults.put(partition, error)
          // We should only call maybeSendResponseCallback once per marker. Otherwise, it causes sending the response
          // prematurely.
          if (numPartitions.decrementAndGet() == 0) {
            maybeSendResponseCallback(producerId, marker.transactionResult, markerResults)
          }
        }

        val controlRecords = mutable.Map.empty[TopicIdPartition, MemoryRecords]
        val markerTransactionVersion = marker.transactionVersion

        partitionsWithCompatibleMessageFormat.foreach { partition =>
          if (partition.topic == GROUP_METADATA_TOPIC_NAME) {
            groupCoordinator.completeTransaction(
              partition,
              marker.producerId,
              marker.producerEpoch,
              marker.coordinatorEpoch,
              marker.transactionResult,
              markerTransactionVersion,
              Duration.ofMillis(config.requestTimeoutMs.toLong)
            ).whenComplete { (_, exception) =>
              val error = if (exception == null) {
                Errors.NONE
              } else {
                if (Errors.maybeUnwrapException(exception).isInstanceOf[IdempotentTransactionMarkerException])
                  Errors.NONE
                else {
                  Errors.forException(exception) match {
                    case Errors.COORDINATOR_NOT_AVAILABLE | Errors.COORDINATOR_LOAD_IN_PROGRESS | Errors.NOT_COORDINATOR =>
                      // The transaction coordinator does not expect those errors so we translate them
                      // to NOT_LEADER_OR_FOLLOWER to signal to it that the coordinator is not ready yet.
                      Errors.NOT_LEADER_OR_FOLLOWER
                    case error =>
                      error
                  }
                }
              }
              addResultAndMaybeComplete(partition, error)
            }
          } else {
            // Otherwise, the regular appendRecords path is used for all the non __consumer_offsets
            // partitions or for all partitions when the new group coordinator is disabled.
            // If topicIdPartition contains Uuid.ZERO_UUid all functionality will fall back on topic name.
            controlRecords += replicaManager.topicIdPartition(partition) -> MemoryRecords.withEndTransactionMarker(
              producerId,
              marker.producerEpoch,
              new EndTransactionMarker(controlRecordType, marker.coordinatorEpoch)
            )
          }
        }

Review comment (refactor):
Could you add a helper method to `IdempotentTransactionMarkerException` to handle this condition?

Code after:
          markerResults.put(partition, error)
          // We should only call maybeSendResponseCallback once per marker. Otherwise, it causes sending the response
          // prematurely.
          if (numPartitions.decrementAndGet() == 0) {
            maybeSendResponseCallback(producerId, marker.transactionResult, markerResults)
          }
        }

        val controlRecords = mutable.Map.empty[TopicIdPartition, MemoryRecords]
        val markerTransactionVersion = marker.transactionVersion

        partitionsWithCompatibleMessageFormat.foreach { partition =>
          if (partition.topic == GROUP_METADATA_TOPIC_NAME) {
            groupCoordinator.completeTransaction(
              partition,
              marker.producerId,
              marker.producerEpoch,
              marker.coordinatorEpoch,
              marker.transactionResult,
              markerTransactionVersion,
              Duration.ofMillis(config.requestTimeoutMs.toLong)
            ).whenComplete { (_, exception) =>
              val error = if (exception == null) {
                Errors.NONE
              } else {
                Errors.forException(exception) match {
                  case Errors.COORDINATOR_NOT_AVAILABLE | Errors.COORDINATOR_LOAD_IN_PROGRESS | Errors.NOT_COORDINATOR =>
                    // The transaction coordinator does not expect those errors so we translate them
                    // to NOT_LEADER_OR_FOLLOWER to signal to it that the coordinator is not ready yet.
                    Errors.NOT_LEADER_OR_FOLLOWER
                  case error =>
                    error
                }
              }
              addResultAndMaybeComplete(partition, error)
            }
          } else {
            // Otherwise, the regular appendRecords path is used for all the non __consumer_offsets
            // partitions or for all partitions when the new group coordinator is disabled.
            // If topicIdPartition contains Uuid.ZERO_UUid all functionality will fall back on topic name.
            controlRecords += replicaManager.topicIdPartition(partition) -> MemoryRecords.withEndTransactionMarker(
              producerId,
              marker.producerEpoch,
              new EndTransactionMarker(controlRecordType, marker.coordinatorEpoch)
            )
          }
        }

        if (controlRecords.nonEmpty) {
          replicaManager.appendRecords(
            timeout = config.requestTimeoutMs.toLong,