chia7712 commented on code in PR #17957:
URL: https://github.com/apache/kafka/pull/17957#discussion_r1864888847
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -389,79 +389,84 @@ public CompletableFuture<Void> maybeInitialize() {
.build())
.build()
).whenComplete((result, exception) -> {
- if (exception != null) {
- log.error("Failed to initialize the share partition:
{}-{}", groupId, topicIdPartition, exception);
- completeInitializationWithException(future, exception);
- return;
- }
+ lock.writeLock().lock();
+ try {
+ if (exception != null) {
+ log.error("Failed to initialize the share partition:
{}-{}", groupId, topicIdPartition, exception);
+ completeInitializationWithException(future, exception);
+ return;
+ }
- if (result == null || result.topicsData() == null ||
result.topicsData().size() != 1) {
- log.error("Failed to initialize the share partition:
{}-{}. Invalid state found: {}.",
- groupId, topicIdPartition, result);
- completeInitializationWithException(future, new
IllegalStateException(String.format("Failed to initialize the share partition
%s-%s", groupId, topicIdPartition)));
- return;
- }
+ if (result == null || result.topicsData() == null ||
result.topicsData().size() != 1) {
+ log.error("Failed to initialize the share partition:
{}-{}. Invalid state found: {}.",
+ groupId, topicIdPartition, result);
+ completeInitializationWithException(future, 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);
- completeInitializationWithException(future, 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);
+ completeInitializationWithException(future, 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);
- completeInitializationWithException(future, 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);
+ completeInitializationWithException(future, 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.error("Failed to initialize the share partition:
{}-{}. Exception occurred: {}.",
- groupId, topicIdPartition, partitionData);
- completeInitializationWithException(future, ex);
- return;
- }
+ if (partitionData.errorCode() != Errors.NONE.code()) {
+ KafkaException ex =
fetchPersisterError(partitionData.errorCode(), partitionData.errorMessage());
+ log.error("Failed to initialize the share partition:
{}-{}. Exception occurred: {}.",
+ groupId, topicIdPartition, partitionData);
+ completeInitializationWithException(future, ex);
+ return;
+ }
- try {
- startOffset =
startOffsetDuringInitialization(partitionData.startOffset());
- } catch (Exception e) {
- completeInitializationWithException(future, e);
- return;
- }
- stateEpoch = partitionData.stateEpoch();
-
- List<PersisterStateBatch> stateBatches =
partitionData.stateBatches();
- 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);
- completeInitializationWithException(future, new
IllegalStateException(String.format("Failed to initialize the share partition
%s-%s", groupId, topicIdPartition)));
+ try {
+ startOffset =
startOffsetDuringInitialization(partitionData.startOffset());
+ } catch (Exception e) {
+ completeInitializationWithException(future, e);
return;
}
- InFlightBatch inFlightBatch = new
InFlightBatch(EMPTY_MEMBER_ID, stateBatch.firstOffset(),
- stateBatch.lastOffset(),
RecordState.forId(stateBatch.deliveryState()), stateBatch.deliveryCount(),
null);
- cachedState.put(stateBatch.firstOffset(), inFlightBatch);
- }
- // Update the endOffset of the partition.
- if (!cachedState.isEmpty()) {
- // If the cachedState is not empty, findNextFetchOffset
flag is set to true so that any AVAILABLE records
- // in the cached state are not missed
- findNextFetchOffset.set(true);
- endOffset =
cachedState.lastEntry().getValue().lastOffset();
- // In case the persister read state RPC result contains no
AVAILABLE records, we can update cached state
- // and start/end offsets.
- maybeUpdateCachedStateAndOffsets();
- } else {
- endOffset = startOffset;
+ stateEpoch = partitionData.stateEpoch();
+
+ List<PersisterStateBatch> stateBatches =
partitionData.stateBatches();
+ 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);
+ completeInitializationWithException(future, new
IllegalStateException(String.format("Failed to initialize the share partition
%s-%s", groupId, topicIdPartition)));
+ return;
+ }
+ InFlightBatch inFlightBatch = new
InFlightBatch(EMPTY_MEMBER_ID, stateBatch.firstOffset(),
+ stateBatch.lastOffset(),
RecordState.forId(stateBatch.deliveryState()), stateBatch.deliveryCount(),
null);
+ cachedState.put(stateBatch.firstOffset(),
inFlightBatch);
+ }
+ // Update the endOffset of the partition.
+ if (!cachedState.isEmpty()) {
+ // If the cachedState is not empty,
findNextFetchOffset flag is set to true so that any AVAILABLE records
+ // in the cached state are not missed
+ findNextFetchOffset.set(true);
+ endOffset =
cachedState.lastEntry().getValue().lastOffset();
+ // In case the persister read state RPC result
contains no AVAILABLE records, we can update cached state
+ // and start/end offsets.
+ maybeUpdateCachedStateAndOffsets();
+ } else {
+ endOffset = startOffset;
+ }
+ // Set the partition state to Active and complete the
future.
+ partitionState = SharePartitionState.ACTIVE;
+ future.complete(null);
Review Comment:
> I'll make these change as part of issue
https://issues.apache.org/jira/browse/KAFKA-18129 .
It seems KAFKA-18129 is intended to add a write lock to `maybeInitialize`.
However, the additional discussion focuses on the appropriate scope of the
write lock, particularly concerning the potential for deadlocks. If we can't
reach a consensus on the lock scope, we could still add the write lock to
`maybeInitialize` in this PR. It doesn't seem like a significant change to make
at this point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
