private void append(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            List<U> records,
            boolean replay,
            boolean isAtomic,
            DeferredEvent event
        ) {
            if (state != CoordinatorState.ACTIVE) {
                throw new IllegalStateException("Coordinator must be active to append records");
            }

            if (records.isEmpty()) {
                // If the records are empty, it was a read operation after all. In this case,
                // the response can be returned directly iff there are no pending write operations;
                // otherwise, the read needs to wait on the last write operation to be completed.
                if (currentBatch != null) {
                    currentBatch.deferredEvents.add(event);
                } else {
                    if (coordinator.lastCommittedOffset() < coordinator.lastWrittenOffset()) {
                        deferredEventQueue.add(coordinator.lastWrittenOffset(), event);
                    } else {
                        event.complete(null);
                    }
                }
            } else {
                // If the records are not empty, first, they are applied to the state machine,
                // second, they are appended to the opened batch.
                long currentTimeMs = time.milliseconds();

                // If the current write operation is transactional, the current batch
                // is written before proceeding with it.
                if (producerId != RecordBatch.NO_PRODUCER_ID) {
                    isAtomic = true;
                    // If flushing fails, we don't catch the exception in order to let
                    // the caller fail the current operation.
                    flushCurrentBatch();
                }

                // Allocate a new batch if none exists.
                maybeAllocateNewBatch(
                    producerId,
                    producerEpoch,
                    verificationGuard,
                    currentTimeMs
                );

                // Prepare the records.
                List<SimpleRecord> recordsToAppend = new ArrayList<>(records.size());
                for (U record : records) {
                    recordsToAppend.add(new SimpleRecord(
                        currentTimeMs,
                        serializer.serializeKey(record),
                        serializer.serializeValue(record)
                    ));
                }

                if (isAtomic) {
                    // Compute the estimated size of the records.
                    int estimatedSize = AbstractRecords.estimateSizeInBytes(
                        currentBatch.builder.magic(),
                        compression.type(),
                        recordsToAppend
                    );

                    // Check if the current batch has enough space. We check this before
                    // replaying the records in order to avoid having to revert back
                    // changes if the records do not fit within a batch.
                    if (estimatedSize > currentBatch.builder.maxAllowedBytes()) {
                        throw new RecordTooLargeException("Message batch size is " + estimatedSize +
                            " bytes in append to partition " + tp + " which exceeds the maximum " +
                            "configured size of " + currentBatch.maxBatchSize + ".");
                    }

                    if (!currentBatch.builder.hasRoomFor(estimatedSize)) {
                        // Otherwise, we write the current batch, allocate a new one and re-verify
                        // whether the records fit in it.
                        // If flushing fails, we don't catch the exception in order to let
                        // the caller fail the current operation.
                        flushCurrentBatch();
                        maybeAllocateNewBatch(
                            producerId,
                            producerEpoch,
                            verificationGuard,
                            currentTimeMs
                        );
                    }
                }


                for (int i = 0; i < records.size(); i++) {
                    U recordToReplay = records.get(i);
                    SimpleRecord recordToAppend = recordsToAppend.get(i);

                    if (!isAtomic) {
                        // Check if the current batch has enough space. We check this before
                        // replaying the record in order to avoid having to revert back
                        // changes if the record do not fit within a batch.
                        boolean hasRoomFor = currentBatch.builder.hasRoomFor(
                            recordToAppend.timestamp(),
                            recordToAppend.key(),
                            recordToAppend.value(),
                            recordToAppend.headers()
                        );

                        if (!hasRoomFor) {
                            // If flushing fails, we don't catch the exception in order to let
                            // the caller fail the current operation.
                            flushCurrentBatch();
                            maybeAllocateNewBatch(
                                producerId,
                                producerEpoch,
                                verificationGuard,
                                currentTimeMs
                            );
                        }
                    }

                    try {
                        if (replay) {
                            coordinator.replay(
                                currentBatch.nextOffset,
                                producerId,
                                producerEpoch,
                                recordToReplay
                            );
                        }

                        currentBatch.builder.append(recordToAppend);
                        currentBatch.nextOffset++;
                    } catch (Throwable t) {
                        log.error("Replaying record {} to {} failed due to: {}.", recordToReplay, tp, t.getMessage());

                        // Add the event to the list of pending events associated with the last
                        // batch in order to fail it too.
                        currentBatch.deferredEvents.add(event);

                        // If an exception is thrown, we fail the entire batch. Exceptions should be
                        // really exceptional in this code path and they would usually be the results
                        // of bugs preventing records to be replayed.
                        failCurrentBatch(t);

                        return;
                    }
                }

                // Add the event to the list of pending events associated with the batch.
                currentBatch.deferredEvents.add(event);

                // Write the current batch if it is transactional or if the linger timeout
                // has expired.
                // If flushing fails, we don't catch the exception in order to let
                // the caller fail the current operation.
                maybeFlushCurrentBatch(currentTimeMs);
            }
        }
Appends records to the log and replay them to the state machine.

@param producerId        The producer id.
@param producerEpoch     The producer epoch.
@param verificationGuard The verification guard.
@param records           The records to append.
@param replay            A boolean indicating whether the records
                         must be replayed or not.
@param isAtomic          A boolean indicating whether the records
                         must be written atomically or not.
@param event             The event that must be completed when the
                         records are written.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java