Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java (Java)

Code before:
            File producerStateSnapshotFile = log.producerStateManager().fetchSnapshot(nextSegmentBaseOffset).orElse(null);

            List<EpochEntry> epochEntries = getLeaderEpochEntries(log, segment.baseOffset(), nextSegmentBaseOffset);
            Map<Integer, Long> segmentLeaderEpochs = new HashMap<>(epochEntries.size());
            epochEntries.forEach(entry -> segmentLeaderEpochs.put(entry.epoch, entry.startOffset));

            boolean isTxnIdxEmpty = segment.txnIndex().isEmpty();
            RemoteLogSegmentMetadata copySegmentStartedRlsm = new RemoteLogSegmentMetadata(segmentId, segment.baseOffset(), endOffset,
                    segment.largestTimestamp(), brokerId, time.milliseconds(), segment.log().sizeInBytes(),
                    segmentLeaderEpochs, isTxnIdxEmpty);

            remoteLogMetadataManagerPlugin.get().addRemoteLogSegmentMetadata(copySegmentStartedRlsm).get();

            ByteBuffer leaderEpochsIndex = epochEntriesAsByteBuffer(getLeaderEpochEntries(log, -1, nextSegmentBaseOffset));
            LogSegmentData segmentData = new LogSegmentData(logFile.toPath(), toPathIfExists(segment.offsetIndex().file()),
                    toPathIfExists(segment.timeIndex().file()), Optional.ofNullable(toPathIfExists(segment.txnIndex().file())),
                    producerStateSnapshotFile.toPath(), leaderEpochsIndex);
            brokerTopicStats.topicStats(log.topicPartition().topic()).remoteCopyRequestRate().mark();
            brokerTopicStats.allTopicsStats().remoteCopyRequestRate().mark();
            Optional<CustomMetadata> customMetadata;
            
            try {
                customMetadata = remoteStorageManagerPlugin.get().copyLogSegmentData(copySegmentStartedRlsm, segmentData);
            } catch (RemoteStorageException e) {
                logger.info("Copy failed, cleaning segment {}", copySegmentStartedRlsm.remoteLogSegmentId());
                recordLagStats(log);
                try {
                    deleteRemoteLogSegment(copySegmentStartedRlsm, ignored -> !isCancelled());
                    LOGGER.info("Cleanup completed for segment {}", copySegmentStartedRlsm.remoteLogSegmentId());
                } catch (RemoteStorageException e1) {
                    LOGGER.info("Cleanup failed, will retry later with segment {}: {}", copySegmentStartedRlsm.remoteLogSegmentId(), e1.getMessage());
                }
                throw e;
            }

            RemoteLogSegmentMetadataUpdate copySegmentFinishedRlsm = new RemoteLogSegmentMetadataUpdate(segmentId, time.milliseconds(),
                    customMetadata, RemoteLogSegmentState.COPY_SEGMENT_FINISHED, brokerId);

            if (customMetadata.isPresent()) {
                long customMetadataSize = customMetadata.get().value().length;
                if (customMetadataSize > this.customMetadataSizeLimit) {
                    CustomMetadataSizeLimitExceededException e = new CustomMetadataSizeLimitExceededException();
                    logger.info("Custom metadata size {} exceeds configured limit {}." +
                                    " Copying will be stopped and copied segment will be attempted to clean." +
                                    " Original metadata: {}",
                            customMetadataSize, this.customMetadataSizeLimit, copySegmentStartedRlsm, e);
                    // For deletion, we provide back the custom metadata by creating a new metadata object from the update.
                    // However, the update itself will not be stored in this case.
                    RemoteLogSegmentMetadata newMetadata = copySegmentStartedRlsm.createWithUpdates(copySegmentFinishedRlsm);
                    try {
                        deleteRemoteLogSegment(newMetadata, ignored -> !isCancelled());

Review comment (suggestion):
This patch handles the error when interacting with the RemoteStorageManager. Should we also handle any error when interacting with RemoteLogMetadataManager? 

The segmentsLag / bytesLag metric should show the real-time value whenever there is any error in interacting with RSM or RLMM plugin.  

Code after:
            }
        }

        private void copyLogSegment(UnifiedLog log, LogSegment segment, RemoteLogSegmentId segmentId, long nextSegmentBaseOffset)
                throws InterruptedException, ExecutionException, RemoteStorageException, IOException,
                CustomMetadataSizeLimitExceededException {
            File logFile = segment.log().file();
            String logFileName = logFile.getName();

            logger.info("Copying {} to remote storage.", logFileName);

            long endOffset = nextSegmentBaseOffset - 1;
            File producerStateSnapshotFile = log.producerStateManager().fetchSnapshot(nextSegmentBaseOffset).orElse(null);

            List<EpochEntry> epochEntries = getLeaderEpochEntries(log, segment.baseOffset(), nextSegmentBaseOffset);
            Map<Integer, Long> segmentLeaderEpochs = new HashMap<>(epochEntries.size());
            epochEntries.forEach(entry -> segmentLeaderEpochs.put(entry.epoch, entry.startOffset));

            boolean isTxnIdxEmpty = segment.txnIndex().isEmpty();
            RemoteLogSegmentMetadata copySegmentStartedRlsm = new RemoteLogSegmentMetadata(segmentId, segment.baseOffset(), endOffset,
                    segment.largestTimestamp(), brokerId, time.milliseconds(), segment.log().sizeInBytes(),
                    segmentLeaderEpochs, isTxnIdxEmpty);

            remoteLogMetadataManagerPlugin.get().addRemoteLogSegmentMetadata(copySegmentStartedRlsm).get();

            ByteBuffer leaderEpochsIndex = epochEntriesAsByteBuffer(getLeaderEpochEntries(log, -1, nextSegmentBaseOffset));
            LogSegmentData segmentData = new LogSegmentData(logFile.toPath(), toPathIfExists(segment.offsetIndex().file()),
                    toPathIfExists(segment.timeIndex().file()), Optional.ofNullable(toPathIfExists(segment.txnIndex().file())),
                    producerStateSnapshotFile.toPath(), leaderEpochsIndex);
            brokerTopicStats.topicStats(log.topicPartition().topic()).remoteCopyRequestRate().mark();
            brokerTopicStats.allTopicsStats().remoteCopyRequestRate().mark();
            Optional<CustomMetadata> customMetadata;
            
            try {
                customMetadata = remoteStorageManagerPlugin.get().copyLogSegmentData(copySegmentStartedRlsm, segmentData);
            } catch (RemoteStorageException e) {
                logger.info("Copy failed, cleaning segment {}", copySegmentStartedRlsm.remoteLogSegmentId());
                try {
                    deleteRemoteLogSegment(copySegmentStartedRlsm, ignored -> !isCancelled());
                    LOGGER.info("Cleanup completed for segment {}", copySegmentStartedRlsm.remoteLogSegmentId());
                } catch (RemoteStorageException e1) {
                    LOGGER.info("Cleanup failed, will retry later with segment {}: {}", copySegmentStartedRlsm.remoteLogSegmentId(), e1.getMessage());
                }
                throw e;
            }

            RemoteLogSegmentMetadataUpdate copySegmentFinishedRlsm = new RemoteLogSegmentMetadataUpdate(segmentId, time.milliseconds(),
                    customMetadata, RemoteLogSegmentState.COPY_SEGMENT_FINISHED, brokerId);

            if (customMetadata.isPresent()) {
                long customMetadataSize = customMetadata.get().value().length;