Repository: apache/kafka
File: core/src/main/java/kafka/server/share/SharePartition.java (Java)

Code before:
                        groupId, topicIdPartition, result);
                    throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                    return;
                }

                TopicData<PartitionAllData> state = result.topicsData().get(0);
                if (state.topicId() != topicIdPartition.topicId() || state.partitions().size() != 1) {
                    log.error("Failed to initialize the share partition: {}-{}. Invalid topic partition response: {}.",
                        groupId, topicIdPartition, result);
                    throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                    return;
                }

                PartitionAllData partitionData = state.partitions().get(0);
                if (partitionData.partition() != topicIdPartition.partition()) {
                    log.error("Failed to initialize the share partition: {}-{}. Invalid partition response: {}.",
                        groupId, topicIdPartition, partitionData);
                    throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                    return;
                }

                if (partitionData.errorCode() != Errors.NONE.code()) {
                    KafkaException ex = fetchPersisterError(partitionData.errorCode(), partitionData.errorMessage());
                    log.debug("Failed to initialize the share partition: {}-{}. Exception occurred: {}.",
                        groupId, topicIdPartition, partitionData);
                    if (!(ex instanceof UnknownServerException)) {
                        log.error("Failed to initialize the share partition: {}-{}. Exception occurred: {}.",
                            groupId, topicIdPartition, partitionData);
                    }
                    throwable = ex;
                    return;
                }

                startOffset = startOffsetDuringInitialization(partitionData.startOffset());
                stateEpoch = partitionData.stateEpoch();

                List<PersisterStateBatch> stateBatches = partitionData.stateBatches();
                long gapStartOffset = -1;
                // The previousBatchLastOffset is used to track the last offset of the previous batch.
                // For the first batch that should ideally start from startOffset if there are no gaps,
                // we assume the previousBatchLastOffset to be startOffset - 1.
                long previousBatchLastOffset = startOffset - 1;
                for (PersisterStateBatch stateBatch : stateBatches) {
                    if (stateBatch.firstOffset() < startOffset) {
                        log.error("Invalid state batch found for the share partition: {}-{}. The base offset: {}"
                                + " is less than the start offset: {}.", groupId, topicIdPartition,
                            stateBatch.firstOffset(), startOffset);
                        throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                        return;
                    }
                    if (gapStartOffset == -1 && stateBatch.firstOffset() > previousBatchLastOffset + 1) {

Review comment (bug):
This doesn't seems to be correct, so if the error code is 13 then only we should not log the eror message. Example log from persister.

```
Failed to initialize the share partition: <.......>. Exception occurred: PartitionData(partition=144,stateEpoch=0,startOffset=0,errorCode=13,errorMessage=Error in read state RPC. The server disconnected before a response was received.,leaderEpoch=0,stateBatches=[]).

Code after:
                    log.error("Failed to initialize the share partition: {}-{}. Invalid state found: {}.",
                        groupId, topicIdPartition, result);
                    throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                    return;
                }

                TopicData<PartitionAllData> state = result.topicsData().get(0);
                if (state.topicId() != topicIdPartition.topicId() || state.partitions().size() != 1) {
                    log.error("Failed to initialize the share partition: {}-{}. Invalid topic partition response: {}.",
                        groupId, topicIdPartition, result);
                    throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                    return;
                }

                PartitionAllData partitionData = state.partitions().get(0);
                if (partitionData.partition() != topicIdPartition.partition()) {
                    log.error("Failed to initialize the share partition: {}-{}. Invalid partition response: {}.",
                        groupId, topicIdPartition, partitionData);
                    throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                    return;
                }

                if (partitionData.errorCode() != Errors.NONE.code()) {
                    KafkaException ex = fetchPersisterError(partitionData.errorCode(), partitionData.errorMessage());
                    logError(String.format("Failed to initialize the share partition: %s-%s. Exception occurred: %s.",
                        groupId, topicIdPartition, partitionData), ex);
                    throwable = ex;
                    return;
                }

                startOffset = startOffsetDuringInitialization(partitionData.startOffset());
                stateEpoch = partitionData.stateEpoch();

                List<PersisterStateBatch> stateBatches = partitionData.stateBatches();
                long gapStartOffset = -1;
                // The previousBatchLastOffset is used to track the last offset of the previous batch.
                // For the first batch that should ideally start from startOffset if there are no gaps,
                // we assume the previousBatchLastOffset to be startOffset - 1.
                long previousBatchLastOffset = startOffset - 1;
                for (PersisterStateBatch stateBatch : stateBatches) {
                    if (stateBatch.firstOffset() < startOffset) {
                        log.error("Invalid state batch found for the share partition: {}-{}. The base offset: {}"
                                + " is less than the start offset: {}.", groupId, topicIdPartition,
                            stateBatch.firstOffset(), startOffset);
                        throwable = new IllegalStateException(String.format("Failed to initialize the share partition %s-%s", groupId, topicIdPartition));
                        return;
                    }
                    if (gapStartOffset == -1 && stateBatch.firstOffset() > previousBatchLastOffset + 1) {
                        gapStartOffset = previousBatchLastOffset + 1;
                    }
                    previousBatchLastOffset = stateBatch.lastOffset();