-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
7c13d26
commit d540530
Showing
6 changed files
with
221 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
server/src/main/java/io/littlehorse/server/streams/taskqueue/TaskQueue.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package io.littlehorse.server.streams.taskqueue; | ||
|
|
||
| import io.littlehorse.common.model.ScheduledTaskModel; | ||
| import io.littlehorse.common.model.getable.objectId.TenantIdModel; | ||
| import io.littlehorse.server.streams.topology.core.RequestExecutionContext; | ||
| import org.apache.kafka.streams.processor.TaskId; | ||
|
|
||
| public interface TaskQueue { | ||
|
|
||
| /** | ||
| * Called when a gRPC client (and its StreamObserver) disconnect, whether due to | ||
| * a clean | ||
| * shutdown (onCompleted()) or connection error (onError()). | ||
| * | ||
| * @param disconnectedObserver is the TaskQueueStreamObserver for the client whose | ||
| * connection is now gone. | ||
| */ | ||
| void onRequestDisconnected(PollTaskRequestObserver disconnectedObserver); | ||
|
|
||
| /** | ||
| * Called in two places: 1. In the CommandProcessorDaoImpl::scheduleTask() 2. In | ||
| * the | ||
| * CommandProcessor::init(). | ||
| * | ||
| * <p> | ||
| * Item 1) is quite self-explanatory. | ||
| * | ||
| * <p> | ||
| * For Item 2), remember that the Task Queue Manager system is only in-memory. | ||
| * Upon a restart | ||
| * or rebalance, we need to rebuild that state. During the init() call, we | ||
| * iterate through all | ||
| * currently scheduled but not started tasks in the state store. | ||
| * | ||
| * @param scheduledTask is the ::getObjectId() for the TaskScheduleRequest | ||
| * that was just | ||
| * scheduled. | ||
| * @return True if the task was successfully scheduled, or False if the queue is full. | ||
| */ | ||
| boolean onTaskScheduled(TaskId streamsTaskId, ScheduledTaskModel scheduledTask); | ||
|
|
||
| /** | ||
| * Called when a grpc client sends a new PollTaskPb. | ||
| * | ||
| * @param requestObserver is the grpc StreamObserver representing the channel | ||
| * that talks to the | ||
| * client who made the PollTaskRequest. | ||
| */ | ||
| void onPollRequest(PollTaskRequestObserver requestObserver, RequestExecutionContext requestContext); | ||
|
|
||
| int size(); | ||
|
|
||
| long rehydratedCount(); | ||
|
|
||
| void drainPartition(TaskId partitionToDrain); | ||
|
|
||
| TenantIdModel tenantId(); | ||
|
|
||
| String taskDefName(); | ||
| } |
132 changes: 132 additions & 0 deletions
132
server/src/main/java/io/littlehorse/server/streams/taskqueue/TaskQueueImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| package io.littlehorse.server.streams.taskqueue; | ||
|
|
||
| import io.littlehorse.common.model.ScheduledTaskModel; | ||
| import io.littlehorse.common.model.getable.objectId.TenantIdModel; | ||
| import io.littlehorse.server.streams.topology.core.RequestExecutionContext; | ||
| import java.util.LinkedList; | ||
| import java.util.Queue; | ||
| import java.util.concurrent.LinkedBlockingQueue; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.concurrent.locks.Lock; | ||
| import java.util.concurrent.locks.ReentrantLock; | ||
| import java.util.function.Supplier; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.kafka.streams.processor.TaskId; | ||
|
|
||
| @Slf4j | ||
| public class TaskQueueImpl implements TaskQueue { | ||
|
|
||
| private final String taskDefName; | ||
| private final TaskQueueManager parent; | ||
| private final int capacity; | ||
| private final TenantIdModel tenantId; | ||
| private final Lock lock = new ReentrantLock(); | ||
| private final Queue<PollTaskRequestObserver> hungryClients = new LinkedList<>(); | ||
| private final String instanceName; | ||
| private final LinkedBlockingQueue<QueueItem> pendingTasks; | ||
| private final AtomicBoolean needsRehydration = new AtomicBoolean(false); | ||
|
|
||
| public TaskQueueImpl(String taskDefName, TaskQueueManager parent, int capacity, TenantIdModel tenantId) { | ||
| this.taskDefName = taskDefName; | ||
| this.parent = parent; | ||
| this.capacity = capacity; | ||
| this.tenantId = tenantId; | ||
| this.instanceName = parent.getBackend().getInstanceName(); | ||
| this.pendingTasks = new LinkedBlockingQueue<>(capacity); | ||
| } | ||
|
|
||
| @Override | ||
| public void onRequestDisconnected(PollTaskRequestObserver disconnectedObserver) { | ||
| synchronizedBlock(() -> { | ||
| hungryClients.removeIf(thing -> { | ||
| log.trace( | ||
| "Instance {}: Removing task queue observer for taskdef {} with" + " client id {}: {}", | ||
| instanceName, | ||
| taskDefName, | ||
| disconnectedObserver.getClientId(), | ||
| disconnectedObserver); | ||
| return thing.equals(disconnectedObserver); | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean onTaskScheduled(TaskId streamTaskId, ScheduledTaskModel scheduledTask) { | ||
| boolean outOfCapacity = synchronizedBlock(() -> { | ||
| if (needsRehydration.get()) { | ||
| return true; | ||
| } | ||
| boolean added = pendingTasks.offer(new QueueItem(streamTaskId, scheduledTask)); | ||
| if (!added) { | ||
| needsRehydration.set(true); | ||
| } | ||
| return !added; | ||
| }); | ||
| if (!outOfCapacity && !hungryClients.isEmpty()) { | ||
| synchronizedBlock(() -> { | ||
| PollTaskRequestObserver hungryClient = hungryClients.poll(); | ||
| if (hungryClient != null) { | ||
| parent.itsAMatch(scheduledTask, hungryClient); | ||
| } | ||
| }); | ||
| } | ||
| return outOfCapacity; | ||
| } | ||
|
|
||
| @Override | ||
| public void onPollRequest(PollTaskRequestObserver requestObserver, RequestExecutionContext requestContext) { | ||
| synchronizedBlock(() -> { | ||
| QueueItem nextItem = pendingTasks.poll(); | ||
| if (nextItem != null) { | ||
| parent.itsAMatch(nextItem.scheduledTask(), requestObserver); | ||
| } else { | ||
| hungryClients.add(requestObserver); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
| public int size() { | ||
| return pendingTasks.size(); | ||
| } | ||
|
|
||
| @Override | ||
| public long rehydratedCount() { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public void drainPartition(TaskId partitionToDrain) { | ||
| pendingTasks.removeIf(queueItem -> queueItem.streamsTaskId().equals(partitionToDrain)); | ||
| } | ||
|
|
||
| @Override | ||
| public TenantIdModel tenantId() { | ||
| return tenantId; | ||
| } | ||
|
|
||
| @Override | ||
| public String taskDefName() { | ||
| return taskDefName; | ||
| } | ||
|
|
||
| private record QueueItem(TaskId streamsTaskId, ScheduledTaskModel scheduledTask) {} | ||
|
|
||
| private void synchronizedBlock(Runnable runnable) { | ||
| try { | ||
| lock.lock(); | ||
| runnable.run(); | ||
| } finally { | ||
| lock.unlock(); | ||
| } | ||
| } | ||
|
|
||
| private boolean synchronizedBlock(Supplier<Boolean> booleanSupplier) { | ||
| try { | ||
| lock.lock(); | ||
| return booleanSupplier.get(); | ||
| } finally { | ||
| lock.unlock(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters