Skip to content

KAFKA-16100: Add timeout to all the CompletableApplicationEvents #15455

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
cadonna merged 5 commits into from
Mar 7, 2024
Merged

KAFKA-16100: Add timeout to all the CompletableApplicationEvents #15455

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a5c2bd9
KAFKA-16100: Add timeout to all the CompletableApplicationEvents
kirktrue Mar 1, 2024
69e1203
Reverting minor change to ApplicationEventProcessor for topic metadat…
kirktrue Mar 1, 2024
6ae929a
Merge remote-tracking branch 'origin/trunk' into KAFKA-16100-add-time…
kirktrue Mar 4, 2024
60682e0
Addressing review comments
kirktrue Mar 4, 2024
b075b29
Minor renaming and refactoring in CommitEvent for clarity
kirktrue Mar 5, 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
44 changes: 26 additions & 18 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,7 +769,8 @@ public void commitAsync(OffsetCommitCallback callback) {
public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
acquireAndEnsureOpen();
try {
AsyncCommitEvent asyncCommitEvent = new AsyncCommitEvent(offsets);
final Timer timer = time.timer(Long.MAX_VALUE);
AsyncCommitEvent asyncCommitEvent = new AsyncCommitEvent(offsets, timer);
CompletableFuture<Void> future = commit(asyncCommitEvent);
future.whenComplete((r, t) -> {

Expand Down Expand Up @@ -937,13 +938,14 @@ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition
return Collections.emptyMap();
}

final Timer timer = time.timer(timeout);
final FetchCommittedOffsetsEvent event = new FetchCommittedOffsetsEvent(
partitions,
timeout.toMillis());
timer);
wakeupTrigger.setActiveTask(event.future());
try {
final Map<TopicPartition, OffsetAndMetadata> committedOffsets = applicationEventHandler.addAndGet(event,
time.timer(timeout));
timer);

@cadonna cadonna Mar 4, 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.

I have two questions/comments here:

  1. Could you also not pass the timer to addAndGet() and use the timer of the event? To me, it seems like the timer passed to the event is always the timer also passed to addAndGet().
  2. Is it correct to use timer.remainingMs() when waiting for the future of the event? Do we not risk to timeout the future before we check the timer for expiration?

@kirktrue kirktrue Mar 4, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

You're correct on both counts! 😄

  1. Could you also not pass the timer to addAndGet() and use the timer of the event? To me, it seems like the timer passed to the event is always the timer also passed to addAndGet().

Indeed it is the same Timer object. I intend to resolve that confusion as part of KAFKA-15974 and to keep this one smaller in scope.

  1. Is it correct to use timer.remainingMs() when waiting for the future of the event? Do we not risk to timeout the future before we check the timer for expiration?

It is not, but the proper fix to change the underlying design is more substantial, so I was saving that change as part of KAFKA-15974.

committedOffsets.forEach(this::updateLastSeenEpochIfNewer);
return committedOffsets;
} catch (TimeoutException e) {
Expand Down Expand Up @@ -989,11 +991,12 @@ public List<PartitionInfo> partitionsFor(String topic, Duration timeout) {
throw new TimeoutException();
}

final TopicMetadataEvent topicMetadataEvent = new TopicMetadataEvent(topic, timeout.toMillis());
final Timer timer = time.timer(timeout);
final TopicMetadataEvent topicMetadataEvent = new TopicMetadataEvent(topic, timer);
wakeupTrigger.setActiveTask(topicMetadataEvent.future());
try {
Map<String, List<PartitionInfo>> topicMetadata =
applicationEventHandler.addAndGet(topicMetadataEvent, time.timer(timeout));
applicationEventHandler.addAndGet(topicMetadataEvent, timer);

return topicMetadata.getOrDefault(topic, Collections.emptyList());
} finally {
Expand All @@ -1017,10 +1020,11 @@ public Map<String, List<PartitionInfo>> listTopics(Duration timeout) {
throw new TimeoutException();
}

final AllTopicsMetadataEvent topicMetadataEvent = new AllTopicsMetadataEvent(timeout.toMillis());
final Timer timer = time.timer(timeout);
final AllTopicsMetadataEvent topicMetadataEvent = new AllTopicsMetadataEvent(timer);
wakeupTrigger.setActiveTask(topicMetadataEvent.future());
try {
return applicationEventHandler.addAndGet(topicMetadataEvent, time.timer(timeout));
return applicationEventHandler.addAndGet(topicMetadataEvent, timer);
} finally {
wakeupTrigger.clearTask();
}
Expand Down Expand Up @@ -1088,16 +1092,18 @@ public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartitio
if (timestampsToSearch.isEmpty()) {
return Collections.emptyMap();
}
final Timer timer = time.timer(timeout);
final ListOffsetsEvent listOffsetsEvent = new ListOffsetsEvent(
timestampsToSearch,
true);
true,
timer);

// If timeout is set to zero return empty immediately; otherwise try to get the results
// and throw timeout exception if it cannot complete in time.
if (timeout.toMillis() == 0L)
return listOffsetsEvent.emptyResult();

return applicationEventHandler.addAndGet(listOffsetsEvent, time.timer(timeout));
return applicationEventHandler.addAndGet(listOffsetsEvent, timer);
} finally {
release();
}
Expand Down Expand Up @@ -1138,12 +1144,14 @@ private Map<TopicPartition, Long> beginningOrEndOffset(Collection<TopicPartition
Map<TopicPartition, Long> timestampToSearch = partitions
.stream()
.collect(Collectors.toMap(Function.identity(), tp -> timestamp));
Timer timer = time.timer(timeout);
ListOffsetsEvent listOffsetsEvent = new ListOffsetsEvent(
timestampToSearch,
false);
false,
timer);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap = applicationEventHandler.addAndGet(
listOffsetsEvent,
time.timer(timeout));
timer);
return offsetAndTimestampMap
.entrySet()
.stream()
Expand Down Expand Up @@ -1273,7 +1281,7 @@ void prepareShutdown(final Timer timer, final AtomicReference<Throwable> firstEx
completeQuietly(
() -> {
maybeRevokePartitions();
applicationEventHandler.addAndGet(new LeaveOnCloseEvent(), timer);
applicationEventHandler.addAndGet(new LeaveOnCloseEvent(timer), timer);
},
"Failed to send leaveGroup heartbeat with a timeout(ms)=" + timer.timeoutMs(), firstException);
}
Expand Down Expand Up @@ -1350,7 +1358,7 @@ public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration
long commitStart = time.nanoseconds();
try {
Timer requestTimer = time.timer(timeout.toMillis());
SyncCommitEvent syncCommitEvent = new SyncCommitEvent(offsets, timeout.toMillis());
SyncCommitEvent syncCommitEvent = new SyncCommitEvent(offsets, requestTimer);
CompletableFuture<Void> commitFuture = commit(syncCommitEvent);
wakeupTrigger.setActiveTask(commitFuture);
ConsumerUtils.getResult(commitFuture, requestTimer);
Expand Down Expand Up @@ -1464,10 +1472,10 @@ public void unsubscribe() {
try {
fetchBuffer.retainAll(Collections.emptySet());
if (groupMetadata.isPresent()) {
UnsubscribeEvent unsubscribeEvent = new UnsubscribeEvent();
Timer timer = time.timer(Long.MAX_VALUE);
UnsubscribeEvent unsubscribeEvent = new UnsubscribeEvent(timer);
applicationEventHandler.add(unsubscribeEvent);
log.info("Unsubscribing all topics or patterns and assigned partitions");
Timer timer = time.timer(Long.MAX_VALUE);

try {
processBackgroundEvents(backgroundEventProcessor, unsubscribeEvent.future(), timer);
Expand Down Expand Up @@ -1568,7 +1576,7 @@ private boolean updateFetchPositions(final Timer timer) {
// Validate positions using the partition leader end offsets, to detect if any partition
// has been truncated due to a leader change. This will trigger an OffsetForLeaderEpoch
// request, retrieve the partition end offsets, and validate the current position against it.
applicationEventHandler.addAndGet(new ValidatePositionsEvent(), timer);
applicationEventHandler.addAndGet(new ValidatePositionsEvent(timer), timer);

cachedSubscriptionHasAllFetchPositions = subscriptions.hasAllFetchPositions();
if (cachedSubscriptionHasAllFetchPositions) return true;
Expand All @@ -1591,7 +1599,7 @@ private boolean updateFetchPositions(final Timer timer) {
// which are awaiting reset. This will trigger a ListOffset request, retrieve the
// partition offsets according to the strategy (ex. earliest, latest), and update the
// positions.
applicationEventHandler.addAndGet(new ResetPositionsEvent(), timer);
applicationEventHandler.addAndGet(new ResetPositionsEvent(timer), timer);
return true;
} catch (TimeoutException e) {
return false;
Expand Down Expand Up @@ -1624,7 +1632,7 @@ private boolean initWithCommittedOffsetsIfNeeded(Timer timer) {
final FetchCommittedOffsetsEvent event =
new FetchCommittedOffsetsEvent(
initializingPartitions,
timer.remainingMs());
timer);
final Map<TopicPartition, OffsetAndMetadata> offsets = applicationEventHandler.addAndGet(event, timer);
refreshCommittedOffsets(offsets, metadata, subscriptions);
return true;
Expand Down
17 changes: 3 additions & 14 deletions ...n/java/org/apache/kafka/clients/consumer/internals/events/AbstractTopicMetadataEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,25 +17,14 @@
package org.apache.kafka.clients.consumer.internals.events;

import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Timer;

import java.util.List;
import java.util.Map;

public abstract class AbstractTopicMetadataEvent extends CompletableApplicationEvent<Map<String, List<PartitionInfo>>> {

private final long timeoutMs;

protected AbstractTopicMetadataEvent(final Type type, final long timeoutMs) {
super(type);
this.timeoutMs = timeoutMs;
}

public long timeoutMs() {
return timeoutMs;
}

@Override
public String toStringBase() {
return super.toStringBase() + ", timeoutMs=" + timeoutMs;
protected AbstractTopicMetadataEvent(final Type type, final Timer timer) {
super(type, timer);
}
}
6 changes: 4 additions & 2 deletions .../main/java/org/apache/kafka/clients/consumer/internals/events/AllTopicsMetadataEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@
*/
package org.apache.kafka.clients.consumer.internals.events;

import org.apache.kafka.common.utils.Timer;

public class AllTopicsMetadataEvent extends AbstractTopicMetadataEvent {

public AllTopicsMetadataEvent(final long timeoutMs) {
super(Type.ALL_TOPICS_METADATA, timeoutMs);
public AllTopicsMetadataEvent(final Timer timer) {
super(Type.ALL_TOPICS_METADATA, timer);
}
}
8 changes: 4 additions & 4 deletions ...in/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,7 @@ private void process(final SyncCommitEvent event) {
}

CommitRequestManager manager = requestManagers.commitRequestManager.get();
long expirationTimeoutMs = getExpirationTimeForTimeout(event.retryTimeoutMs());
long expirationTimeoutMs = getExpirationTimeForTimeout(event.deadlineMs());
CompletableFuture<Void> future = manager.commitSync(event.offsets(), expirationTimeoutMs);
future.whenComplete(complete(event.future()));
}
Expand All @@ -177,7 +177,7 @@ private void process(final FetchCommittedOffsetsEvent event) {
return;
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
long expirationTimeMs = getExpirationTimeForTimeout(event.timeout());
long expirationTimeMs = getExpirationTimeForTimeout(event.deadlineMs());

@cadonna cadonna Mar 4, 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.

Isn't event.deadlineMs() already a point in time instead of a time span? If yes, you do not need to compute the point in time anymore, do you?

@kirktrue kirktrue Mar 4, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Correct. This PR is intentionally limited in its design and scope. My hope was to leave that for this PR and rectify it as part of KAFKA-15974.

@cadonna cadonna Mar 4, 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.

Wait, I think we are misunderstanding each other. Method getExpirationTimeForTimeout() computes a point in time by adding a given timeout, i.e., a time delta, to the the current time. Method event.deadlineMS() returns a point in time computed by adding the remaining time of the timer, i.e., a time delta, to the current time. With this call you are trying to compute a point in time by adding a time point to the current time. This does not look like a limitation but rather like a mistake.

@kirktrue kirktrue Mar 4, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

You're right! Thanks for catching that!

Previously, we used delta values and we're moving to absolute values. The absolute value is now calculated once in the CompletableApplicationEvent and used subsequently.

CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = manager.fetchOffsets(event.partitions(), expirationTimeMs);
future.whenComplete(complete(event.future()));
}
Expand Down Expand Up @@ -250,14 +250,14 @@ private void process(final ValidatePositionsEvent event) {
}

private void process(final TopicMetadataEvent event) {
final long expirationTimeMs = getExpirationTimeForTimeout(event.timeoutMs());
final long expirationTimeMs = getExpirationTimeForTimeout(event.deadlineMs());
final CompletableFuture<Map<String, List<PartitionInfo>>> future =
requestManagers.topicMetadataRequestManager.requestTopicMetadata(event.topic(), expirationTimeMs);
future.whenComplete(complete(event.future()));
}

private void process(final AllTopicsMetadataEvent event) {
final long expirationTimeMs = getExpirationTimeForTimeout(event.timeoutMs());
final long expirationTimeMs = getExpirationTimeForTimeout(event.deadlineMs());
final CompletableFuture<Map<String, List<PartitionInfo>>> future =
requestManagers.topicMetadataRequestManager.requestAllTopicsMetadata(expirationTimeMs);
future.whenComplete(complete(event.future()));
Expand Down
5 changes: 3 additions & 2 deletions ...ts/src/main/java/org/apache/kafka/clients/consumer/internals/events/AsyncCommitEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Timer;

import java.util.Map;

Expand All @@ -26,7 +27,7 @@
*/
public class AsyncCommitEvent extends CommitEvent {

public AsyncCommitEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(Type.COMMIT_ASYNC, offsets);
public AsyncCommitEvent(final Map<TopicPartition, OffsetAndMetadata> offsets, final Timer timer) {
super(Type.COMMIT_ASYNC, timer, offsets);
}
}
5 changes: 3 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Timer;

import java.util.Collections;
import java.util.Map;
Expand All @@ -29,8 +30,8 @@ public abstract class CommitEvent extends CompletableApplicationEvent<Void> {
*/
private final Map<TopicPartition, OffsetAndMetadata> offsets;

protected CommitEvent(final Type type, final Map<TopicPartition, OffsetAndMetadata> offsets) {
super(type);
protected CommitEvent(final Type type, final Timer timer, final Map<TopicPartition, OffsetAndMetadata> offsets) {
Comment thread
kirktrue marked this conversation as resolved.
Outdated
super(type, timer);
this.offsets = Collections.unmodifiableMap(offsets);

for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
Expand Down
14 changes: 12 additions & 2 deletions .../java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@
*/
package org.apache.kafka.clients.consumer.internals.events;

import org.apache.kafka.common.utils.Timer;

import java.util.Objects;
import java.util.concurrent.CompletableFuture;

/**
Expand All @@ -27,19 +30,26 @@
public abstract class CompletableApplicationEvent<T> extends ApplicationEvent implements CompletableEvent<T> {

private final CompletableFuture<T> future;
private final long deadlineMs;

protected CompletableApplicationEvent(final Type type) {
protected CompletableApplicationEvent(final Type type, final Timer timer) {
super(type);
this.future = new CompletableFuture<>();
Objects.requireNonNull(timer);
this.deadlineMs = timer.remainingMs() + timer.currentTimeMs();
}

@Override
public CompletableFuture<T> future() {
return future;
}

public long deadlineMs() {
return deadlineMs;
}

@Override
protected String toStringBase() {
return super.toStringBase() + ", future=" + future;
return super.toStringBase() + ", future=" + future + ", deadlineMs=" + deadlineMs;
}
}
17 changes: 4 additions & 13 deletions ...n/java/org/apache/kafka/clients/consumer/internals/events/FetchCommittedOffsetsEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Timer;

import java.util.Collections;
import java.util.Map;
Expand All @@ -30,27 +31,17 @@ public class FetchCommittedOffsetsEvent extends CompletableApplicationEvent<Map<
*/
private final Set<TopicPartition> partitions;

/**
* Time until which the request will be retried if it fails with a retriable error.
*/
private final long timeoutMs;

public FetchCommittedOffsetsEvent(final Set<TopicPartition> partitions, final long timeoutMs) {
super(Type.FETCH_COMMITTED_OFFSETS);
public FetchCommittedOffsetsEvent(final Set<TopicPartition> partitions, final Timer timer) {
super(Type.FETCH_COMMITTED_OFFSETS, timer);
this.partitions = Collections.unmodifiableSet(partitions);
this.timeoutMs = timeoutMs;
}

public Set<TopicPartition> partitions() {
return partitions;
}

public long timeout() {
return timeoutMs;
}

@Override
public String toStringBase() {
return super.toStringBase() + ", partitions=" + partitions + ", partitions=" + partitions;
return super.toStringBase() + ", partitions=" + partitions;
}
}
6 changes: 4 additions & 2 deletions ...s/src/main/java/org/apache/kafka/clients/consumer/internals/events/LeaveOnCloseEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@
*/
package org.apache.kafka.clients.consumer.internals.events;

import org.apache.kafka.common.utils.Timer;

public class LeaveOnCloseEvent extends CompletableApplicationEvent<Void> {

public LeaveOnCloseEvent() {
super(Type.LEAVE_ON_CLOSE);
public LeaveOnCloseEvent(final Timer timer) {
super(Type.LEAVE_ON_CLOSE, timer);
}
}
5 changes: 3 additions & 2 deletions ...ts/src/main/java/org/apache/kafka/clients/consumer/internals/events/ListOffsetsEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Timer;

import java.util.Collections;
import java.util.HashMap;
Expand All @@ -36,8 +37,8 @@ public class ListOffsetsEvent extends CompletableApplicationEvent<Map<TopicParti
private final Map<TopicPartition, Long> timestampsToSearch;
private final boolean requireTimestamps;

public ListOffsetsEvent(final Map<TopicPartition, Long> timestampToSearch, final boolean requireTimestamps) {
super(Type.LIST_OFFSETS);
public ListOffsetsEvent(final Map<TopicPartition, Long> timestampToSearch, final boolean requireTimestamps, final Timer timer) {
super(Type.LIST_OFFSETS, timer);
this.timestampsToSearch = Collections.unmodifiableMap(timestampToSearch);
this.requireTimestamps = requireTimestamps;
}
Expand Down
6 changes: 4 additions & 2 deletions ...src/main/java/org/apache/kafka/clients/consumer/internals/events/ResetPositionsEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,16 @@

package org.apache.kafka.clients.consumer.internals.events;

import org.apache.kafka.common.utils.Timer;

/**
* Event for resetting offsets for all assigned partitions that require it. This is an
* asynchronous event that generates ListOffsets requests, and completes by updating in-memory
* positions when responses are received.
*/
public class ResetPositionsEvent extends CompletableApplicationEvent<Void> {

public ResetPositionsEvent() {
super(Type.RESET_POSITIONS);
public ResetPositionsEvent(final Timer timer) {
super(Type.RESET_POSITIONS, timer);
}
}
Loading