Skip to content

MINOR: Add read/write all operation #15462

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dajac merged 4 commits into from
Mar 7, 2024
Merged

MINOR: Add read/write all operation #15462

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
335e7eb
MINOR: Introduce multi read operation in coordiantor runtime
dajac Mar 4, 2024
8f818de
add write all op
dajac Mar 4, 2024
294a166
cleanup
dajac Mar 4, 2024
6b38320
refactor
dajac Mar 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 18 additions & 36 deletions ...coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,6 @@
import java.util.Map;
import java.util.OptionalInt;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
Expand Down Expand Up @@ -498,29 +497,17 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
);
}

final Set<TopicPartition> existingPartitionSet = runtime.partitions();

if (existingPartitionSet.isEmpty()) {
return CompletableFuture.completedFuture(new ListGroupsResponseData());
}

final List<CompletableFuture<List<ListGroupsResponseData.ListedGroup>>> futures =
new ArrayList<>();

for (TopicPartition tp : existingPartitionSet) {
futures.add(runtime.scheduleReadOperation(
"list-groups",
tp,
(coordinator, lastCommittedOffset) -> coordinator.listGroups(request.statesFilter(), request.typesFilter(), lastCommittedOffset)
).exceptionally(exception -> {
exception = Errors.maybeUnwrapException(exception);
if (exception instanceof NotCoordinatorException) {
return Collections.emptyList();
} else {
throw new CompletionException(exception);
}
}));
}
final List<CompletableFuture<List<ListGroupsResponseData.ListedGroup>>> futures = runtime.scheduleReadAllOperation(
"list-groups",
(coordinator, lastCommittedOffset) -> coordinator.listGroups(request.statesFilter(), request.typesFilter(), lastCommittedOffset)
).stream().map(future -> future.exceptionally(exception -> {

@chia7712 chia7712 Mar 5, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how about adding the exception function to scheduleReadAllOperation? It brings two benefits:

  1. simplify the code from callers
  2. avoid creating list repeatedly

@dajac dajac Mar 6, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I actually considered this but eventually rejected it because those new methods could also be used with a different handling than exceptionally. The separation of concerns would not be too good in my opinion. I was also considering adding an helper like mapExceptionally to add an exception handler on a list of futures. Do you think that it could also help?

@chia7712 chia7712 Mar 6, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

adding an helper like mapExceptionally to add an exception handler on a list of futures

Do you mean to enhance FutureUtils#combineFutures to take one more parameter? Current PR is good to me but I'd like to see more design (mapExceptionally) before merging :)

@dajac dajac Mar 7, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just pushed a commit with the mapExceptionally construct that I was thinking about. Please let me know what you think.

exception = Errors.maybeUnwrapException(exception);
if (exception instanceof NotCoordinatorException) {
return Collections.emptyList();
} else {
throw new CompletionException(exception);
}
})).collect(Collectors.toList());

return FutureUtils
.combineFutures(futures, ArrayList::new, List::addAll)
Expand Down Expand Up @@ -963,23 +950,18 @@ public void onPartitionsDeleted(
) throws ExecutionException, InterruptedException {
throwIfNotActive();

final Set<TopicPartition> existingPartitionSet = runtime.partitions();
final List<CompletableFuture<Void>> futures = new ArrayList<>(existingPartitionSet.size());

existingPartitionSet.forEach(partition -> futures.add(
runtime.scheduleWriteOperation(
CompletableFuture.allOf(
runtime.scheduleWriteAllOperation(
"on-partition-deleted",
partition,
Duration.ofMillis(config.offsetCommitTimeoutMs),
coordinator -> coordinator.onPartitionsDeleted(topicPartitions)
).exceptionally(exception -> {
log.error("Could not delete offsets for deleted partitions {} in coordinator {} due to: {}.",
partition, partition, exception.getMessage(), exception);
).stream().map(future -> future.exceptionally(exception -> {
log.error("Could not delete offsets for deleted partitions {} due to: {}.",
topicPartitions, exception.getMessage(), exception
);
return null;
})
));

CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
).toArray(CompletableFuture[]::new)).get();
}

/**
Expand Down
67 changes: 54 additions & 13 deletions ...rdinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,19 +41,19 @@

import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
* The CoordinatorRuntime provides a framework to implement coordinators such as the group coordinator
Expand Down Expand Up @@ -1446,6 +1446,32 @@ public <T> CompletableFuture<T> scheduleWriteOperation(
return event.future;
}

/**
* Schedule a write operation for each coordinator.
*
* @param name The name of the write operation.
* @param timeout The write operation timeout.
* @param op The write operation.
*
* @return A list of futures where each future will be completed with the result of the write operation
* when the operation is completed or an exception if the write operation failed.
*
* @param <T> The type of the result.
*/
public <T> List<CompletableFuture<T>> scheduleWriteAllOperation(
String name,
Duration timeout,
CoordinatorWriteOperation<S, T, U> op
) {
throwIfNotRunning();
log.debug("Scheduled execution of write all operation {}.", name);
return coordinators
.keySet()
.stream()
.map(tp -> scheduleWriteOperation(name, tp, timeout, op))
.collect(Collectors.toList());
}

/**
* Schedules a transactional write operation.
*
Expand Down Expand Up @@ -1535,12 +1561,12 @@ public CompletableFuture<Void> scheduleTransactionCompletion(
/**
* Schedules a read operation.
*
* @param name The name of the write operation.
* @param name The name of the read operation.
* @param tp The address of the coordinator (aka its topic-partitions).
* @param op The read operation.
*
* @return A future that will be completed with the result of the read operation
* when the operation is completed or an exception if the write operation failed.
* when the operation is completed or an exception if the read operation failed.
*
* @param <T> The type of the result.
*/
Expand All @@ -1556,6 +1582,30 @@ public <T> CompletableFuture<T> scheduleReadOperation(
return event.future;
}

/**
* Schedules a read operation for each coordinator.
*
* @param name The name of the read operation.
* @param op The read operation.
*
* @return A list of futures where each future will be completed with the result of the read operation
* when the operation is completed or an exception if the read operation failed.
*
* @param <T> The type of the result.
*/
public <T> List<CompletableFuture<T>> scheduleReadAllOperation(
String name,
CoordinatorReadOperation<S, T> op
) {
throwIfNotRunning();
log.debug("Scheduled execution of read all operation {}.", name);
return coordinators
.keySet()
.stream()
.map(tp -> scheduleReadOperation(name, tp, op))
.collect(Collectors.toList());
}

/**
* Schedules an internal event.
*
Expand All @@ -1572,15 +1622,6 @@ private void scheduleInternalOperation(
enqueue(new CoordinatorInternalEvent(name, tp, op));
}

/**
* @return The topic partitions of the coordinators currently registered in the
* runtime.
*/
public Set<TopicPartition> partitions() {
throwIfNotRunning();
return new HashSet<>(coordinators.keySet());
}

/**
* Schedules the loading of a coordinator. This is called when the broker is elected as
* the leader for a partition.
Expand Down
Loading