You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2024/10/17 22:32:48 UTC
[PR] KAFKA-17545: Removing process fetch queue [kafka]
apoorvmittal10 opened a new pull request, #17534:
URL: https://github.com/apache/kafka/pull/17534
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1807763887
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -547,31 +522,11 @@ private static String partitionsToLogString(Collection<TopicIdPartition> partiti
return ShareSession.partitionsToLogString(partitions, log.isTraceEnabled());
}
- /**
- * Recursive function to process all the fetch requests present inside the fetch queue
- */
// Visible for testing.
- void maybeProcessFetchQueue() {
- if (!acquireProcessFetchQueueLock()) {
- // The queue is already being processed hence avoid re-triggering.
- return;
- }
-
- ShareFetchData shareFetchData = fetchQueue.poll();
- if (shareFetchData == null) {
- // No more requests to process, so release the lock. Though we should not reach here as the lock
- // is acquired only when there are requests in the queue. But still, it's safe to release the lock.
- releaseProcessFetchQueueLock();
- return;
- }
-
+ void processShareFetch(ShareFetchData shareFetchData) {
Review Comment:
Done. Though I am reafactoring the initilization handling in Share Partition Manager as we need to address the concern when Share Partition is loaded late and request is added to purgatory, etc. The PR will be out soon.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1809589222
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -611,20 +564,9 @@ void maybeProcessFetchQueue() {
// Add the share fetch to the delayed share fetch purgatory to process the fetch request.
addDelayedShareFetch(new DelayedShareFetch(shareFetchData, replicaManager, this),
delayedShareFetchWatchKeys);
-
- // Release the lock so that other threads can process the queue.
- releaseProcessFetchQueueLock();
- // If there are more requests in the queue, then process them.
- if (!fetchQueue.isEmpty())
- maybeProcessFetchQueue();
-
} catch (Exception e) {
// In case exception occurs then release the locks so queue can be further processed.
log.error("Error processing fetch queue for share partitions", e);
- releaseProcessFetchQueueLock();
Review Comment:
Could we remove the comment in the catch clause? Also, is the code really throwing any exception? If not, we can just remove the catch clause.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #17534:
URL: https://github.com/apache/kafka/pull/17534#issuecomment-2421848473
Future tasks associated with refactor:
[KAFKA-17829](https://issues.apache.org/jira/browse/KAFKA-17829) - We need to add an integration test which verifies that on shutdown of the delayed share fetch purgatory, the share fetch requests which are present inside the purgatory return with an erroneous future.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1807763516
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -184,30 +172,28 @@ private SharePartitionManager(
GroupConfigManager groupConfigManager,
Metrics metrics
) {
- this.replicaManager = replicaManager;
- this.time = time;
- this.cache = cache;
- this.partitionCacheMap = partitionCacheMap;
- this.fetchQueue = new ConcurrentLinkedQueue<>();
- this.processFetchQueueLock = new AtomicBoolean(false);
- this.defaultRecordLockDurationMs = defaultRecordLockDurationMs;
- this.timer = new SystemTimerReaper("share-group-lock-timeout-reaper",
- new SystemTimer("share-group-lock-timeout"));
- this.maxDeliveryCount = maxDeliveryCount;
- this.maxInFlightMessages = maxInFlightMessages;
- this.persister = persister;
- this.groupConfigManager = groupConfigManager;
- this.shareGroupMetrics = new ShareGroupMetrics(Objects.requireNonNull(metrics), time);
+ this(
+ replicaManager,
Review Comment:
Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1815560941
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -611,20 +564,9 @@ void maybeProcessFetchQueue() {
// Add the share fetch to the delayed share fetch purgatory to process the fetch request.
addDelayedShareFetch(new DelayedShareFetch(shareFetchData, replicaManager, this),
delayedShareFetchWatchKeys);
-
- // Release the lock so that other threads can process the queue.
- releaseProcessFetchQueueLock();
- // If there are more requests in the queue, then process them.
- if (!fetchQueue.isEmpty())
- maybeProcessFetchQueue();
-
} catch (Exception e) {
// In case exception occurs then release the locks so queue can be further processed.
log.error("Error processing fetch queue for share partitions", e);
- releaseProcessFetchQueueLock();
Review Comment:
Thanks. Also, `SharePartition.maybeAcquireFetchLock` still has comments on fetch queue.
```
/**
* Prior to fetching records from the leader, the fetch lock is acquired to ensure that the same
* share partition does not enter a fetch queue while another one is being fetched within the queue.
* The fetch lock is released once the records are fetched from the leader.
*
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #17534:
URL: https://github.com/apache/kafka/pull/17534
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1810815657
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -611,20 +564,9 @@ void maybeProcessFetchQueue() {
// Add the share fetch to the delayed share fetch purgatory to process the fetch request.
addDelayedShareFetch(new DelayedShareFetch(shareFetchData, replicaManager, this),
delayedShareFetchWatchKeys);
-
- // Release the lock so that other threads can process the queue.
- releaseProcessFetchQueueLock();
- // If there are more requests in the queue, then process them.
- if (!fetchQueue.isEmpty())
- maybeProcessFetchQueue();
-
} catch (Exception e) {
// In case exception occurs then release the locks so queue can be further processed.
log.error("Error processing fetch queue for share partitions", e);
- releaseProcessFetchQueueLock();
Review Comment:
Sure @junrao, I am adding an additional handling where the catch block is required. I ll send the changes in open partition leader PR: https://github.com/apache/kafka/pull/16842
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1806854706
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -547,31 +522,11 @@ private static String partitionsToLogString(Collection<TopicIdPartition> partiti
return ShareSession.partitionsToLogString(partitions, log.isTraceEnabled());
}
- /**
- * Recursive function to process all the fetch requests present inside the fetch queue
- */
// Visible for testing.
- void maybeProcessFetchQueue() {
- if (!acquireProcessFetchQueueLock()) {
- // The queue is already being processed hence avoid re-triggering.
- return;
- }
-
- ShareFetchData shareFetchData = fetchQueue.poll();
- if (shareFetchData == null) {
- // No more requests to process, so release the lock. Though we should not reach here as the lock
- // is acquired only when there are requests in the queue. But still, it's safe to release the lock.
- releaseProcessFetchQueueLock();
- return;
- }
-
+ void processShareFetch(ShareFetchData shareFetchData) {
Review Comment:
While you are at this class, in the following code, `return` seems unnecessary.
```
sharePartition.maybeInitialize().whenComplete((result, throwable) -> {
if (throwable != null) {
maybeCompleteInitializationWithException(sharePartitionKey, shareFetchData.future(), throwable);
return;
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17545: Removing process fetch queue [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17534:
URL: https://github.com/apache/kafka/pull/17534#discussion_r1806849633
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -184,30 +172,28 @@ private SharePartitionManager(
GroupConfigManager groupConfigManager,
Metrics metrics
) {
- this.replicaManager = replicaManager;
- this.time = time;
- this.cache = cache;
- this.partitionCacheMap = partitionCacheMap;
- this.fetchQueue = new ConcurrentLinkedQueue<>();
- this.processFetchQueueLock = new AtomicBoolean(false);
- this.defaultRecordLockDurationMs = defaultRecordLockDurationMs;
- this.timer = new SystemTimerReaper("share-group-lock-timeout-reaper",
- new SystemTimer("share-group-lock-timeout"));
- this.maxDeliveryCount = maxDeliveryCount;
- this.maxInFlightMessages = maxInFlightMessages;
- this.persister = persister;
- this.groupConfigManager = groupConfigManager;
- this.shareGroupMetrics = new ShareGroupMetrics(Objects.requireNonNull(metrics), time);
+ this(
+ replicaManager,
Review Comment:
merge with previous line?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org