Repository: apache/kafka
File: metadata/src/main/java/org/apache/kafka/controller/QuorumController.java (Java)

Code before:
                        offsetControl.handleScheduleAppend(lastOffset);
                        return lastOffset;
                    }
                );
                op.processBatchEndOffset(offset);
                resultAndOffset = ControllerResultAndOffset.of(offset, result);

                log.debug("Read-write operation {} will be completed when the log " +
                    "reaches offset {}.", this, resultAndOffset.offset());
            }

            // Remember the latest offset and future if it is not already completed
            if (!future.isDone()) {
                controllerMetrics.updateIdleStartTime();
                deferredEventQueue.add(resultAndOffset.offset(), this);
            }
        }

        @Override
        public void handleException(Throwable exception) {
            complete(exception);
        }

        @Override
        public void complete(Throwable exception) {
            controllerMetrics.updateIdleEndTime();
            if (exception == null) {
                handleEventEnd(this.toString(), startProcessingTimeNs.getAsLong());
                future.complete(resultAndOffset.response());
            } else {
                future.completeExceptionally(
                    handleEventException(name, startProcessingTimeNs, exception));
            }
        }

        @Override
        public String toString() {
            return name + "(" + System.identityHashCode(this) + ")";
        }
    }

    /**
     * Append records to the Raft log. They will be written out asynchronously.
     *
     * @param log                   The log4j logger.
     * @param result                The controller result we are writing out.
     * @param maxRecordsPerBatch    The maximum number of records to allow in a batch.
     * @param appender              The callback to invoke for each batch. The arguments are last
     *                              write offset, record list, and the return result is the new
     *                              last write offset.
     * @return                      The final offset that was returned from the Raft layer.

Review comment (refactor):
We can remove this call since it will be called in either `handleEventEnd` or `handleEventException`.

Code after:
                            }
                            recordIndex++;
                        }
                        raftClient.schedulePreparedAppend();
                        offsetControl.handleScheduleAppend(lastOffset);
                        return lastOffset;
                    }
                );
                op.processBatchEndOffset(offset);
                resultAndOffset = ControllerResultAndOffset.of(offset, result);

                log.debug("Read-write operation {} will be completed when the log " +
                    "reaches offset {}.", this, resultAndOffset.offset());
            }

            // Remember the latest offset and future if it is not already completed
            if (!future.isDone()) {
                deferredEventQueue.add(resultAndOffset.offset(), this);
            }
        }

        @Override
        public void handleException(Throwable exception) {
            complete(exception);
        }

        @Override
        public void complete(Throwable exception) {
            if (exception == null) {
                handleEventEnd(this.toString(), startProcessingTimeNs.getAsLong());
                future.complete(resultAndOffset.response());
            } else {
                future.completeExceptionally(
                    handleEventException(name, startProcessingTimeNs, exception));
            }
        }

        @Override
        public String toString() {
            return name + "(" + System.identityHashCode(this) + ")";
        }
    }

    /**
     * Append records to the Raft log. They will be written out asynchronously.
     *
     * @param log                   The log4j logger.
     * @param result                The controller result we are writing out.
     * @param maxRecordsPerBatch    The maximum number of records to allow in a batch.
     * @param appender              The callback to invoke for each batch. The arguments are last
     *                              write offset, record list, and the return result is the new