You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/10/20 05:56:06 UTC
(pinot) 01/01: Add fenced native minion scheduling and offline execution
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-minion-execution
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 4d1ceb46beea1801cbe20ab424b8145d3b581d27
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Mon Oct 19 22:55:52 2026 -0700
Add fenced native minion scheduling and offline execution
---
.../common/coordination/MetadataRpcClient.java | 28 ++
.../common/coordination/MetadataRpcConnection.java | 12 +
.../coordination/NativeTaskOutputIdentity.java | 78 ++++
pinot-common/src/main/proto/metadata.proto | 103 ++++-
.../coordination/NativeControllerTaskResource.java | 219 ++++++++++
.../coordination/NativeControllerTasks.java | 462 ++++++++++++++++++++
.../coordination/RatisControllerProvider.java | 2 +-
.../coordination/RatisControllerRuntime.java | 19 +-
.../coordination/NativeControllerTasksTest.java | 318 ++++++++++++++
pinot-coordination-ratis/CONTROLLER-API-PARITY.md | 5 +-
pinot-coordination-ratis/MINION-TASK-PLAN.md | 101 ++++-
pinot-coordination-ratis/README.md | 11 +-
.../coordination/ratis/MetadataStateMachine.java | 7 +
.../coordination/ratis/RatisMetadataAccess.java | 13 +-
.../ratis/metadata/MembershipProtocol.java | 6 +-
.../ratis/metadata/MembershipState.java | 7 +-
.../ratis/metadata/MetadataCatalog.java | 96 ++++-
.../ratis/metadata/MetadataProtocol.java | 12 +-
.../ratis/metadata/OfflineSegmentCatalog.java | 2 +-
.../coordination/ratis/metadata/TaskCatalog.java | 446 +++++++++++++++++++-
.../coordination/ratis/metadata/TaskProtocol.java | 163 +++++++-
.../ratis/transport/MetadataGrpcServer.java | 14 +
.../ratis/transport/MetadataRpcCodec.java | 15 +
.../ratis/transport/ParticipantAccessPolicy.java | 7 +-
.../coordination/ratis/transport/TaskAccess.java | 32 ++
.../ratis/transport/TaskGrpcService.java | 165 ++++++++
.../coordination/ratis/transport/TaskRpcCodec.java | 186 ++++++++-
.../ratis/metadata/MembershipStateTest.java | 2 +-
.../ratis/metadata/MetadataCatalogTest.java | 2 +-
.../ratis/metadata/MetadataChangeFeedTest.java | 4 +-
.../ratis/metadata/TaskCatalogTest.java | 228 +++++++++-
.../ratis/transport/MetadataGrpcTransportTest.java | 157 ++++++-
.../ratis/transport/TaskRpcCodecTest.java | 34 +-
.../org/apache/pinot/minion/BaseMinionStarter.java | 23 +
.../minion/coordination/NativeMinionRuntime.java | 465 +++++++++++++++++++++
.../coordination/NativeSegmentGenerationTask.java | 256 ++++++++++++
.../minion/coordination/NativeTaskOutbox.java | 127 ++++++
.../coordination/NativeMinionRuntimeTest.java | 126 ++++++
.../NativeSegmentGenerationTaskTest.java | 139 ++++++
.../minion/coordination/NativeTaskOutboxTest.java | 100 +++++
.../pinot/tools/service/PinotServiceManager.java | 6 +-
.../pinot/tools/service/RatisStartersTest.java | 137 +++++-
42 files changed, 4245 insertions(+), 90 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java
index ae9bd91b44a..c04f5aa2010 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcClient.java
@@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
import org.apache.pinot.coordination.ratis.rpc.MetadataServiceGrpc;
+import org.apache.pinot.coordination.ratis.rpc.MinionTaskServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.ParticipantMembershipServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.ServerInstallationServiceGrpc;
@@ -43,10 +44,12 @@ public class MetadataRpcClient implements AutoCloseable {
public static final int MAX_RESPONSE_BYTES = 2 * MAX_COMMAND_BYTES;
public static final int MAX_MEMBERSHIP_BYTES = 64 * 1024;
public static final int MAX_INSTALLATION_REQUEST_BYTES = 64 * 1024;
+ public static final int MAX_TASK_REQUEST_BYTES = 64 * 1024;
private final ManagedChannel _channel;
private final MetadataServiceGrpc.MetadataServiceBlockingStub _stub;
private final ParticipantMembershipServiceGrpc.ParticipantMembershipServiceBlockingStub _membership;
private final ServerInstallationServiceGrpc.ServerInstallationServiceBlockingStub _installation;
+ private final MinionTaskServiceGrpc.MinionTaskServiceBlockingStub _task;
private final long _timeoutMillis;
public MetadataRpcClient(String host, int port, Path certificateChain, Path privateKey, Path trustCertificates,
@@ -63,6 +66,31 @@ public class MetadataRpcClient implements AutoCloseable {
_stub = MetadataServiceGrpc.newBlockingStub(_channel);
_membership = ParticipantMembershipServiceGrpc.newBlockingStub(_channel);
_installation = ServerInstallationServiceGrpc.newBlockingStub(_channel);
+ _task = MinionTaskServiceGrpc.newBlockingStub(_channel);
+ }
+
+ public MetadataRpc.CommandResult taskExecute(MetadataRpc.TaskCommandRequest request, long timeoutMillis) {
+ return taskCall(request, timeoutMillis, stub -> stub.execute(request));
+ }
+
+ public MetadataRpc.TaskWorkResult taskWork(MetadataRpc.TaskWorkRequest request, long timeoutMillis) {
+ return taskCall(request, timeoutMillis, stub -> stub.work(request));
+ }
+
+ public MetadataRpc.CommandResult taskOutcome(MetadataRpc.TaskOutcomeRequest request, long timeoutMillis) {
+ return taskCall(request, timeoutMillis, stub -> stub.outcome(request));
+ }
+
+ private <T> T taskCall(Message request, long timeoutMillis,
+ Function<MinionTaskServiceGrpc.MinionTaskServiceBlockingStub, T> operation) {
+ if (request == null || request.getSerializedSize() > MAX_TASK_REQUEST_BYTES) {
+ throw Status.RESOURCE_EXHAUSTED.withDescription("Task request exceeds transport byte limit")
+ .asRuntimeException();
+ }
+ if (timeoutMillis <= 0) {
+ throw Status.DEADLINE_EXCEEDED.withDescription("Task request deadline expired").asRuntimeException();
+ }
+ return operation.apply(_task.withDeadlineAfter(Math.min(_timeoutMillis, timeoutMillis), TimeUnit.MILLISECONDS));
}
public MetadataRpc.CommandResult installationExecute(MetadataRpc.InstallationCommandRequest request,
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
index 52527d847b1..69aeb714ab9 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/MetadataRpcConnection.java
@@ -83,6 +83,18 @@ public class MetadataRpcConnection implements AutoCloseable {
return call(timeoutMillis, (client, timeout) -> client.membershipExecute(request, timeout));
}
+ public MetadataRpc.CommandResult taskExecute(MetadataRpc.TaskCommandRequest request, long timeoutMillis) {
+ return call(timeoutMillis, (client, timeout) -> client.taskExecute(request, timeout));
+ }
+
+ public MetadataRpc.TaskWorkResult taskWork(MetadataRpc.TaskWorkRequest request, long timeoutMillis) {
+ return call(timeoutMillis, (client, timeout) -> client.taskWork(request, timeout));
+ }
+
+ public MetadataRpc.CommandResult taskOutcome(MetadataRpc.TaskOutcomeRequest request, long timeoutMillis) {
+ return call(timeoutMillis, (client, timeout) -> client.taskOutcome(request, timeout));
+ }
+
public MetadataRpc.InstallationSnapshot snapshot(MetadataRpc.InstallationSnapshotRequest request,
long timeoutMillis) {
return call(timeoutMillis, (client, timeout) -> client.installationSnapshot(request, timeout));
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/coordination/NativeTaskOutputIdentity.java b/pinot-common/src/main/java/org/apache/pinot/common/coordination/NativeTaskOutputIdentity.java
new file mode 100644
index 00000000000..66398e10f30
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/coordination/NativeTaskOutputIdentity.java
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.coordination;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+
+
+/// Deterministic identities shared by native task staging and consensus publication validation.
+/// These names identify immutable attempt output; constructing one does not authorize its publication.
+public final class NativeTaskOutputIdentity {
+ public static final String TASK_TYPE = "NativeSegmentGenerationTask";
+
+ private NativeTaskOutputIdentity() {
+ }
+
+ public static String outputSegmentName(UUID taskId, long attempt, int ordinal) {
+ validate(taskId, attempt, ordinal);
+ return "task_" + taskId + "_" + attempt + "_" + ordinal;
+ }
+
+ public static UUID outputObjectId(UUID historyId, UUID taskId, long attempt, int ordinal) {
+ validate(taskId, attempt, ordinal);
+ if (historyId == null) {
+ throw new IllegalArgumentException("Output history is required");
+ }
+ return UUID.nameUUIDFromBytes((historyId + "\0TASK_OUTPUT\0" + taskId + "\0" + attempt + "\0" + ordinal)
+ .getBytes(StandardCharsets.UTF_8));
+ }
+
+ public static String outputUri(String rootUri, UUID historyId, UUID taskId, long attempt, UUID incarnation,
+ int ordinal, String sha256) {
+ validate(taskId, attempt, ordinal);
+ if (historyId == null || incarnation == null || sha256 == null || !sha256.matches("[a-f0-9]{64}")) {
+ throw new IllegalArgumentException("Invalid output history, incarnation or SHA-256");
+ }
+ return outputRoot(rootUri).resolve(historyId + "/" + taskId + "/" + attempt + "/" + incarnation + "/"
+ + ordinal + "-" + sha256 + ".tar.gz").toASCIIString();
+ }
+
+ public static URI outputRoot(String rootUri) {
+ try {
+ URI root = new URI(rootUri);
+ if (!"file".equals(root.getScheme()) || root.getAuthority() != null || root.getQuery() != null
+ || root.getFragment() != null || root.getPath() == null || !root.getPath().startsWith("/")
+ || root.getRawPath().contains("%") || root.getPath().contains("\\") || !root.normalize().equals(root)) {
+ throw new IllegalArgumentException("Output policy requires an absolute canonical file root");
+ }
+ return new URI("file", "", root.getPath().endsWith("/") ? root.getPath() : root.getPath() + "/", null);
+ } catch (URISyntaxException | NullPointerException e) {
+ throw new IllegalArgumentException("Invalid output root URI", e);
+ }
+ }
+
+ private static void validate(UUID taskId, long attempt, int ordinal) {
+ if (taskId == null || attempt <= 0 || attempt > 16 || ordinal < 0 || ordinal >= 16) {
+ throw new IllegalArgumentException("Invalid task output identity");
+ }
+ }
+}
diff --git a/pinot-common/src/main/proto/metadata.proto b/pinot-common/src/main/proto/metadata.proto
index b1f7b4addb7..cc7fdf1e78a 100644
--- a/pinot-common/src/main/proto/metadata.proto
+++ b/pinot-common/src/main/proto/metadata.proto
@@ -49,6 +49,14 @@ service ServerInstallationService {
rpc Snapshot(InstallationSnapshotRequest) returns (InstallationSnapshot);
}
+// A separate bounded lane for a MINION's own tasks. Identity and capability come exclusively from current TLS
+// policy. These requests cannot access another minion's input metadata or perform generic catalog mutations.
+service MinionTaskService {
+ rpc Execute(TaskCommandRequest) returns (CommandResult);
+ rpc Work(TaskWorkRequest) returns (TaskWorkResult);
+ rpc Outcome(TaskOutcomeRequest) returns (CommandResult);
+}
+
enum RecordKind {
RECORD_KIND_UNSPECIFIED = 0;
CLUSTER_CONFIG = 1;
@@ -65,6 +73,8 @@ enum RecordKind {
TASK_QUEUE = 12;
TASK = 13;
TASK_ATTEMPT = 14;
+ TASK_SCHEDULE = 15;
+ TASK_PUBLICATION = 16;
}
enum ResultStatus {
@@ -106,6 +116,8 @@ message CatalogValue {
TaskQueueState task_queue = 12;
TaskExecutionState task = 13;
TaskAttemptState task_attempt = 14;
+ TaskScheduleState task_schedule = 15;
+ TaskPublicationState task_publication = 16;
}
}
@@ -218,6 +230,7 @@ enum ServingCapability {
SERVING_CAPABILITY_UNSPECIFIED = 0;
BROKER_QUERY = 1;
SERVER_QUERY = 2;
+ MINION_TASK = 3;
}
message ParticipantToken {
@@ -568,14 +581,16 @@ message InstallationWorkResult {
InstallationWork work = 4;
}
-// Read/feed projections only. No task command or execution service is exposed by this protocol.
+// Typed task state and execution projection. Worker requests omit participant and durable client identities.
message TaskQueueKey { string table_name = 1; string table_object_id = 2; string task_type = 3; }
message TaskInput {
TaskQueueKey queue = 1;
string minion_id = 2;
map<string, string> configs = 3;
int32 max_attempts = 4;
+ TaskOutputPolicy output_policy = 5;
}
+message TaskOutputPolicy { string root_uri = 1; int32 max_outputs = 2; }
message TaskQueueState {
string history_id = 1;
TaskQueueKey queue = 2;
@@ -591,6 +606,7 @@ message TaskExecutionState {
RETRYABLE_FAILED = 4;
FATAL_FAILED = 5;
CANCELLED = 6;
+ SUCCEEDED = 7;
}
string history_id = 1;
string task_id = 2;
@@ -609,6 +625,7 @@ message TaskAttemptState {
RETRYABLE_FAILED = 3;
FATAL_FAILED = 4;
FENCED = 5;
+ SUCCEEDED = 6;
}
string history_id = 1;
string task_id = 2;
@@ -619,3 +636,87 @@ message TaskAttemptState {
State state = 7;
TaskProgress progress = 8;
}
+message TaskScheduleState {
+ string history_id = 1;
+ string schedule_id = 2;
+ TaskInput input = 3;
+ repeated ObjectVersion dependencies = 4;
+ OwnerToken controller = 5;
+ int64 first_round = 6;
+ int64 interval_rounds = 7;
+ int64 max_occurrences = 8;
+ int64 fired_occurrences = 9;
+ int64 policy_version = 10;
+}
+message TaskPublicationState {
+ string history_id = 1;
+ string task_id = 2;
+ int64 attempt = 3;
+ ParticipantToken minion = 4;
+ repeated OfflineSegmentContent outputs = 5;
+ int64 committed_index = 6;
+}
+message TaskClaim { string task_id = 1; int64 expected_version = 2; }
+message TaskCheckpoint {
+ string task_id = 1;
+ int64 expected_version = 2;
+ int64 attempt = 3;
+ TaskProgress progress = 4;
+}
+message TaskComplete {
+ string task_id = 1;
+ int64 expected_version = 2;
+ int64 attempt = 3;
+ TaskAttemptState.State result = 4;
+ TaskProgress progress = 5;
+}
+// Publication is constrained by the captured task output policy and exact active attempt. It is never a generic
+// metadata mutation; the consensus transition commits content, provenance and terminal task outcome atomically.
+message TaskPublishOfflineOutputs {
+ string task_id = 1;
+ int64 expected_version = 2;
+ int64 attempt = 3;
+ repeated OfflineSegmentContent outputs = 4;
+ TaskProgress progress = 5;
+}
+message SequencedTaskOperation {
+ int64 client_epoch = 1;
+ string stream_id = 2;
+ int64 sequence = 3;
+ oneof operation {
+ TaskClaim claim = 4;
+ TaskCheckpoint checkpoint = 5;
+ TaskComplete complete = 6;
+ TaskPublishOfflineOutputs publish = 7;
+ }
+}
+message TaskCommandRequest {
+ string history_id = 1;
+ ParticipantSession session = 2;
+ oneof command {
+ RegisterClient register = 3;
+ Acknowledge acknowledge = 4;
+ SequencedTaskOperation operation = 5;
+ }
+}
+// The complete own-session projection includes dependencies for precisely the tasks assigned to this MINION.
+// Oversize returns RESPONSE_TOO_LARGE without partial records. This image can advance its applied revision.
+message TaskWorkRequest {
+ string history_id = 1;
+ ParticipantSession session = 2;
+ int32 max_records = 3;
+ int32 max_bytes = 4;
+}
+message TaskWorkResult {
+ ResultStatus status = 1;
+ Revision revision = 2;
+ string detail = 3;
+ ParticipantState participant = 4;
+ repeated CatalogRecord records = 5;
+ MembershipControl control = 6;
+}
+message TaskOutcomeRequest {
+ string history_id = 1;
+ ParticipantSession session = 2;
+ OutcomeQuery outcome = 3;
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTaskResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTaskResource.java
new file mode 100644
index 00000000000..27b1c2e022e
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTaskResource.java
@@ -0,0 +1,219 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.coordination;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.StreamReadFeature;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.common.utils.DatabaseUtils;
+import org.apache.pinot.controller.api.access.AccessControlFactory;
+import org.apache.pinot.controller.api.access.AccessControlUtils;
+import org.apache.pinot.controller.api.access.AccessType;
+import org.apache.pinot.controller.api.access.Authenticate;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol;
+import org.apache.pinot.core.auth.Actions;
+import org.apache.pinot.core.auth.Authorize;
+import org.apache.pinot.core.auth.TargetType;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+
+
+/// Table-scoped experimental scheduling API. It exposes durable native states directly; SUCCEEDED means output
+/// publication committed. Existing Helix task API resources and plugin activation are unchanged.
+@Path("/tasks/native/{tableName}")
+@Produces(MediaType.APPLICATION_JSON)
+public final class NativeControllerTaskResource {
+ private static final ObjectMapper REQUEST_JSON = JsonMapper.builder(JsonFactory.builder()
+ .enable(StreamReadFeature.STRICT_DUPLICATE_DETECTION).build())
+ .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+ .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+ .enable(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES)
+ .enable(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES).build();
+ private final NativeControllerTasks _tasks;
+ private final AccessControlFactory _access;
+
+ public NativeControllerTaskResource(NativeControllerTasks tasks, AccessControlFactory access) {
+ _tasks = tasks;
+ _access = access;
+ }
+
+ @PUT
+ @Path("/schedules/{scheduleId}")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Authenticate(AccessType.UPDATE)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.CREATE_TASK, tableScopeParam = "tableName")
+ public Response schedule(@PathParam("tableName") String tableName, @PathParam("scheduleId") String scheduleId,
+ String body, @Context HttpHeaders headers) {
+ String table = authorize(tableName, headers, AccessType.UPDATE, Actions.Cluster.CREATE_TASK);
+ try {
+ if (body == null || body.length() > 8192) {
+ throw new IllegalArgumentException("Schedule request exceeds 8192 characters");
+ }
+ NativeControllerTasks.ScheduleRequest request;
+ try {
+ request = REQUEST_JSON.readValue(body, NativeControllerTasks.ScheduleRequest.class);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Invalid schedule request", e);
+ }
+ return result(_tasks.schedule(table, UUID.fromString(scheduleId), request));
+ } catch (IllegalArgumentException e) {
+ throw error(400, e.getMessage());
+ } catch (IOException e) {
+ throw error(503, "Native task scheduling is unavailable");
+ }
+ }
+
+ @GET
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_TASK, tableScopeParam = "tableName")
+ public Response tasks(@PathParam("tableName") String tableName, @Context HttpHeaders headers) {
+ String table = authorize(tableName, headers, AccessType.READ, Actions.Cluster.GET_TASK);
+ try {
+ var image = _tasks.snapshot();
+ List<CatalogRecord> schedules = new ArrayList<>();
+ List<CatalogRecord> tasks = new ArrayList<>();
+ List<CatalogRecord> queues = new ArrayList<>();
+ List<CatalogRecord> publications = new ArrayList<>();
+ long round = -1;
+ boolean exists = false;
+ for (CatalogRecord record : image.records()) {
+ switch (record.kind()) {
+ case TABLE_CONFIG -> exists |= record.name().equals(table);
+ case MEMBERSHIP_CONTROL -> round = MembershipProtocol.decodeControl(record.value()).round();
+ case TASK_SCHEDULE -> {
+ if (TaskProtocol.decodeSchedule(record.value()).input().queue().tableName().equals(table)) {
+ schedules.add(record);
+ }
+ }
+ case TASK_QUEUE -> {
+ if (TaskProtocol.decodeQueue(record.value()).queue().tableName().equals(table)) {
+ queues.add(record);
+ }
+ }
+ case TASK -> {
+ if (TaskProtocol.decodeTask(record.value()).input().queue().tableName().equals(table)) {
+ tasks.add(record);
+ }
+ }
+ default -> { }
+ }
+ }
+ for (CatalogRecord record : image.records()) {
+ if (record.kind() == RecordKind.TASK_PUBLICATION) {
+ UUID taskId = TaskProtocol.decodePublication(record.value()).taskId();
+ if (tasks.stream().anyMatch(task -> task.name().equals(taskId.toString()))) {
+ publications.add(record);
+ }
+ }
+ }
+ if (!exists) {
+ throw error(404, "Table not found");
+ }
+ return Response.ok(JsonUtils.objectToString(Map.of("round", round, "schedules", schedules, "tasks", tasks,
+ "queues", queues, "publications", publications, "schedulerFailure",
+ _tasks.failure() == null ? "" : "Native scheduler is unavailable")))
+ .header(HttpHeaders.CACHE_CONTROL, "no-store")
+ .header(NativeControllerResource.REVISION_HEADER,
+ image.cursor().historyId() + ":" + image.cursor().index()).build();
+ } catch (IOException e) {
+ throw error(503, "Native task state is unavailable");
+ }
+ }
+
+ @POST
+ @Path("/{taskId}/cancel")
+ @Authenticate(AccessType.UPDATE)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.STOP_TASK, tableScopeParam = "tableName")
+ public Response cancel(@PathParam("tableName") String tableName, @PathParam("taskId") String taskId,
+ @QueryParam("expectedVersion") long expectedVersion, @Context HttpHeaders headers) {
+ String table = authorize(tableName, headers, AccessType.UPDATE, Actions.Cluster.STOP_TASK);
+ try {
+ return result(_tasks.cancel(table, UUID.fromString(taskId), expectedVersion));
+ } catch (IllegalArgumentException e) {
+ throw error(400, e.getMessage());
+ } catch (IOException e) {
+ throw error(503, "Native task cancellation is unavailable");
+ }
+ }
+
+ private String authorize(String name, HttpHeaders headers, AccessType type, String action) {
+ String table;
+ try {
+ table = DatabaseUtils.translateTableName(name, headers);
+ if (TableNameBuilder.getTableTypeFromTableName(table) == TableType.REALTIME) {
+ throw error(501, "Native tasks currently support OFFLINE tables");
+ }
+ String raw = TableNameBuilder.extractRawTableName(table);
+ if (raw == null || raw.isBlank() || TableNameBuilder.getTableTypeFromTableName(raw) != null) {
+ throw new IllegalArgumentException("Ambiguous table name");
+ }
+ table = TableNameBuilder.OFFLINE.tableNameWithType(table);
+ } catch (IllegalArgumentException e) {
+ throw error(400, "Invalid table name");
+ }
+ var access = _access.create();
+ AccessControlUtils.validatePermission(table, type, headers, "/tasks/native/" + table, access);
+ if (!access.hasAccess(headers, TargetType.CLUSTER, null, action)) {
+ throw error(403, "Task permission denied");
+ }
+ return table;
+ }
+
+ private static Response result(Result result) throws IOException {
+ int status = switch (result.status()) {
+ case OK -> 200;
+ case INVALID_REQUEST -> 400;
+ case NOT_FOUND -> 404;
+ case CONFLICT, FENCED, HISTORY_MISMATCH -> 409;
+ default -> 503;
+ };
+ return Response.status(status).entity(JsonUtils.objectToString(result))
+ .header(HttpHeaders.CACHE_CONTROL, "no-store").build();
+ }
+
+ private static WebApplicationException error(int status, String message) {
+ return new WebApplicationException(Response.status(status).type(MediaType.APPLICATION_JSON_TYPE)
+ .header(HttpHeaders.CACHE_CONTROL, "no-store").entity(Map.of("error", message)).build());
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java
new file mode 100644
index 00000000000..b836531a55c
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java
@@ -0,0 +1,462 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.coordination;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongFunction;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.coordination.NativeTaskOutputIdentity;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ControlState;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Response;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogImage;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Command;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipCommand;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipQuery;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol;
+import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.Assignment;
+import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.Layout;
+import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.OwnerToken;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.ControllerOperation;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Input;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.ScheduleState;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskState;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskStatus;
+import org.apache.pinot.coordination.ratis.transport.MembershipAccess;
+import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// Bounded native task scheduling using committed membership rounds, never local wall-clock authority. A controller
+/// claims only unowned or expired table shards. Commands and heartbeats retain their identity across unknown replies.
+/// Methods serialize the controller request stream; immutable schedule identities survive process restart.
+public final class NativeControllerTasks implements AutoCloseable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(NativeControllerTasks.class);
+ private static final FeedScope SCOPE = new FeedScope(EnumSet.of(RecordKind.TABLE_CONFIG, RecordKind.SCHEMA,
+ RecordKind.CLUSTER_CONFIG, RecordKind.OFFLINE_SEGMENT_CONTENT,
+ RecordKind.MEMBERSHIP_CONTROL, RecordKind.PARTICIPANT, RecordKind.OWNER_LAYOUT, RecordKind.OWNER_ASSIGNMENT,
+ RecordKind.TASK_SCHEDULE, RecordKind.TASK_QUEUE, RecordKind.TASK, RecordKind.TASK_ATTEMPT,
+ RecordKind.TASK_PUBLICATION), "native-controller-tasks");
+ private final MetadataAccess _access;
+ private final MembershipAccess _membership;
+ private final UUID _history;
+ private final String _participantId;
+ private final String _clientId;
+ private final UUID _clientIncarnation;
+ private final long _timeoutMillis;
+ private final long _pollMillis;
+ @Nullable
+ private final String _outputRoot;
+ @Nullable
+ private ScheduledExecutorService _executor;
+ @Nullable
+ private ParticipantToken _participant;
+ @Nullable
+ private MembershipProtocol.Register _registration;
+ @Nullable
+ private MembershipProtocol.Renewal _renewal;
+ @Nullable
+ private Command _pending;
+ private boolean _closed;
+ private int _scanCursor;
+ @Nullable
+ private volatile String _failure;
+
+ /// The public schedule ID is also its immutable configuration generation. Round zero is useful for an immediate
+ /// one-shot job; recurring callers should select a first round from the current task image.
+ public record ScheduleRequest(String minionId, String sourceUri, String sourceSha256, long firstRound,
+ long intervalRounds, int occurrences, int maxAttempts) {
+ }
+
+ public NativeControllerTasks(MetadataAccess access, MembershipAccess membership, UUID history, String participantId,
+ ControllerConf config) throws IOException {
+ _access = access;
+ _membership = membership;
+ _history = history;
+ _participantId = participantId;
+ _clientId = "controller-tasks/" + NativeMetadataConfig.required(config, "id");
+ _clientIncarnation = UUID.nameUUIDFromBytes((_history + "/" + _clientId).getBytes(StandardCharsets.UTF_8));
+ _timeoutMillis = config.getProperty(NativeMetadataConfig.PREFIX + "controller.tasks.timeoutMs", 5000L);
+ _pollMillis = config.getProperty(NativeMetadataConfig.PREFIX + "controller.tasks.pollMillis", 1000L);
+ checkArgument(_timeoutMillis > 0 && _timeoutMillis <= 30_000 && _pollMillis > 0 && _pollMillis <= 30_000,
+ "Native task RPC and polling intervals must be between 1 and 30000 milliseconds");
+ String root = config.getProperty(NativeMetadataConfig.PREFIX + "controller.tasks.outputRoot");
+ if (root == null) {
+ _outputRoot = null;
+ } else {
+ Path directory = Path.of(root).toRealPath();
+ checkArgument(Files.isDirectory(directory), "Native task output root must be an existing shared directory");
+ _outputRoot = directory.toUri().toString();
+ }
+ }
+
+ public synchronized void start() {
+ checkArgument(!_closed && _executor == null, "Native task scheduler is already started or closed");
+ _executor = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable, "native-controller-tasks");
+ thread.setDaemon(true);
+ return thread;
+ });
+ _executor.scheduleWithFixedDelay(this::tickSafely, 0, _pollMillis, TimeUnit.MILLISECONDS);
+ }
+
+ private void tickSafely() {
+ try {
+ tick();
+ _failure = null;
+ } catch (Exception e) {
+ _failure = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
+ }
+ }
+
+ /// One cycle admits at most one due occurrence or retry. A quorum failure leaves pending command identity intact.
+ synchronized void tick() throws IOException {
+ ensureOpen();
+ resolvePending();
+ heartbeat();
+ CatalogImage image = image();
+ Map<UUID, CatalogRecord> records = image.records().stream()
+ .collect(Collectors.toMap(CatalogRecord::objectId, record -> record));
+ List<CatalogRecord> candidates = image.records().stream()
+ .filter(record -> record.kind() == RecordKind.TASK_SCHEDULE || record.kind() == RecordKind.TASK)
+ .sorted(Comparator.comparing(record -> record.objectId().toString())).toList();
+ int start = candidates.isEmpty() ? 0 : _scanCursor % candidates.size();
+ for (int offset = 0; offset < candidates.size(); offset++) {
+ int position = (start + offset) % candidates.size();
+ _scanCursor = (position + 1) % candidates.size();
+ CatalogRecord record = candidates.get(position);
+ if (record.kind() == RecordKind.TASK_SCHEDULE) {
+ ScheduleState schedule = TaskProtocol.decodeSchedule(record.value());
+ if (schedule.firedOccurrences() >= schedule.maxOccurrences()) {
+ continue;
+ }
+ long due = Math.addExact(schedule.firstRound(),
+ Math.multiplyExact(schedule.intervalRounds(), schedule.firedOccurrences()));
+ ControlState control = control(image);
+ if (control.round() < due || control.policy().version() != schedule.policyVersion()) {
+ continue;
+ }
+ OwnerToken owner = owner(image, schedule.input().queue().tableName());
+ if (owner != null) {
+ task(new TaskProtocol.FireSchedule(owner, schedule.scheduleId(), record.version()));
+ return;
+ }
+ }
+ if (record.kind() == RecordKind.TASK) {
+ TaskState task = TaskProtocol.decodeTask(record.value());
+ if ((task.status() == TaskStatus.READY || task.status() == TaskStatus.RUNNING)
+ && task.dependencies().stream().anyMatch(dependency -> {
+ CatalogRecord input = records.get(dependency.objectId());
+ return input == null || input.deleted() || input.version() != dependency.version();
+ })) {
+ OwnerToken owner = owner(image, task.input().queue().tableName());
+ if (owner != null) {
+ task(new TaskProtocol.Cancel(owner, task.taskId(), record.version(), "Captured task inputs changed"));
+ return;
+ }
+ continue;
+ }
+ if (task.status() == TaskStatus.RUNNING
+ || task.status() == TaskStatus.RETRYABLE_FAILED && task.attempt() < task.input().maxAttempts()) {
+ if (task.status() == TaskStatus.RUNNING && attemptIsLive(image, task)) {
+ continue;
+ }
+ OwnerToken owner = owner(image, task.input().queue().tableName());
+ if (owner != null) {
+ // The state machine alone decides whether the previous exact worker session is expired or fenced.
+ task(task.attempt() < task.input().maxAttempts()
+ ? new TaskProtocol.Retry(owner, task.taskId(), record.version())
+ : new TaskProtocol.Cancel(owner, task.taskId(), record.version(),
+ "Worker expired after final attempt"));
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ public synchronized Result schedule(String tableName, UUID scheduleId, ScheduleRequest request) throws IOException {
+ ensureOpen();
+ checkArgument(_outputRoot != null, "Native controller task outputRoot is not configured");
+ checkArgument(request != null && request.minionId() != null && request.sourceUri() != null
+ && request.sourceSha256() != null, "Minion identity, source URI and digest are required");
+ resolvePending();
+ heartbeat();
+ CatalogImage image = image();
+ CatalogRecord table = record(image, RecordKind.TABLE_CONFIG, tableName);
+ checkArgument(table != null, "Table does not exist");
+ Input input = new Input(new TaskProtocol.QueueKey(tableName, table.objectId(), NativeTaskOutputIdentity.TASK_TYPE),
+ request.minionId(), Map.of("sourceUri", request.sourceUri(), "sourceSha256", request.sourceSha256()),
+ request.maxAttempts(), new TaskProtocol.OutputPolicy(_outputRoot, 1));
+ OwnerToken owner = owner(image, tableName);
+ if (owner == null) {
+ throw new IOException("Task table is owned by another controller; submit to that controller");
+ }
+ return task(new TaskProtocol.CreateSchedule(owner, scheduleId, input, List.of(), request.firstRound(),
+ request.intervalRounds(), request.occurrences()));
+ }
+
+ public synchronized Result cancel(String tableName, UUID taskId, long expectedVersion) throws IOException {
+ ensureOpen();
+ resolvePending();
+ heartbeat();
+ CatalogImage image = image();
+ CatalogRecord record = record(image, RecordKind.TASK, taskId.toString());
+ checkArgument(record != null
+ && TaskProtocol.decodeTask(record.value()).input().queue().tableName().equals(tableName),
+ "Task does not belong to this table");
+ OwnerToken owner = owner(image, tableName);
+ if (owner == null) {
+ throw new IOException("Task table is owned by another controller");
+ }
+ return task(new TaskProtocol.Cancel(owner, taskId, expectedVersion, "Cancelled through controller API"));
+ }
+
+ public synchronized CatalogImage snapshot() throws IOException {
+ ensureOpen();
+ return image();
+ }
+
+ @Nullable
+ public String failure() {
+ return _failure;
+ }
+
+ private CatalogImage image() throws IOException {
+ return _access.image(SCOPE, 4096, MetadataProtocol.MAX_IMAGE_BYTES, _timeoutMillis);
+ }
+
+ private void heartbeat() throws IOException {
+ Response current = _membership.query(new MembershipQuery(_history,
+ new MembershipProtocol.GetParticipant(_participantId)), _timeoutMillis);
+ require(current.status() == Status.OK || current.status() == Status.NOT_FOUND, "Cannot read controller session");
+ require(current.control() != null && current.control().controlEpoch() != null && current.control().policy() != null,
+ "Native task membership control is unavailable");
+ ParticipantState observed = current.participant();
+ boolean live = observed != null && !observed.retired()
+ && observed.expiryRound() > current.control().round();
+ if (_participant != null && (observed == null || !observed.participant().equals(_participant) || !live)) {
+ require(!live, "Native task controller session was replaced by a live incarnation");
+ _participant = null;
+ _registration = null;
+ _renewal = null;
+ }
+ if (_participant == null) {
+ // A lost registration reply retains its exact request. Resolve against the committed observation before
+ // allocating another incarnation: a live matching registration is retried, a live different one is fenced.
+ if (_registration != null && observed != null) {
+ boolean matching = observed.participant().incarnation().equals(_registration.incarnation())
+ && observed.participant().registrationEpoch() == _registration.expectedEpoch() + 1;
+ require(!live || matching, "Another native task controller incarnation remains live");
+ if (!live && (matching || observed.participant().registrationEpoch() != _registration.expectedEpoch())) {
+ _registration = null;
+ }
+ }
+ if (_registration == null) {
+ require(!live, "Another native task controller incarnation remains live");
+ long epoch = observed == null ? 0 : observed.participant().registrationEpoch();
+ _registration = new MembershipProtocol.Register(_participantId, MembershipProtocol.Role.CONTROLLER,
+ UUID.randomUUID(), epoch, Map.of(), Set.of());
+ }
+ Response registered = _membership.execute(new MembershipCommand(_history, _participantId, _registration),
+ _timeoutMillis);
+ require(registered.status() == Status.OK && registered.participant() != null,
+ "Native task controller registration is not available");
+ _participant = registered.participant().participant();
+ current = registered;
+ }
+ require(current.participant() != null && current.participant().participant().equals(_participant),
+ "Native task controller session was replaced");
+ if (_renewal == null) {
+ _renewal = new MembershipProtocol.Renewal(_participant,
+ current.participant().lastRenewalSequence() + 1, current.revision(), Set.of());
+ }
+ Response renewed = _membership.execute(new MembershipCommand(_history, _participantId,
+ new MembershipProtocol.RenewBatch(List.of(_renewal))), _timeoutMillis);
+ require(renewed.status() == Status.OK && renewed.renewals().size() == 1
+ && renewed.renewals().get(0).status() == Status.OK, "Native task controller renewal is unavailable");
+ _renewal = null;
+ Response ready = _membership.execute(new MembershipCommand(_history, _participantId,
+ new MembershipProtocol.SetReady(_participant, renewed.revision(), true)), _timeoutMillis);
+ require(ready.status() == Status.OK, "Native task controller is not ready");
+ }
+
+ private static boolean attemptIsLive(CatalogImage image, TaskState task) throws IOException {
+ CatalogRecord attempt = record(image, RecordKind.TASK_ATTEMPT,
+ TaskProtocol.attemptName(task.taskId(), task.attempt()));
+ require(attempt != null, "Running task has no current attempt");
+ ParticipantToken minion = TaskProtocol.decodeAttempt(attempt.value()).minion();
+ CatalogRecord registration = record(image, RecordKind.PARTICIPANT, minion.participantId());
+ if (registration == null) {
+ return false;
+ }
+ ParticipantState participant = MembershipProtocol.decodeParticipant(registration.value());
+ return participant.participant().equals(minion) && !participant.retired()
+ && participant.expiryRound() > control(image).round();
+ }
+
+ @Nullable
+ private OwnerToken owner(CatalogImage image, String tableName) throws IOException {
+ CatalogRecord layoutRecord = record(image, RecordKind.OWNER_LAYOUT, OwnershipProtocol.LAYOUT_NAME);
+ Layout layout;
+ if (layoutRecord == null) {
+ layout = new Layout(1, 24, OwnershipProtocol.HashVersion.MURMUR2_UTF8_MASK_POSITIVE_V1);
+ Layout initial = layout;
+ Result configured = command(sequence -> new MetadataProtocol.OwnershipCommand(_history, _clientId, 1,
+ "tasks", sequence, new OwnershipProtocol.ConfigureLayout(controlUnchecked(image).controlEpoch(), initial)));
+ require(configured.status() == Status.OK, "Ownership layout initialization changed");
+ } else {
+ layout = OwnershipProtocol.decodeLayout(layoutRecord.value());
+ }
+ var scope = OwnershipProtocol.tableScope(layout, tableName);
+ CatalogRecord assignmentRecord = record(image, RecordKind.OWNER_ASSIGNMENT, OwnershipProtocol.name(scope));
+ Assignment assignment =
+ assignmentRecord == null ? null : OwnershipProtocol.decodeAssignment(assignmentRecord.value());
+ long epoch = assignment == null ? 0 : assignment.ownerEpoch();
+ if (assignment != null && _participant.equals(assignment.owner())) {
+ return new OwnerToken(_history, layout.generation(), scope, epoch, _participant);
+ }
+ if (assignment != null && assignment.owner() != null) {
+ CatalogRecord previous = record(image, RecordKind.PARTICIPANT, assignment.owner().participantId());
+ ParticipantState participant = previous == null ? null : MembershipProtocol.decodeParticipant(previous.value());
+ if (participant != null && participant.participant().equals(assignment.owner()) && !participant.retired()
+ && participant.expiryRound() > control(image).round()) {
+ return null;
+ }
+ }
+ long generation = layout.generation();
+ Result assigned = command(sequence -> new MetadataProtocol.OwnershipCommand(_history, _clientId, 1, "tasks",
+ sequence, new OwnershipProtocol.AssignOwner(controlUnchecked(image).controlEpoch(), generation, scope, epoch,
+ _participant)));
+ require(assigned.status() == Status.OK, "Table ownership changed while scheduling");
+ return new OwnerToken(_history, generation, scope, epoch + 1, _participant);
+ }
+
+ private Result task(ControllerOperation operation) throws IOException {
+ return command(sequence -> new MetadataProtocol.TaskCommand(_history, _clientId, 1, "tasks", sequence, operation));
+ }
+
+ private Result command(LongFunction<Command> factory) throws IOException {
+ resolvePending();
+ Result registration = _access.execute(new MetadataProtocol.RegisterClient(_history, _clientId,
+ _clientIncarnation, 0), _timeoutMillis);
+ require(registration.status() == Status.OK && Long.valueOf(1).equals(registration.clientEpoch()),
+ "Task controller request stream is fenced");
+ var tip = _access.query(new MetadataProtocol.GetStreamTip(_history, _clientId, 1, "tasks"), _timeoutMillis);
+ require(tip.status() == Status.OK && tip.sequence() >= 0 && tip.sequence() < Long.MAX_VALUE,
+ "Task controller request stream is unavailable");
+ if (tip.sequence() > tip.floor()) {
+ require(tip.outcome() != null, "Task controller request outcome is unresolved");
+ Result acknowledged = _access.execute(new MetadataProtocol.Acknowledge(_history, _clientId, 1, "tasks",
+ tip.sequence()), _timeoutMillis);
+ require(acknowledged.status() == Status.OK, "Cannot acknowledge resolved task controller outcome");
+ }
+ _pending = factory.apply(tip.sequence() + 1);
+ Result result = _access.execute(_pending, _timeoutMillis);
+ _pending = null;
+ return result;
+ }
+
+ private void resolvePending() throws IOException {
+ if (_pending != null) {
+ _access.execute(_pending, _timeoutMillis);
+ _pending = null;
+ }
+ }
+
+ @Nullable
+ private static CatalogRecord record(CatalogImage image, RecordKind kind, String name) {
+ return image.records().stream().filter(value -> value.kind() == kind && value.name().equals(name)).findFirst()
+ .orElse(null);
+ }
+
+ private static ControlState control(CatalogImage image) throws IOException {
+ CatalogRecord record = record(image, RecordKind.MEMBERSHIP_CONTROL, MembershipProtocol.CONTROL_NAME);
+ require(record != null, "Native membership control is unavailable");
+ return MembershipProtocol.decodeControl(record.value());
+ }
+
+ private static ControlState controlUnchecked(CatalogImage image) {
+ try {
+ return control(image);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static void require(boolean valid, String message) throws IOException {
+ if (!valid) {
+ throw new IOException(message);
+ }
+ }
+
+ private void ensureOpen() throws IOException {
+ require(!_closed, "Native task scheduler is closed");
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ if (_closed) {
+ return;
+ }
+ _closed = true;
+ if (_executor != null) {
+ _executor.shutdownNow();
+ }
+ if (_participant != null) {
+ try {
+ Response response = _membership.execute(new MembershipCommand(_history, _participantId,
+ new MembershipProtocol.Unregister(_participant)), _timeoutMillis);
+ if (response.status() != Status.OK) {
+ LOGGER.warn("Native task scheduler stopped; controller retirement returned {}", response.status());
+ }
+ } catch (IOException e) {
+ _failure = "Stopped; controller retirement outcome is unknown";
+ LOGGER.warn("Native task scheduler stopped; controller retirement outcome is unknown", e);
+ }
+ }
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java
index cedef5ff253..f932bb35254 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerProvider.java
@@ -29,7 +29,7 @@ import org.apache.pinot.spi.env.PinotConfiguration;
/// Supplies explicitly qualified clock and certificate authorization inputs to the native controller lifecycle.
/// There is no default implementation: selecting a class or configuring numeric bounds does not qualify a clock.
/// Authorization policies must be current, nonblocking and thread-safe. This starter denies participant bindings
-/// with serving capabilities; query listeners and controller HTTP resources are outside this lifecycle.
+/// with query-serving capabilities. MINION_TASK is accepted only for explicitly bound MINION identities.
public interface RatisControllerProvider extends AutoCloseable {
void init(PinotConfiguration configuration) throws Exception;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java
index 787b61bb8dc..92683bcf735 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java
@@ -46,10 +46,12 @@ import org.apache.pinot.controller.api.resources.WebApplicationExceptionMapper;
import org.apache.pinot.coordination.ratis.RatisMetadataAccess;
import org.apache.pinot.coordination.ratis.RatisMetadataStore;
import org.apache.pinot.coordination.ratis.RatisMetadataStore.MembershipClockSettings;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ConfigurePolicy;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.GetControl;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Response;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipQuery;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
@@ -73,8 +75,8 @@ import static java.util.Objects.requireNonNull;
/// Owns one embedded voter, its qualified membership clock and authenticated metadata endpoint. Lifecycle methods
-/// are serialized; the normal HTTP/UI shell exposes committed metadata management, without scheduling or query
-/// readiness.
+/// are serialized; the normal HTTP/UI shell exposes committed metadata management and bounded native task scheduling.
+/// Query readiness remains disabled.
/// The current voter transport remains restricted to three loopback peers. No Helix resources are constructed.
public final class RatisControllerRuntime implements AutoCloseable, ServiceStatus.ServiceStatusCallback {
private final ControllerConf _config;
@@ -101,6 +103,8 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
private NativeControllerMetadata _metadata;
@Nullable
private NativeControllerTableSizeResource _tableSizes;
+ @Nullable
+ private NativeControllerTasks _tasks;
private volatile boolean _started;
private volatile boolean _closed;
private volatile String _status = "Native controller has not started";
@@ -151,7 +155,10 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
_endpoint = new MetadataGrpcServer(_address, _certificate, _privateKey, _trust, _access, controllers, _access,
fingerprint -> {
ParticipantIdentity identity = participants.authorize(fingerprint);
- return identity != null && identity.capabilities().isEmpty() ? identity : null;
+ return identity != null && (identity.capabilities().isEmpty()
+ || identity.role() == Role.MINION
+ && identity.capabilities().equals(Set.of(Capability.MINION_TASK)))
+ ? identity : null;
}, audit, Limits.defaults());
startHttp(accessControlFactory);
checkClock();
@@ -186,6 +193,9 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
PinotControllerAppConfigs.class, PinotControllerHealthCheck.class, WebApplicationExceptionMapper.class));
_tableSizes = new NativeControllerTableSizeResource(metadata, _config, accessControlFactory, metrics);
_adminApp.register(_tableSizes);
+ _tasks =
+ new NativeControllerTasks(requireNonNull(_access), requireNonNull(_access), _history, _instanceId, _config);
+ _adminApp.register(new NativeControllerTaskResource(_tasks, accessControlFactory));
Instant startTime = Instant.now();
_adminApp.registerBinder(new AbstractBinder() {
@Override
@@ -200,6 +210,7 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
});
_adminApp.registerBinder(new AuditServiceBinder(metadata, ServiceRole.CONTROLLER, metrics));
_adminApp.start(_listeners, metrics);
+ _tasks.start();
}
private static void validateBootstrapPolicy(Policy policy) {
@@ -303,7 +314,7 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
_closed = true;
_started = false;
Exception failure = null;
- AutoCloseable[] resources = {_adminApp == null ? null : _adminApp::stop, _tableSizes, _metadata,
+ AutoCloseable[] resources = {_adminApp == null ? null : _adminApp::stop, _tasks, _tableSizes, _metadata,
_endpoint, _access, _store, _provider};
for (AutoCloseable resource : resources) {
if (resource != null) {
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java
new file mode 100644
index 00000000000..5966cb1ecfc
--- /dev/null
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java
@@ -0,0 +1,318 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.coordination;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MetadataCatalog;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogImage;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ChangePage;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Command;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedCursor;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetStreamTip;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipCommand;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipQuery;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Mutation;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Query;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ReadBarrier;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.StreamTipResult;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol;
+import org.apache.pinot.coordination.ratis.transport.MembershipAccess;
+import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Runs controller scheduling against the durable catalog to cover lost replies and owner recovery across restart.
+public class NativeControllerTasksTest {
+ @Test
+ public void testScheduleLostReplyFailoverAndImmutableOccurrence() throws Exception {
+ try (Fixture f = new Fixture();
+ NativeControllerTasks first = f.controller("first");
+ NativeControllerTasks competing = f.controller("second")) {
+ UUID scheduleId = UUID.randomUUID();
+ var request = new NativeControllerTasks.ScheduleRequest("minion", "file:///approved/rows.json", "1".repeat(64),
+ 0, 0, 1, 2);
+ f._loseScheduleReply = true;
+ expectThrows(IOException.class, () -> first.schedule("orders_OFFLINE", scheduleId, request));
+ assertEquals(first.schedule("orders_OFFLINE", scheduleId, request).status(), Status.OK);
+ expectThrows(IOException.class, () -> competing.schedule("orders_OFFLINE", UUID.randomUUID(), request));
+ first.tick();
+ first.tick();
+ assertEquals(f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK), "tasks")).records().size(), 1);
+ first.close();
+ MetadataCatalog recovered = new MetadataCatalog(f._cluster, f._history);
+ recovered.restoreSnapshot(f._catalog.captureSnapshot());
+ f._catalog = recovered;
+ try (NativeControllerTasks replacement = f.controller("first")) {
+ assertEquals(replacement.schedule("orders_OFFLINE", scheduleId, request).status(), Status.OK);
+ replacement.tick();
+ assertEquals(f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK), "tasks")).records().size(), 1);
+ var different = new NativeControllerTasks.ScheduleRequest("minion", "file:///approved/other.json",
+ "1".repeat(64), 0, 0, 1, 2);
+ assertEquals(replacement.schedule("orders_OFFLINE", scheduleId, different).status(), Status.CONFLICT);
+ var schedules = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK_SCHEDULE), "schedules")).records();
+ assertEquals(TaskProtocol.decodeSchedule(schedules.getFirst().value()).firedOccurrences(), 1L);
+ }
+ }
+ }
+
+ @Test
+ public void testUnknownMembershipRepliesAndExpiredControllerRecoverWithoutStealingLiveSession() throws Exception {
+ try (Fixture f = new Fixture(); NativeControllerTasks controller = f.controller("first");
+ NativeControllerTasks competing = f.controller("first")) {
+ f._loseRegistrationReply = true;
+ expectThrows(IOException.class, controller::tick);
+ var first = f.participant("Controller_first").participant();
+ controller.tick();
+ assertEquals(f.participant("Controller_first").participant(), first,
+ "An unknown registration outcome must retry the same incarnation");
+ f._loseRenewalReply = true;
+ expectThrows(IOException.class, controller::tick);
+ long renewal = f.participant("Controller_first").lastRenewalSequence();
+ controller.tick();
+ assertEquals(f.participant("Controller_first").lastRenewalSequence(), renewal,
+ "An unknown renewal outcome must not allocate another sequence");
+ expectThrows(IOException.class, competing::tick);
+ assertEquals(f.participant("Controller_first").participant(), first);
+ f.expireLeases();
+ controller.tick();
+ var recovered = f.participant("Controller_first").participant();
+ assertEquals(recovered.registrationEpoch(), first.registrationEpoch() + 1);
+ assertNotEquals(recovered.incarnation(), first.incarnation());
+ assertTrue(f.participant("Controller_first").ready());
+ f.expireLeases();
+ competing.tick();
+ var replacement = f.participant("Controller_first").participant();
+ expectThrows(IOException.class, controller::tick);
+ assertEquals(f.participant("Controller_first").participant(), replacement,
+ "A recovered process must not replace another live incarnation");
+ }
+ }
+
+ @Test
+ public void testLiveAttemptDoesNotWriteRetriesAndExpiredFinalAttemptIsFenced() throws Exception {
+ try (Fixture f = new Fixture(); NativeControllerTasks controller = f.controller("first")) {
+ var request = new NativeControllerTasks.ScheduleRequest("minion", "file:///approved/rows.json", "1".repeat(64),
+ 0, 0, 1, 1);
+ assertEquals(controller.schedule("orders_OFFLINE", UUID.randomUUID(), request).status(), Status.OK);
+ controller.tick();
+ var taskRecord = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK), "tasks")).records().getFirst();
+ var task = TaskProtocol.decodeTask(taskRecord.value());
+ var minionRead = f.query(new MembershipQuery(f._history, new MembershipProtocol.GetParticipant("minion")), 1000);
+ var minion = minionRead.participant().participant();
+ assertEquals(f.execute(new MembershipCommand(f._history, "minion", new MembershipProtocol.SetReady(minion,
+ minionRead.revision(), true)), 1000).status(), Status.OK);
+ assertEquals(f.execute(new MetadataProtocol.TaskSessionCommand(minion, new MetadataProtocol.RegisterClient(
+ f._history, TaskProtocol.clientId(minion), minion.incarnation(), 0)), 1000).status(), Status.OK);
+ assertEquals(f.execute(new MetadataProtocol.TaskSessionCommand(minion, new MetadataProtocol.NativeTaskCommand(
+ f._history, TaskProtocol.clientId(minion), 1, "work", 1, minion,
+ new TaskProtocol.Claim(task.taskId(), taskRecord.version()))), 1000).status(), Status.OK);
+ int taskCommands = f._taskCommands;
+ controller.tick();
+ assertEquals(f._taskCommands, taskCommands, "A live exact attempt must not generate doomed retry commands");
+ f.expireLeases();
+ controller.tick();
+ var current = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK), "tasks")).records().getFirst();
+ assertEquals(TaskProtocol.decodeTask(current.value()).status(), TaskProtocol.TaskStatus.CANCELLED);
+ var attempt = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK_ATTEMPT), "attempts"))
+ .records().getFirst();
+ assertEquals(TaskProtocol.decodeAttempt(attempt.value()).status(), TaskProtocol.AttemptStatus.FENCED);
+ }
+ }
+
+ @Test
+ public void testChangedInputsCancelReadyAndRunningTasks() throws Exception {
+ for (boolean running : List.of(false, true)) {
+ try (Fixture f = new Fixture(); NativeControllerTasks controller = f.controller("first")) {
+ var request = new NativeControllerTasks.ScheduleRequest("minion", "file:///approved/rows.json", "1".repeat(64),
+ 0, 0, 1, 2);
+ assertEquals(controller.schedule("orders_OFFLINE", UUID.randomUUID(), request).status(), Status.OK);
+ controller.tick();
+ var taskRecord = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK), "tasks")).records().getFirst();
+ var task = TaskProtocol.decodeTask(taskRecord.value());
+ if (running) {
+ var read = f.query(new MembershipQuery(f._history, new MembershipProtocol.GetParticipant("minion")), 1000);
+ var minion = read.participant().participant();
+ assertEquals(f.execute(new MembershipCommand(f._history, "minion", new MembershipProtocol.SetReady(minion,
+ read.revision(), true)), 1000).status(), Status.OK);
+ assertEquals(f.execute(new MetadataProtocol.TaskSessionCommand(minion, new MetadataProtocol.RegisterClient(
+ f._history, TaskProtocol.clientId(minion), minion.incarnation(), 0)), 1000).status(), Status.OK);
+ assertEquals(f.execute(new MetadataProtocol.TaskSessionCommand(minion, new MetadataProtocol.NativeTaskCommand(
+ f._history, TaskProtocol.clientId(minion), 1, "work", 1, minion,
+ new TaskProtocol.Claim(task.taskId(), taskRecord.version()))), 1000).status(), Status.OK);
+ }
+ var schema = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.SCHEMA), "schema")).records().getFirst();
+ assertEquals(f.execute(new MetadataProtocol.Acknowledge(f._history, "bootstrap", 1, "config", 1), 1000)
+ .status(), Status.OK);
+ assertEquals(f.execute(new MetadataProtocol.Transaction(f._history, "bootstrap", 1, "config", 2,
+ List.of(Mutation.updateSchema(new ObjectVersion(f._history, schema.objectId(), schema.version()),
+ new Schema.SchemaBuilder().setSchemaName("orders").addSingleValueDimension("id", DataType.STRING)
+ .build()))), 1000).status(), Status.OK);
+ controller.tick();
+ var cancelled = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK), "tasks")).records().getFirst();
+ assertEquals(TaskProtocol.decodeTask(cancelled.value()).status(), TaskProtocol.TaskStatus.CANCELLED);
+ if (running) {
+ var attempt = f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK_ATTEMPT), "attempts"))
+ .records().getFirst();
+ assertEquals(TaskProtocol.decodeAttempt(attempt.value()).status(), TaskProtocol.AttemptStatus.FENCED);
+ }
+ }
+ }
+ }
+
+ private static final class Fixture implements MetadataAccess, MembershipAccess, AutoCloseable {
+ private final UUID _cluster = UUID.randomUUID();
+ private final UUID _history = UUID.randomUUID();
+ private final Path _output = Files.createTempDirectory("native-controller-tasks-");
+ private MetadataCatalog _catalog = new MetadataCatalog(_cluster, _history);
+ private long _index;
+ private boolean _loseScheduleReply;
+ private boolean _loseRegistrationReply;
+ private boolean _loseRenewalReply;
+ private int _taskCommands;
+
+ private Fixture() throws Exception {
+ var control = execute(new MembershipCommand(_history, "bootstrap",
+ new MembershipProtocol.ActivateControlEpoch(1, "leader", UUID.randomUUID())), 1000);
+ assertEquals(execute(new MembershipCommand(_history, "bootstrap",
+ new MembershipProtocol.ConfigurePolicy(control.control().controlEpoch(), 0,
+ new MembershipProtocol.Policy(1, 1000, 4, 1000, 500, 100))), 1000).status(), Status.OK);
+ assertEquals(execute(new MembershipCommand(_history, "bootstrap", new MembershipProtocol.Register("minion",
+ MembershipProtocol.Role.MINION, UUID.randomUUID(), 0, Map.of(),
+ Set.of(MembershipProtocol.Capability.MINION_TASK))), 1000).status(), Status.OK);
+ assertEquals(execute(new MetadataProtocol.RegisterClient(_history, "bootstrap", UUID.randomUUID(), 0),
+ 1000).status(), Status.OK);
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("orders")
+ .addSingleValueDimension("id", DataType.LONG).build();
+ var table = new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").setNumReplicas(1).build();
+ assertEquals(execute(new MetadataProtocol.Transaction(_history, "bootstrap", 1, "config", 1, List.of(
+ Mutation.createSchema(_history, UUID.randomUUID(), schema),
+ Mutation.createTableConfig(_history, UUID.randomUUID(), table))), 1000).status(),
+ Status.OK);
+ }
+
+ private NativeControllerTasks controller(String id) throws IOException {
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(NativeMetadataConfig.PREFIX + "id", id);
+ conf.setProperty(NativeMetadataConfig.PREFIX + "controller.tasks.outputRoot", _output.toString());
+ return new NativeControllerTasks(this, this, _history, "Controller_" + id, conf);
+ }
+
+ private MembershipProtocol.ParticipantState participant(String id) throws IOException {
+ return query(new MembershipQuery(_history, new MembershipProtocol.GetParticipant(id)), 1000).participant();
+ }
+
+ private void expireLeases() throws IOException {
+ for (int i = 0; i < 4; i++) {
+ var control = _catalog.membershipControl();
+ assertEquals(execute(new MembershipCommand(_history, "bootstrap", new MembershipProtocol.Tick(
+ control.controlEpoch(), control.round(), control.policy().version())), 1000).status(), Status.OK);
+ }
+ }
+
+ private byte[] apply(Command command) throws IOException {
+ return _catalog.execute(MetadataProtocol.encode(command), 1, ++_index);
+ }
+
+ @Override
+ public Result execute(Command command, long timeoutMillis) throws IOException {
+ Result result = MetadataProtocol.decodeResult(apply(command));
+ if (command instanceof MetadataProtocol.TaskCommand) {
+ _taskCommands++;
+ }
+ if (_loseScheduleReply && command instanceof MetadataProtocol.TaskCommand task
+ && task.request() instanceof TaskProtocol.CreateSchedule) {
+ _loseScheduleReply = false;
+ assertEquals(result.status(), Status.OK);
+ throw new IOException("Lost response after committed schedule");
+ }
+ return result;
+ }
+
+ @Override
+ public MembershipProtocol.Response execute(MembershipCommand command, long timeoutMillis) throws IOException {
+ MembershipProtocol.Response result = MembershipProtocol.decodeResponse(apply(command));
+ if (_loseRegistrationReply && command.request() instanceof MembershipProtocol.Register) {
+ _loseRegistrationReply = false;
+ throw new IOException("Lost committed registration response");
+ }
+ if (_loseRenewalReply && command.request() instanceof MembershipProtocol.RenewBatch) {
+ _loseRenewalReply = false;
+ throw new IOException("Lost committed renewal response");
+ }
+ return result;
+ }
+
+ @Override
+ public MembershipProtocol.Response query(MembershipQuery query, long timeoutMillis) throws IOException {
+ return MembershipProtocol.decodeResponse(apply(new ReadBarrier(query)));
+ }
+
+ @Override
+ public Result query(Query query, long timeoutMillis) throws IOException {
+ return MetadataProtocol.decodeResult(apply(new ReadBarrier(query)));
+ }
+
+ @Override
+ public StreamTipResult query(GetStreamTip query, long timeoutMillis) throws IOException {
+ return MetadataProtocol.decodeStreamTipResult(apply(new ReadBarrier(query)));
+ }
+
+ @Override
+ public CatalogImage image(FeedScope scope, int maxRecords, int maxBytes, long timeoutMillis) throws IOException {
+ return _catalog.captureImage(scope, maxRecords, maxBytes);
+ }
+
+ @Override
+ public ChangePage changes(FeedScope scope, FeedCursor cursor, int maxTransactions, int maxRecords, int maxBytes,
+ long timeoutMillis) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {
+ assertTrue(Files.isDirectory(_output));
+ FileUtils.deleteDirectory(_output.toFile());
+ }
+ }
+}
diff --git a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md
index 8cd6bbc86d0..ea6d742360e 100644
--- a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md
+++ b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md
@@ -29,8 +29,9 @@ The standard resource package has **264 annotated HTTP methods across 43 resourc
- Baseline: validated configuration reads/writes, controller authentication, HTTP/HTTPS, and metadata UI.
- Added in this change: shared table-size metadata adapter, optional read-only native server admin endpoint,
identity-checked native OFFLINE table-size REST/UI support. Query listeners remain disabled.
-- In progress separately: minion metadata-read extraction. It does not authorize native task execution.
-- Outstanding: offline lifecycle and placement, durable minion scheduling/execution/publication, realtime,
+- Native minions: explicit `MINION_TASK` authorization, durable bounded schedules, normal minion startup and atomic
+ output publication for `NativeSegmentGenerationTask`; see [the task contract](MINION-TASK-PLAN.md).
+- Outstanding: automatic offline placement, full existing task API/plugin parity, realtime,
remaining feature families and query-serving qualification. A full replacement is not yet implemented.
- ZooKeeper tree APIs are not applicable to the replacement; a mutable compatibility tree will not be introduced.
diff --git a/pinot-coordination-ratis/MINION-TASK-PLAN.md b/pinot-coordination-ratis/MINION-TASK-PLAN.md
index 8debf3634f5..55d88c0b57f 100644
--- a/pinot-coordination-ratis/MINION-TASK-PLAN.md
+++ b/pinot-coordination-ratis/MINION-TASK-PLAN.md
@@ -20,31 +20,110 @@
-->
# Native minion task framework implementation plan
-Status: **draft; the internal execution ledger is implemented, while native scheduling, worker runtime and task
-publication remain unimplemented**.
+Status: **draft; the execution ledger now includes bounded occurrence scheduling, an authenticated minion worker,
+and fenced publication for `NativeSegmentGenerationTask`. General task-plugin and task-API parity remain incomplete.**
The metadata read adapter is the first integration step. `MinionMetadataAccessor` exposes table, schema and segment
reads; `HelixMinionMetadataAccessor` preserves the current behavior. Existing property-store bindings install that
adapter. An explicit alternate read accessor cannot enable legacy task execution. The existing task factories still
accept `MinionTaskZkMetadataManager`, and existing plugins still contain direct Helix calls and unfenced publication.
-The `MINION` membership enum alone is insufficient: participant transport authorization currently rejects this role.
+The dedicated minion transport requires an explicit certificate binding with the `MINION_TASK` capability.
+No minion receives a controller metadata identity or query-serving capability.
The internal ledger now supports table-owned submissions, queue pause/resume, explicit minion identity assignments,
session-fenced claims, monotonic progress, execution outcomes, bounded retries and cancellation. Captured table/schema
-and declared input versions are immutable. Snapshot format 9 preserves task queues, tasks, attempt history and the
-existing durable request-stream sequence/floor; format 8 remains readable when it contains no task records. Typed
-protobuf values expose controller read/feed projections only. No task execution service or public task command exists.
-`EXECUTION_FINISHED` records worker execution and does not certify publication or successful task completion.
+and declared input versions are immutable. Snapshot format 10 adds schedules and publication manifests to task queues,
+tasks, attempt history and the durable request-stream sequence/floor. Formats 8 and 9 remain readable only with their
+original record families and states. Typed
+protobuf values expose bounded controller and own-minion projections. `EXECUTION_FINISHED` remains an execution-only
+legacy ledger state. Output-producing native tasks can become `SUCCEEDED` only in the transaction that creates their
+publication manifest and immutable offline content records.
Retry of a running attempt requires expiry, retirement or replacement of its exact session. Readiness and draining
-changes alone cannot permit stealing that attempt. Controller occurrence scheduling, plugin capability negotiation,
-resource/concurrency admission, timers/backoff, checkpoint mutation and output publication remain future work.
+changes alone cannot permit stealing that attempt. The first worker has one active attempt at a time, finite retries,
+and immutable inputs and outputs. General plugin negotiation, resource pools, cron schedules, execution timeouts,
+retry backoff and plugin checkpoint mutation remain future work.
-Helix/ZooKeeper remains the default. This plan does not enable native query listeners or authorize native minion
-execution. A task type becomes available only after its scheduler, executor, observers and publication paths satisfy
+Helix/ZooKeeper remains the default. Native query listeners remain disabled. A task type becomes available only
+after its scheduler, executor, observers and publication paths satisfy
the gates below. An unavailable task type must return an explicit unsupported error, rather than an empty successful
queue or a task marked complete without its effects.
+## Implemented execution slice
+
+`StartMinion` now selects the native runtime only for `pinot.coordination.backend=RATIS`, before constructing a Helix
+manager or applying ZK cluster configuration. It uses the normal minion instance/data-directory settings and the
+existing native identity, endpoints and TLS settings. It never initializes legacy task factories or mutable
+`MinionContext` bindings.
+
+The qualified task generates a genuine Pinot OFFLINE segment from a digest-pinned JSON-lines file using Pinot's
+segment creation driver. This loopback foundation requires approved local source and shared output directories:
+
+```properties
+pinot.coordination.backend=RATIS
+pinot.minion.instance.id=Minion_native_0
+dataDir=/absolute/minion-state
+pinot.coordination.ratis.participantId=Minion_native_0
+pinot.coordination.ratis.minion.sourceRoot=/absolute/task-input
+pinot.coordination.ratis.minion.outputRoot=/absolute/task-output
+```
+
+Configure the usual `pinot.coordination.ratis.historyId`, `metadata.endpoints`, `tls.certChain`, `tls.privateKey` and
+`tls.trustCerts` values as for other native participants. The controller's provider must explicitly bind the minion
+certificate to this participant ID, role `MINION` and capability `MINION_TASK`. On each controller, configure
+`pinot.coordination.ratis.controller.tasks.outputRoot=/absolute/task-output`. All output roots must identify the same
+existing directory; runtime construction does not grant access to an arbitrary path from a REST request.
+
+The controller exposes these experimental, authorized, table-scoped APIs:
+
+- `PUT /tasks/native/{table}/schedules/{uuid}` creates an immutable schedule generation.
+- `GET /tasks/native/{table}` returns committed schedules, queues, tasks and publications, plus the current round.
+- `POST /tasks/native/{table}/{taskUuid}/cancel?expectedVersion={version}` fences the selected task version.
+
+For an immediate one-shot task, submit the following body with a fresh schedule UUID:
+
+```json
+{
+ "minionId": "Minion_native_0",
+ "sourceUri": "file:///absolute/task-input/rows.json",
+ "sourceSha256": "<64 lowercase hexadecimal SHA-256 characters>",
+ "firstRound": 0,
+ "intervalRounds": 0,
+ "occurrences": 1,
+ "maxAttempts": 2
+}
+```
+
+The same schedule ID and identical request resolve the same schedule; different inputs conflict. A recurring finite
+schedule uses an absolute `firstRound`, positive `intervalRounds` and bounded `occurrences`. It catches up one
+occurrence per scheduler cycle. These are committed membership rounds, not cron or wall-clock timestamps; a membership
+timing-policy change fences that schedule generation. Only its live table owner schedules or retries tasks.
+Admission applies backpressure before a new runnable task can exceed the worker's complete bounded projection;
+an unaccepted occurrence does not advance its schedule watermark.
+
+The worker stores exact pending requests and outcomes in its durable directory, retries unknown responses without
+changing their identity, and separates heartbeats from execution. A new process receives a new registration incarnation
+and cannot replay an old attempt's publication. Lease uncertainty stops admission and generation; replicated validation
+is the final authority. Cancellation acceptance revokes publication and does not assert that remote computation has
+already stopped.
+
+Files are created without overwrite under history/task/attempt/incarnation paths and synced before publication.
+The publication transaction validates the current live minion session, attempt, table incarnation, captured input
+versions and output identities. It atomically creates the immutable content records, publication manifest and
+`SUCCEEDED` task/attempt states. A stale worker can leave an unreferenced artifact but cannot publish it. No
+automatic physical cleanup or exactly-once computation is claimed.
+
+The initial bounds are explicit: one active attempt per minion, a 128-record/2 MiB worker projection, a 4,096-record
+controller task image, and 128 retained local process journals before explicit archival is required. Terminal
+catalog history is retained; exceeding the controller image limit fails closed. Retention/compaction and larger-scale
+qualification remain outstanding. Input files default to 16 MiB and 100,000 rows, with component-configured ceilings
+of 64 MiB and 1,000,000 rows.
+
+Publication means committed content, not replica installation or query availability. The integration test passes a
+real generated output through the existing native assignment/installation protocol and measures its installed bytes.
+Automatic placement, full existing `/tasks` REST/UI compatibility, remote deep-store adapters, plugin checkpoint
+mutation, replacement/deletion/lineage and other legacy task types remain separate requirements below.
+
## Authority and durability requirements
- Task records, scheduling decisions, claims and outcomes are replicated through Ratis. A local scheduler queue or
diff --git a/pinot-coordination-ratis/README.md b/pinot-coordination-ratis/README.md
index 766a4b5f509..4414c4dae39 100644
--- a/pinot-coordination-ratis/README.md
+++ b/pinot-coordination-ratis/README.md
@@ -541,7 +541,12 @@ Validate the shared query wire context and deadline/permit composition:
-Dsurefire.failIfNoSpecifiedTests=false test
```
+The native minion execution slice adds bounded durable schedules, the normal `StartMinion` lifecycle and fenced
+publication for actual OFFLINE JSON-lines segment generation. Its explicit configuration, REST endpoints and remaining
+plugin/API gaps are documented in [MINION-TASK-PLAN.md](MINION-TASK-PLAN.md). Helix/ZK stays the default and native query
+listeners stay disabled.
+
Metadata durability, distribution, membership and starter wiring are parts of the replacement. Full role integration,
-qualified liveness and query fencing, controller ownership integration, assignment and state transitions, realtime
-ingestion, minion tasks, migration and a standalone Ratis quickstart require their own integration and qualification
-before a supported replacement is claimed.
+qualified liveness and query fencing, automatic assignment and state transitions, realtime ingestion, general minion
+plugin/API parity, migration and a standalone Ratis quickstart require their own integration and qualification before
+a supported replacement is claimed.
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java
index af21b8830e0..27c38caf755 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/MetadataStateMachine.java
@@ -34,6 +34,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ChangePage;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedCursor;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
import org.apache.ratis.proto.RaftProtos.LogEntryProto;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientRequest;
@@ -174,6 +175,12 @@ final class MetadataStateMachine extends BaseStateMachine {
return _catalog.installationSnapshot(server, maxRecords, maxBytes);
}
+ synchronized WorkSnapshot taskSnapshot(ParticipantToken minion, int maxRecords, int maxBytes, long minimumIndex,
+ long deadlineNanos) throws IOException {
+ awaitApplied(minimumIndex, deadlineNanos);
+ return _catalog.taskSnapshot(minion, maxRecords, maxBytes);
+ }
+
synchronized ChangePage changes(FeedScope scope, FeedCursor cursor, int maxTransactions, int maxRecords,
int maxBytes, long minimumIndex, long deadlineNanos)
throws IOException {
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java
index 7d256786410..4ba416cf0dc 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/RatisMetadataAccess.java
@@ -40,15 +40,18 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.StreamTipResult;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
import org.apache.pinot.coordination.ratis.transport.InstallationAccess;
import org.apache.pinot.coordination.ratis.transport.MembershipAccess;
import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
+import org.apache.pinot.coordination.ratis.transport.TaskAccess;
/// Controller-side metadata access for an authenticated service. Reads establish a consensus barrier before a local
/// coherent capture; no full catalog image crosses the Raft client transport. The enclosing service authorizes actors.
/// This class is thread-safe and owns its clients, but does not own the voter.
-public final class RatisMetadataAccess implements MetadataAccess, MembershipAccess, InstallationAccess, AutoCloseable {
+public final class RatisMetadataAccess
+ implements MetadataAccess, MembershipAccess, InstallationAccess, TaskAccess, AutoCloseable {
private final MetadataStateMachine _stateMachine;
private final RatisMetadataStore.Client _client;
private final RatisMetadataStore.Client _membershipClient;
@@ -125,6 +128,14 @@ public final class RatisMetadataAccess implements MetadataAccess, MembershipAcce
return _stateMachine.installationSnapshot(server, maxRecords, maxBytes, index, deadline);
}
+ @Override
+ public WorkSnapshot taskSnapshot(ParticipantToken minion, int maxRecords, int maxBytes, long timeoutMillis)
+ throws IOException {
+ long deadline = deadline(timeoutMillis);
+ long index = readBarrier(deadline);
+ return _stateMachine.taskSnapshot(minion, maxRecords, maxBytes, index, deadline);
+ }
+
@Override
public ChangePage changes(FeedScope scope, FeedCursor cursor, int maxTransactions, int maxRecords, int maxBytes,
long timeoutMillis)
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipProtocol.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipProtocol.java
index 8713f66e715..a33de01e543 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipProtocol.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipProtocol.java
@@ -38,6 +38,8 @@ public final class MembershipProtocol {
public static final String CONTROL_NAME = "control";
public static final int MAX_RENEWALS = MetadataProtocol.MAX_MUTATIONS;
public static final int MAX_ENDPOINTS = 16;
+ public static final int MAX_IDENTIFIER_LENGTH = 1024;
+ public static final int MAX_ENDPOINT_VALUE_LENGTH = 2048;
private MembershipProtocol() {
}
@@ -46,9 +48,9 @@ public final class MembershipProtocol {
CONTROLLER, BROKER, SERVER, MINION
}
- /// These capabilities never authorize ingestion, ownership, publication or destructive cleanup.
+ /// Capabilities select a protocol lane; task publication additionally requires exact task/attempt and output grants.
public enum Capability {
- BROKER_QUERY, SERVER_QUERY
+ BROKER_QUERY, SERVER_QUERY, MINION_TASK
}
public record ParticipantToken(UUID historyId, String participantId, Role role, UUID incarnation,
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipState.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipState.java
index 595523abddd..d79c6aa1676 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipState.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MembershipState.java
@@ -489,7 +489,7 @@ final class MembershipState {
"Participant endpoints exceed the supported bound");
endpoints.forEach((name, value) -> {
identifier(name);
- check(value != null && !value.isBlank() && value.length() <= 2048
+ check(value != null && !value.isBlank() && value.length() <= MembershipProtocol.MAX_ENDPOINT_VALUE_LENGTH
&& value.chars().noneMatch(Character::isISOControl), Status.INVALID_REQUEST, "Invalid participant endpoint");
});
}
@@ -498,13 +498,14 @@ final class MembershipState {
check(role != null && capabilities != null, Status.INVALID_REQUEST, "Participant role/capabilities are required");
for (Capability capability : capabilities) {
check((role == Role.BROKER && capability == Capability.BROKER_QUERY)
- || (role == Role.SERVER && capability == Capability.SERVER_QUERY), Status.INVALID_REQUEST,
+ || (role == Role.SERVER && capability == Capability.SERVER_QUERY)
+ || (role == Role.MINION && capability == Capability.MINION_TASK), Status.INVALID_REQUEST,
"Capability is incompatible with participant role");
}
}
private static void identifier(String value) {
- check(value != null && !value.isBlank() && value.length() <= 1024
+ check(value != null && !value.isBlank() && value.length() <= MembershipProtocol.MAX_IDENTIFIER_LENGTH
&& value.chars().noneMatch(Character::isISOControl), Status.INVALID_REQUEST, "Invalid membership identifier");
}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java
index eb2b0824a81..c6d84e22aba 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalog.java
@@ -74,6 +74,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.StreamTipResult;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskSessionCommand;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskSessionQuery;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Transaction;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.WorkResult;
@@ -81,6 +82,7 @@ import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Assig
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.AssignedOfflineSegmentReload;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.OperationState;
import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.State;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.Schema;
@@ -97,7 +99,7 @@ import org.apache.pinot.spi.utils.builder.TableNameBuilder;
public final class MetadataCatalog {
private static final int MAX_IDENTIFIER_LENGTH = 1024;
private static final int MAX_RESULT_BYTES = 2 * MetadataProtocol.MAX_COMMAND_BYTES;
- private static final int SNAPSHOT_FORMAT_VERSION = 9;
+ private static final int SNAPSHOT_FORMAT_VERSION = 10;
private static final int JOURNAL_POLICY_VERSION = 1;
static final int MAX_JOURNAL_TRANSACTIONS = 1024;
static final int MAX_JOURNAL_BYTES = 16 * MetadataProtocol.MAX_COMMAND_BYTES;
@@ -266,6 +268,13 @@ public final class MetadataCatalog {
Status.FENCED, "Installation outcome belongs to another participant session");
yield readOutcome(session.request());
}
+ case TaskSessionQuery session -> {
+ TaskCatalog.validateSession(session.minion(), coordinationView());
+ require(session.request() != null
+ && TaskProtocol.clientId(session.minion()).equals(session.request().clientId()),
+ Status.FENCED, "Task outcome belongs to another participant session");
+ yield readOutcome(session.request());
+ }
case GetInstallationWork ignored -> throw new IllegalStateException("Installation work dispatch failed");
case GetStreamTip ignored -> throw new IllegalStateException("Stream tip dispatch failed");
case MembershipQuery ignored -> throw new IllegalStateException("Membership dispatch failed");
@@ -324,6 +333,70 @@ public final class MetadataCatalog {
return _membership.control();
}
+ /// One complete actionable own-minion image at the read barrier. Terminal history is resolved through outcomes;
+ /// it does not consume the work image bound. Absence revokes work, never certifies successful completion.
+ public synchronized WorkSnapshot taskSnapshot(ParticipantToken minion, int maxRecords, int maxBytes)
+ throws IOException {
+ try {
+ require(maxRecords > 0 && maxRecords <= TaskProtocol.MAX_SNAPSHOT_RECORDS
+ && maxBytes >= TaskProtocol.MIN_SNAPSHOT_BYTES && maxBytes <= TaskProtocol.MAX_SNAPSHOT_BYTES,
+ Status.INVALID_REQUEST, "Invalid task snapshot bounds");
+ OperationCatalog.View view = coordinationView();
+ TaskCatalog.validateSession(minion, view);
+ WorkSnapshot header = new WorkSnapshot(Status.OK, revision(_lastAppliedIndex), "",
+ view.participant(minion.participantId()), List.of(), view.control());
+ long bytes = MetadataProtocol.encodeValue(header).length;
+ require(bytes <= maxBytes, Status.RESPONSE_TOO_LARGE, "Task snapshot header exceeds its byte bound");
+ Map<UUID, CatalogRecord> selected = new HashMap<>();
+ for (CatalogRecord record : _records.values()) {
+ if (record.kind() != RecordKind.TASK || record.deleted()) {
+ continue;
+ }
+ TaskProtocol.TaskState task = TaskProtocol.decodeTask(record.value());
+ if (!task.input().minionId().equals(minion.participantId())
+ || (task.status() != TaskProtocol.TaskStatus.READY && task.status() != TaskProtocol.TaskStatus.RUNNING)) {
+ continue;
+ }
+ bytes = addTaskRecord(selected, record, bytes, maxRecords, maxBytes);
+ CatalogRecord queue = view.record(RecordKind.TASK_QUEUE, TaskProtocol.queueName(task.input().queue()));
+ require(queue != null, Status.NOT_READY, "Assigned task queue is absent");
+ bytes = addTaskRecord(selected, queue, bytes, maxRecords, maxBytes);
+ for (ObjectVersion dependency : task.dependencies()) {
+ CatalogRecord input = view.record(dependency.objectId());
+ require(input != null, Status.NOT_READY, "Assigned task dependency is absent");
+ bytes = addTaskRecord(selected, input, bytes, maxRecords, maxBytes);
+ }
+ if (task.status() == TaskProtocol.TaskStatus.RUNNING) {
+ CatalogRecord attempt = view.record(RecordKind.TASK_ATTEMPT,
+ TaskProtocol.attemptName(task.taskId(), task.attempt()));
+ require(attempt != null, Status.NOT_READY, "Assigned task attempt is absent");
+ bytes = addTaskRecord(selected, attempt, bytes, maxRecords, maxBytes);
+ }
+ }
+ List<CatalogRecord> records = selected.values().stream().sorted(Comparator.comparing(CatalogRecord::kind)
+ .thenComparing(CatalogRecord::name)).toList();
+ return new WorkSnapshot(Status.OK, header.revision(), "", header.participant(), records, header.control());
+ } catch (Rejection e) {
+ return new WorkSnapshot(e._status, revision(_lastAppliedIndex), e.getMessage(), null, List.of(), null);
+ } catch (CoordinationRejection e) {
+ return new WorkSnapshot(e.status(), revision(_lastAppliedIndex), e.getMessage(), null, List.of(), null);
+ }
+ }
+
+ /// Use maintained canonical sizes before adding records, so oversized work cannot force large serialization
+ /// while holding the Raft apply monitor. The transport separately enforces its protobuf byte budget.
+ private long addTaskRecord(Map<UUID, CatalogRecord> selected, CatalogRecord record, long bytes, int maxRecords,
+ int maxBytes) {
+ if (selected.containsKey(record.objectId())) {
+ return bytes;
+ }
+ long nextBytes = bytes + _recordBytes.get(record.objectId()) + (selected.isEmpty() ? 0 : 1);
+ require(selected.size() < maxRecords && nextBytes <= maxBytes, Status.RESPONSE_TOO_LARGE,
+ "Assigned task image exceeds its record or byte bound");
+ selected.put(record.objectId(), record);
+ return nextBytes;
+ }
+
/// Captures one complete own-session projection after the caller's consensus read barrier. Readiness is not
/// required for bootstrap. The revision describes this whole projection, not arbitrary operation lookup progress.
/// Fixed bounds deliberately reject a large image; pagination must not be simulated by dropping records here.
@@ -456,6 +529,17 @@ public final class MetadataCatalog {
return records.get(objectId);
}
+ @Override
+ public List<CatalogRecord> records(RecordKind kind) {
+ return records.values().stream().filter(record -> record.kind() == kind && !record.deleted()).toList();
+ }
+
+ @Override
+ public long recordBytes(CatalogRecord record) throws IOException {
+ Integer known = _records.get(record.objectId()) == record ? _recordBytes.get(record.objectId()) : null;
+ return known == null ? MetadataProtocol.encodeValue(record).length : known;
+ }
+
@Override
public CatalogRecord retainedRecord(ObjectVersion version) {
CatalogRecord current = records.get(version.objectId());
@@ -757,7 +841,7 @@ public final class MetadataCatalog {
legacy.journalFloor(), legacy.journal(), new MembershipState(_historyId).snapshot());
} else {
Snapshot decoded = MetadataProtocol.fromTree(detached, Snapshot.class);
- restored = decoded.formatVersion() == 8
+ restored = decoded.formatVersion() == 8 || decoded.formatVersion() == 9
? new Snapshot(SNAPSHOT_FORMAT_VERSION, decoded.clusterId(), decoded.historyId(),
decoded.lastAppliedIndex(), decoded.records(), decoded.clients(), decoded.journalPolicyVersion(),
decoded.journalFloor(), decoded.journal(), decoded.membership())
@@ -1196,7 +1280,8 @@ public final class MetadataCatalog {
private static int minimumSnapshotFormat(CatalogRecord record) {
return switch (record.kind()) {
case CLUSTER_CONFIG, SCHEMA, TABLE_CONFIG -> 1;
- case MEMBERSHIP_CONTROL, PARTICIPANT -> 8;
+ case MEMBERSHIP_CONTROL -> 8;
+ case PARTICIPANT -> record.value().path("capabilities").toString().contains("MINION_TASK") ? 10 : 8;
case OWNER_LAYOUT, OWNER_ASSIGNMENT -> 4;
case CONTROLLER_OPERATION -> {
JsonNode value = record.value();
@@ -1214,7 +1299,10 @@ public final class MetadataCatalog {
}
case OFFLINE_SEGMENT_CONTENT, OFFLINE_REPLICA_ASSIGNMENT -> 5;
case OFFLINE_INSTALLATION -> 6;
- case TASK_QUEUE, TASK, TASK_ATTEMPT -> 9;
+ case TASK_SCHEDULE, TASK_PUBLICATION -> 10;
+ case TASK_QUEUE -> 9;
+ case TASK, TASK_ATTEMPT -> record.value().at("/input/outputPolicy").isObject()
+ || "SUCCEEDED".equals(record.value().path("status").asText()) ? 10 : 9;
};
}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataProtocol.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataProtocol.java
index 385e2d08a8e..437fdbca6e9 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataProtocol.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/MetadataProtocol.java
@@ -83,7 +83,7 @@ public final class MetadataProtocol {
public enum RecordKind {
CLUSTER_CONFIG, SCHEMA, TABLE_CONFIG, MEMBERSHIP_CONTROL, PARTICIPANT, OWNER_LAYOUT, OWNER_ASSIGNMENT,
CONTROLLER_OPERATION, OFFLINE_SEGMENT_CONTENT, OFFLINE_REPLICA_ASSIGNMENT, OFFLINE_INSTALLATION,
- TASK_QUEUE, TASK, TASK_ATTEMPT
+ TASK_QUEUE, TASK, TASK_ATTEMPT, TASK_SCHEDULE, TASK_PUBLICATION
}
public enum Action {
@@ -384,10 +384,11 @@ public final class MetadataProtocol {
@JsonSubTypes.Type(value = GetStreamTip.class, name = "GET_STREAM_TIP"),
@JsonSubTypes.Type(value = GetRevision.class, name = "GET_REVISION"),
@JsonSubTypes.Type(value = InstallationSessionQuery.class, name = "INSTALLATION_SESSION"),
+ @JsonSubTypes.Type(value = TaskSessionQuery.class, name = "TASK_SESSION"),
@JsonSubTypes.Type(value = GetInstallationWork.class, name = "GET_INSTALLATION_WORK"),
@JsonSubTypes.Type(value = MembershipQuery.class, name = "MEMBERSHIP")})
public sealed interface Query permits GetRecord, GetOutcome, GetStreamTip, GetRevision, MembershipQuery,
- InstallationSessionQuery, GetInstallationWork {
+ InstallationSessionQuery, TaskSessionQuery, GetInstallationWork {
UUID historyId();
}
@@ -424,6 +425,13 @@ public final class MetadataProtocol {
}
}
+ public record TaskSessionQuery(MembershipProtocol.ParticipantToken minion, GetOutcome request) implements Query {
+ @Override
+ public UUID historyId() {
+ return request == null ? null : request.historyId();
+ }
+ }
+
/// One coherent, authorized target capture. It neither lists other work nor exposes another server's target.
public record GetInstallationWork(UUID historyId, MembershipProtocol.ParticipantToken server, UUID operationId)
implements Query {
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java
index 7719e982428..1a851638e9c 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java
@@ -261,7 +261,7 @@ final class OfflineSegmentCatalog {
"Offline catalog references an absent, mismatched or future dependency");
}
- private static void validateContent(Content content) {
+ static void validateContent(Content content) {
require(content != null && content.tableObjectId() != null && content.segmentObjectId() != null,
Status.INVALID_REQUEST, "Offline content requires complete table and segment identities");
tableName(content.tableName());
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalog.java
index a67c84fa683..3578db27dda 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalog.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalog.java
@@ -20,12 +20,14 @@ package org.apache.pinot.coordination.ratis.metadata;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ControlState;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
@@ -33,7 +35,9 @@ import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.AttemptState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.AttemptStatus;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Cancel;
@@ -41,28 +45,38 @@ import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Checkpoint;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Claim;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Complete;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.ControllerOperation;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.CreateSchedule;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.FireSchedule;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Input;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Progress;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.PublicationState;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.PublishOfflineOutputs;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueKey;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Retry;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.ScheduleState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.SetQueuePaused;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Submit;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskStatus;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkerOperation;
+import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
-/// Deterministic task execution staging over one committed view. No scheduler, plugin I/O or output publication runs
+/// Deterministic scheduling and task publication staging over one committed view. No plugin or filesystem I/O runs
/// here. Every returned record is committed atomically with its durable request outcome by [MetadataCatalog].
final class TaskCatalog {
private TaskCatalog() {
}
- interface View extends OwnershipCatalog.View {
+ interface View extends OfflineSegmentCatalog.View {
CatalogRecord record(UUID objectId);
+ List<CatalogRecord> records(RecordKind kind);
+
+ long recordBytes(CatalogRecord record) throws IOException;
+
default CatalogRecord retainedRecord(ObjectVersion version) {
CatalogRecord current = record(version.objectId());
return current != null && current.version() == version.version() ? current : null;
@@ -76,6 +90,8 @@ final class TaskCatalog {
case SetQueuePaused pause -> pause(pause, view, index);
case Cancel cancel -> cancel(cancel, view, index);
case Retry retry -> retry(retry, view, index);
+ case CreateSchedule create -> createSchedule(create, view, index);
+ case FireSchedule fire -> fireSchedule(fire, view, index);
};
}
@@ -87,6 +103,7 @@ final class TaskCatalog {
case Claim claim -> claim(claim, minion, view, index);
case Checkpoint checkpoint -> checkpoint(checkpoint, minion, view, index);
case Complete complete -> complete(complete, minion, view, index);
+ case PublishOfflineOutputs publish -> publish(publish, minion, view, index);
};
}
@@ -104,6 +121,10 @@ final class TaskCatalog {
}
private static List<CatalogRecord> submit(Submit request, View view, long index) throws IOException {
+ return submit(request, view, index, true);
+ }
+
+ private static List<CatalogRecord> submit(Submit request, View view, long index, boolean admit) throws IOException {
require(request.taskId() != null, Status.INVALID_REQUEST, "Task identity is required");
validateInput(request.input());
QueueKey key = request.input().queue();
@@ -114,6 +135,7 @@ final class TaskCatalog {
ParticipantState assigned = view.participant(request.input().minionId());
require(assigned != null && assigned.participant().role() == Role.MINION, Status.NOT_FOUND,
"Task assignment requires a known minion identity");
+ validateTaskCapability(request.input(), assigned);
String name = TaskProtocol.taskName(request.taskId());
require(view.record(RecordKind.TASK, name) == null
&& view.record(TaskProtocol.objectId(view.historyId(), RecordKind.TASK, name)) == null, Status.CONFLICT,
@@ -139,6 +161,9 @@ final class TaskCatalog {
}
records.add(taskRecord(new TaskState(view.historyId(), request.taskId(), request.input(), captured,
request.owner(), TaskStatus.READY, 0), 1, index));
+ if (admit) {
+ admitWork(request.input(), records, view);
+ }
return records;
}
@@ -159,7 +184,8 @@ final class TaskCatalog {
CatalogRecord previous = currentTask(request.taskId(), request.expectedVersion(), view);
TaskState task = TaskProtocol.decodeTask(previous.value());
OwnershipCatalog.validateTableOwner(request.owner(), task.input().queue().tableName(), view);
- require(task.status() != TaskStatus.CANCELLED && task.status() != TaskStatus.FATAL_FAILED, Status.CONFLICT,
+ require(task.status() != TaskStatus.CANCELLED && task.status() != TaskStatus.FATAL_FAILED
+ && task.status() != TaskStatus.SUCCEEDED, Status.CONFLICT,
"Task is already cancelled or fatally failed");
List<CatalogRecord> records = new ArrayList<>();
if (task.status() == TaskStatus.RUNNING) {
@@ -195,6 +221,7 @@ final class TaskCatalog {
}
records.add(taskRecord(new TaskState(task.historyId(), task.taskId(), task.input(), task.dependencies(),
request.owner(), TaskStatus.READY, task.attempt()), nextVersion(previous), index));
+ admitWork(task.input(), records, view);
return records;
}
@@ -212,6 +239,13 @@ final class TaskCatalog {
require(queue != null && !TaskProtocol.decodeQueue(queue.value()).paused(), Status.NOT_READY,
"Task queue is absent or paused");
currentDependencies(task.dependencies(), view);
+ validateTaskCapability(task.input(), participant);
+ for (CatalogRecord other : view.records(RecordKind.TASK)) {
+ TaskState active = TaskProtocol.decodeTask(other.value());
+ if (active.status() == TaskStatus.RUNNING && active.input().minionId().equals(minion.participantId())) {
+ throw new CoordinationRejection(Status.NOT_READY, "Minion already has an active task attempt");
+ }
+ }
long number = task.attempt() + 1;
String name = TaskProtocol.attemptName(task.taskId(), number);
require(view.record(RecordKind.TASK_ATTEMPT, name) == null
@@ -223,7 +257,10 @@ final class TaskCatalog {
require(participantRecord != null, Status.NOT_READY, "Minion registration record is absent");
AttemptState attempt = new AttemptState(task.historyId(), task.taskId(), number, participant,
version(view, participantRecord), index, AttemptStatus.RUNNING, new Progress(0, ""));
- return List.of(taskRecord(running, nextVersion(previous), index), attemptRecord(attempt, 1, index));
+ List<CatalogRecord> records =
+ List.of(taskRecord(running, nextVersion(previous), index), attemptRecord(attempt, 1, index));
+ admitWork(task.input(), records, view);
+ return records;
}
private static List<CatalogRecord> checkpoint(Checkpoint request, ParticipantToken minion, View view, long index)
@@ -253,17 +290,274 @@ final class TaskCatalog {
require(progress.completedUnits() >= state.progress().completedUnits(), Status.CONFLICT,
"Task progress cannot move backwards");
currentDependencies(task.dependencies(), view);
+ validateTaskCapability(task.input(), view.participant(minion.participantId()));
+ require(task.input().outputPolicy() == null || result != AttemptStatus.EXECUTION_FINISHED,
+ Status.INVALID_REQUEST, "Output-producing tasks must publish outputs before successful completion");
TaskStatus taskStatus = switch (result) {
case RUNNING -> TaskStatus.RUNNING;
case EXECUTION_FINISHED -> TaskStatus.EXECUTION_FINISHED;
case RETRYABLE_FAILED -> TaskStatus.RETRYABLE_FAILED;
case FATAL_FAILED -> TaskStatus.FATAL_FAILED;
- case FENCED -> throw new CoordinationRejection(Status.INVALID_REQUEST, "Worker cannot fence task attempts");
+ case FENCED, SUCCEEDED -> throw new CoordinationRejection(Status.INVALID_REQUEST,
+ "Worker cannot directly fence or certify successful task attempts");
};
- return List.of(taskRecord(new TaskState(task.historyId(), task.taskId(), task.input(), task.dependencies(),
- task.controller(), taskStatus, task.attempt()), nextVersion(previous), index),
+ List<CatalogRecord> records = List.of(taskRecord(new TaskState(task.historyId(), task.taskId(), task.input(),
+ task.dependencies(), task.controller(), taskStatus, task.attempt()), nextVersion(previous), index),
attemptRecord(new AttemptState(state.historyId(), state.taskId(), state.attempt(), state.participantAtClaim(),
state.participantVersion(), state.claimedIndex(), result, progress), nextVersion(attempt), index));
+ if (result == AttemptStatus.RUNNING) {
+ admitWork(task.input(), records, view);
+ }
+ return records;
+ }
+
+ private static List<CatalogRecord> createSchedule(CreateSchedule request, View view, long index) throws IOException {
+ require(request.scheduleId() != null, Status.INVALID_REQUEST, "Schedule identity is required");
+ validateScheduleBounds(request.firstRound(), request.intervalRounds(), request.maxOccurrences());
+ require(view.control() != null && view.control().policy() != null, Status.NOT_READY,
+ "Schedule requires committed membership policy");
+ UUID firstTask = TaskProtocol.occurrenceTaskId(view.historyId(), request.scheduleId(), 1);
+ CatalogRecord existing = view.record(RecordKind.TASK_SCHEDULE, request.scheduleId().toString());
+ if (existing != null) {
+ ScheduleState state = TaskProtocol.decodeSchedule(existing.value());
+ OwnershipCatalog.validateTableOwner(request.owner(), state.input().queue().tableName(), view);
+ require(state.input().equals(request.input()) && state.firstRound() == request.firstRound()
+ && state.intervalRounds() == request.intervalRounds() && state.maxOccurrences() == request.maxOccurrences()
+ && sameDependencies(state.dependencies(), request.dependencies()),
+ Status.CONFLICT, "Schedule identity already belongs to a different immutable configuration");
+ currentDependencies(state.dependencies(), view);
+ return List.of();
+ }
+ List<CatalogRecord> staged = submit(new Submit(request.owner(), firstTask, request.input(), request.dependencies()),
+ view, index, false);
+ TaskState captured = TaskProtocol.decodeTask(staged.getLast().value());
+ List<CatalogRecord> records = new ArrayList<>(staged.subList(0, staged.size() - 1));
+ ScheduleState state = new ScheduleState(view.historyId(), request.scheduleId(), request.input(),
+ captured.dependencies(), request.owner(), request.firstRound(), request.intervalRounds(),
+ request.maxOccurrences(), 0, view.control().policy().version());
+ records.add(scheduleRecord(state, 1, index));
+ return records;
+ }
+
+ private static boolean sameDependencies(List<ObjectVersion> captured, List<ObjectVersion> declared) {
+ if (declared == null || declared.size() > TaskProtocol.MAX_DEPENDENCIES - 2) {
+ return false;
+ }
+ Set<ObjectVersion> expected = new HashSet<>(declared);
+ expected.add(captured.get(0));
+ expected.add(captured.get(1));
+ return expected.equals(new HashSet<>(captured));
+ }
+
+ private static List<CatalogRecord> fireSchedule(FireSchedule request, View view, long index) throws IOException {
+ require(request.scheduleId() != null && request.expectedVersion() > 0, Status.INVALID_REQUEST,
+ "Schedule identity and version are required");
+ CatalogRecord previous = view.record(RecordKind.TASK_SCHEDULE, request.scheduleId().toString());
+ require(previous != null && previous.version() == request.expectedVersion(), Status.CONFLICT,
+ "Schedule is absent or its version changed");
+ ScheduleState state = TaskProtocol.decodeSchedule(previous.value());
+ OwnershipCatalog.validateTableOwner(request.owner(), state.input().queue().tableName(), view);
+ currentDependencies(state.dependencies(), view);
+ ControlState control = view.control();
+ require(control != null && control.policy() != null && control.policy().version() == state.policyVersion(),
+ Status.FENCED, "Schedule membership timing policy changed; create a new schedule generation");
+ require(state.firedOccurrences() < state.maxOccurrences(), Status.CONFLICT, "Schedule occurrence bound reached");
+ CatalogRecord queue = view.record(RecordKind.TASK_QUEUE, TaskProtocol.queueName(state.input().queue()));
+ require(queue != null && !TaskProtocol.decodeQueue(queue.value()).paused(), Status.NOT_READY,
+ "Task queue is absent or paused");
+ long due = state.firstRound() + state.intervalRounds() * state.firedOccurrences();
+ require(control.round() >= due, Status.NOT_READY, "Schedule occurrence is not due");
+ long occurrence = state.firedOccurrences() + 1;
+ UUID taskId = TaskProtocol.occurrenceTaskId(view.historyId(), state.scheduleId(), occurrence);
+ require(view.record(RecordKind.TASK, taskId.toString()) == null, Status.CONFLICT,
+ "Schedule occurrence already has a task");
+ List<CatalogRecord> records = List.of(scheduleRecord(new ScheduleState(state.historyId(), state.scheduleId(),
+ state.input(),
+ state.dependencies(), request.owner(), state.firstRound(), state.intervalRounds(), state.maxOccurrences(),
+ occurrence, state.policyVersion()), nextVersion(previous), index),
+ taskRecord(new TaskState(state.historyId(), taskId, state.input(), state.dependencies(), request.owner(),
+ TaskStatus.READY, 0), 1, index));
+ admitWork(state.input(), records, view);
+ return records;
+ }
+
+ private static List<CatalogRecord> publish(PublishOfflineOutputs request, ParticipantToken minion, View view,
+ long index) throws IOException {
+ validateProgress(request.progress());
+ CatalogRecord previous = currentTask(request.taskId(), request.expectedVersion(), view);
+ TaskState task = TaskProtocol.decodeTask(previous.value());
+ require(task.status() == TaskStatus.RUNNING && task.attempt() == request.attempt(), Status.FENCED,
+ "Publication requires the current running task attempt");
+ CatalogRecord attemptRecord = attempt(task, view);
+ AttemptState attempt = TaskProtocol.decodeAttempt(attemptRecord.value());
+ require(attempt.status() == AttemptStatus.RUNNING && attempt.minion().equals(minion), Status.FENCED,
+ "Publication belongs to another minion session");
+ require(request.progress().completedUnits() >= attempt.progress().completedUnits(), Status.CONFLICT,
+ "Task progress cannot move backwards");
+ currentDependencies(task.dependencies(), view);
+ currentTable(task.input().queue(), view);
+ validateTaskCapability(task.input(), view.participant(minion.participantId()));
+ validateOutputs(task, attempt.minion(), request.outputs());
+ require(view.record(RecordKind.TASK_PUBLICATION, task.taskId().toString()) == null, Status.CONFLICT,
+ "Task output publication already exists");
+ List<CatalogRecord> records = new ArrayList<>();
+ for (Content output : request.outputs()) {
+ String name = OfflineSegmentProtocol.contentName(output.tableObjectId(), output.segmentName());
+ require(view.record(RecordKind.OFFLINE_SEGMENT_CONTENT, name) == null
+ && view.record(output.segmentObjectId()) == null, Status.CONFLICT,
+ "Task output content cannot overwrite or reuse a segment identity");
+ records.add(new CatalogRecord(RecordKind.OFFLINE_SEGMENT_CONTENT, name, output.segmentObjectId(), 1,
+ MetadataProtocol.FORMAT_VERSION, index, false, MetadataProtocol.toTree(output)));
+ }
+ PublicationState publication = new PublicationState(task.historyId(), task.taskId(), task.attempt(), minion,
+ request.outputs(), index);
+ records.add(publicationRecord(publication, index));
+ records.add(taskRecord(new TaskState(task.historyId(), task.taskId(), task.input(), task.dependencies(),
+ task.controller(), TaskStatus.SUCCEEDED, task.attempt()), nextVersion(previous), index));
+ records.add(attemptRecord(new AttemptState(attempt.historyId(), attempt.taskId(), attempt.attempt(),
+ attempt.participantAtClaim(), attempt.participantVersion(), attempt.claimedIndex(), AttemptStatus.SUCCEEDED,
+ request.progress()), nextVersion(attemptRecord), index));
+ return records;
+ }
+
+ /// Admission counts the complete candidate work image, sharing queue/input records across tasks. The reserved
+ /// slot and bytes cover a subsequent claim plus maximum progress/registration payloads, so workers can drain it.
+ /// Scheduling a configuration does not consume work capacity; only firing an occurrence does.
+ private static void admitWork(Input input, List<CatalogRecord> staged, View view) throws IOException {
+ if (input.outputPolicy() == null) {
+ return;
+ }
+ Map<UUID, CatalogRecord> candidates = new LinkedHashMap<>();
+ for (CatalogRecord record : view.records(RecordKind.TASK)) {
+ candidates.put(record.objectId(), record);
+ }
+ Map<UUID, CatalogRecord> replacements = new HashMap<>();
+ Map<String, CatalogRecord> queues = new HashMap<>();
+ for (CatalogRecord record : staged) {
+ replacements.put(record.objectId(), record);
+ if (record.kind() == RecordKind.TASK) {
+ candidates.put(record.objectId(), record);
+ } else if (record.kind() == RecordKind.TASK_QUEUE) {
+ queues.put(record.name(), record);
+ }
+ }
+ Set<UUID> selected = new HashSet<>();
+ long bytes = AdmissionReserve.BYTES;
+ int running = 0;
+ for (CatalogRecord record : candidates.values()) {
+ TaskState task = TaskProtocol.decodeTask(record.value());
+ if (!task.input().minionId().equals(input.minionId())
+ || (task.status() != TaskStatus.READY && task.status() != TaskStatus.RUNNING)) {
+ continue;
+ }
+ bytes = admitRecord(record, selected, bytes, view);
+ // Normalize mutable scalar widths, so a claim/checkpoint does not consume admission headroom twice.
+ int longWidth = Long.toString(Long.MAX_VALUE).length();
+ bytes += longWidth - Long.toString(record.version()).length()
+ + longWidth - Long.toString(record.lastModifiedIndex()).length()
+ + Long.toString(TaskProtocol.MAX_ATTEMPTS).length() - Long.toString(task.attempt()).length()
+ + TaskStatus.RUNNING.name().length() - task.status().name().length();
+ String queueName = TaskProtocol.queueName(task.input().queue());
+ CatalogRecord queue = queues.getOrDefault(queueName, view.record(RecordKind.TASK_QUEUE, queueName));
+ bytes = admitRecord(queue, selected, bytes, view);
+ for (ObjectVersion dependency : task.dependencies()) {
+ bytes = admitRecord(view.record(dependency.objectId()), selected, bytes, view);
+ }
+ if (task.status() == TaskStatus.RUNNING) {
+ UUID attemptId = TaskProtocol.objectId(task.historyId(), RecordKind.TASK_ATTEMPT,
+ TaskProtocol.attemptName(task.taskId(), task.attempt()));
+ CatalogRecord attempt = replacements.getOrDefault(attemptId, view.record(attemptId));
+ require(attempt != null, Status.NOT_READY, "Task admission attempt is unavailable");
+ require(++running <= 1, Status.NOT_READY, "Minion has multiple active attempts requiring recovery");
+ // The globally reserved slot already accounts for this exact running attempt and any legal progress.
+ require(view.recordBytes(attempt) <= AdmissionReserve.ATTEMPT_BYTES, Status.NOT_READY,
+ "Task attempt exceeds its reserved work budget");
+ }
+ require(selected.size() < TaskProtocol.MAX_SNAPSHOT_RECORDS && bytes <= TaskProtocol.MAX_SNAPSHOT_BYTES,
+ Status.NOT_READY, "Minion actionable work is full; finish or cancel existing work before admitting more");
+ }
+ }
+
+ private static long admitRecord(CatalogRecord record, Set<UUID> selected, long bytes, View view) throws IOException {
+ require(record != null, Status.NOT_READY, "Task admission dependency is unavailable");
+ if (selected.add(record.objectId())) {
+ bytes += view.recordBytes(record) + 1;
+ }
+ require(selected.size() < TaskProtocol.MAX_SNAPSHOT_RECORDS && bytes <= TaskProtocol.MAX_SNAPSHOT_BYTES,
+ Status.NOT_READY, "Minion actionable work exceeds its bounded projection budget");
+ return bytes;
+ }
+
+ /// Lazy initialization avoids any dependency on mapper initialization order. The serialized reservation contains
+ /// the maximum string lengths at full JSON escape width, widest counters and escaped progress detail.
+ /// Synthetic strings are size bounds only; they are never accepted as participant metadata.
+ /// One header and one future attempt suffice because claims enforce one active attempt per minion.
+ private static final class AdmissionReserve {
+ private static final long[] SIZES = sizes();
+ private static final long BYTES = SIZES[0] + SIZES[1] + 1;
+ private static final long ATTEMPT_BYTES = SIZES[1];
+
+ private static long[] sizes() {
+ UUID id = new UUID(0, 0);
+ String identifier = "\u0000".repeat(MembershipProtocol.MAX_IDENTIFIER_LENGTH);
+ Map<String, String> endpoints = new HashMap<>();
+ for (int ordinal = 0; ordinal < MembershipProtocol.MAX_ENDPOINTS; ordinal++) {
+ String prefix = Integer.toString(ordinal);
+ endpoints.put(prefix + identifier.substring(prefix.length()),
+ "\u0000".repeat(MembershipProtocol.MAX_ENDPOINT_VALUE_LENGTH));
+ }
+ ParticipantToken token = new ParticipantToken(id, identifier, Role.MINION, id, Long.MAX_VALUE);
+ ParticipantState participant = new ParticipantState(token, endpoints, Set.of(Capability.MINION_TASK),
+ Long.MAX_VALUE, false, false, false, new Revision(id, Long.MAX_VALUE), Long.MAX_VALUE, Long.MAX_VALUE);
+ MembershipProtocol.ControlToken control =
+ new MembershipProtocol.ControlToken(id, Long.MAX_VALUE, Long.MAX_VALUE, identifier, id);
+ MembershipProtocol.Policy policy = new MembershipProtocol.Policy(Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE,
+ Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE);
+ ControlState state = new ControlState(control, Long.MAX_VALUE, policy, Long.MAX_VALUE, Long.MAX_VALUE,
+ Long.MAX_VALUE);
+ TaskProtocol.WorkSnapshot header = new TaskProtocol.WorkSnapshot(Status.OK, new Revision(id, Long.MAX_VALUE),
+ "", participant, List.of(), state);
+ AttemptState attempt = new AttemptState(id, id, TaskProtocol.MAX_ATTEMPTS, participant,
+ new ObjectVersion(id, id, Long.MAX_VALUE), Long.MAX_VALUE, AttemptStatus.RUNNING,
+ new Progress(Long.MAX_VALUE, "\u0000".repeat(TaskProtocol.MAX_DETAIL_LENGTH)));
+ try {
+ return new long[]{MetadataProtocol.encodeValue(header).length,
+ MetadataProtocol.encodeValue(attemptRecord(attempt, Long.MAX_VALUE, Long.MAX_VALUE)).length};
+ } catch (IOException e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
+ }
+
+ private static void validateTaskCapability(Input input, ParticipantState participant) {
+ require(input.outputPolicy() == null || (participant != null
+ && participant.capabilities().contains(Capability.MINION_TASK)), Status.FENCED,
+ "Output-producing tasks require the MINION_TASK capability");
+ }
+
+ private static void validateOutputs(TaskState task, ParticipantToken minion, List<Content> outputs) {
+ require(task.input().outputPolicy() != null && outputs != null && !outputs.isEmpty()
+ && outputs.size() <= task.input().outputPolicy().maxOutputs(), Status.INVALID_REQUEST,
+ "Publication lacks an output grant or exceeds its output bound");
+ for (int ordinal = 0; ordinal < outputs.size(); ordinal++) {
+ Content output = outputs.get(ordinal);
+ OfflineSegmentCatalog.validateContent(output);
+ require(output.tableName().equals(task.input().queue().tableName())
+ && output.tableObjectId().equals(task.input().queue().tableObjectId())
+ && output.segmentName().equals(TaskProtocol.outputSegmentName(task.taskId(), task.attempt(), ordinal))
+ && output.segmentObjectId().equals(TaskProtocol.outputObjectId(task.historyId(), task.taskId(),
+ task.attempt(), ordinal))
+ && output.downloadUri().equals(TaskProtocol.outputUri(task.input().outputPolicy(), task.historyId(),
+ task.taskId(), task.attempt(), minion.incarnation(), ordinal, output.contentSha256())),
+ Status.INVALID_REQUEST, "Output identity or artifact URI is outside this task attempt's grant");
+ }
+ }
+
+ private static void validateScheduleBounds(long firstRound, long intervalRounds, long occurrences) {
+ require(firstRound >= 0 && intervalRounds >= 0 && occurrences > 0 && occurrences <= TaskProtocol.MAX_OCCURRENCES
+ && (intervalRounds > 0 || occurrences == 1)
+ && (intervalRounds == 0 || intervalRounds <= (Long.MAX_VALUE - firstRound) / occurrences),
+ Status.INVALID_REQUEST, "Invalid or overflowing bounded schedule");
}
private static CatalogRecord currentTask(UUID taskId, long expectedVersion, View view) {
@@ -319,6 +613,16 @@ final class TaskCatalog {
"Task attempt limit is outside the supported bounds");
require(input.configs() != null && input.configs().size() <= TaskProtocol.MAX_CONFIGS, Status.INVALID_REQUEST,
"Task configuration bound exceeded");
+ if (input.outputPolicy() != null) {
+ try {
+ TaskProtocol.outputRoot(input.outputPolicy());
+ } catch (IllegalArgumentException e) {
+ throw new CoordinationRejection(Status.INVALID_REQUEST, e.getMessage());
+ }
+ require(input.queue().taskType().equals(TaskProtocol.NATIVE_SEGMENT_GENERATION_TASK)
+ && TableNameBuilder.getTableTypeFromTableName(input.queue().tableName()) == TableType.OFFLINE,
+ Status.INVALID_REQUEST, "Output grant requires the supported native OFFLINE task type");
+ }
input.configs().forEach((key, value) -> {
OwnershipCatalog.identifier(key, 256);
require(value != null && value.length() <= 16_384, Status.INVALID_REQUEST,
@@ -358,6 +662,33 @@ final class TaskCatalog {
&& TaskProtocol.objectId(history, record.kind(), record.name()).equals(record.objectId()),
Status.INVALID_REQUEST, "Invalid task record envelope");
switch (record.kind()) {
+ case TASK_SCHEDULE -> {
+ ScheduleState schedule = TaskProtocol.decodeSchedule(record.value());
+ require(history.equals(schedule.historyId()) && schedule.scheduleId() != null
+ && schedule.scheduleId().toString().equals(record.name()), Status.INVALID_REQUEST,
+ "Schedule identity differs");
+ validateInput(schedule.input());
+ validateScheduleBounds(schedule.firstRound(), schedule.intervalRounds(), schedule.maxOccurrences());
+ require(schedule.firedOccurrences() >= 0 && schedule.firedOccurrences() <= schedule.maxOccurrences()
+ && record.version() == schedule.firedOccurrences() + 1 && schedule.policyVersion() > 0,
+ Status.INVALID_REQUEST, "Schedule occurrence watermark or policy is invalid");
+ validateDependencies(schedule.dependencies(), schedule.input().queue(), history);
+ OwnershipCatalog.validateStoredTableOwnerShape(schedule.controller(), schedule.input().queue().tableName(),
+ history);
+ }
+ case TASK_PUBLICATION -> {
+ PublicationState publication = TaskProtocol.decodePublication(record.value());
+ require(history.equals(publication.historyId()) && publication.taskId() != null
+ && publication.taskId().toString().equals(record.name()) && record.version() == 1
+ && publication.committedIndex() == record.lastModifiedIndex() && publication.attempt() > 0
+ && publication.attempt() <= TaskProtocol.MAX_ATTEMPTS && publication.outputs() != null
+ && !publication.outputs().isEmpty() && publication.outputs().size() <= TaskProtocol.MAX_OUTPUTS,
+ Status.INVALID_REQUEST, "Invalid immutable task publication");
+ validateMinionToken(publication.minion(), history);
+ for (Content output : publication.outputs()) {
+ OfflineSegmentCatalog.validateContent(output);
+ }
+ }
case TASK_QUEUE -> {
QueueState queue = TaskProtocol.decodeQueue(record.value());
require(history.equals(queue.historyId()), Status.HISTORY_MISMATCH, "Task queue history differs");
@@ -377,16 +708,7 @@ final class TaskCatalog {
require(task.status() != null && task.attempt() >= 0 && task.attempt() <= task.input().maxAttempts()
&& (task.attempt() > 0 || task.status() == TaskStatus.READY || task.status() == TaskStatus.CANCELLED),
Status.INVALID_REQUEST, "Invalid task state or attempt");
- require(task.dependencies() != null && task.dependencies().size() >= 2
- && task.dependencies().size() <= TaskProtocol.MAX_DEPENDENCIES, Status.INVALID_REQUEST,
- "Task dependency bound exceeded");
- Set<UUID> identities = new HashSet<>();
- for (ObjectVersion dependency : task.dependencies()) {
- validateVersion(dependency, history);
- require(identities.add(dependency.objectId()), Status.INVALID_REQUEST, "Duplicate task dependency");
- }
- require(identities.contains(task.input().queue().tableObjectId()), Status.INVALID_REQUEST,
- "Task omits its target table dependency");
+ validateDependencies(task.dependencies(), task.input().queue(), history);
}
case TASK_ATTEMPT -> {
AttemptState attempt = TaskProtocol.decodeAttempt(record.value());
@@ -411,6 +733,18 @@ final class TaskCatalog {
}
}
+ private static void validateDependencies(List<ObjectVersion> dependencies, QueueKey queue, UUID history) {
+ require(dependencies != null && dependencies.size() >= 2 && dependencies.size() <= TaskProtocol.MAX_DEPENDENCIES,
+ Status.INVALID_REQUEST, "Task dependency bound exceeded");
+ Set<UUID> identities = new HashSet<>();
+ for (ObjectVersion dependency : dependencies) {
+ validateVersion(dependency, history);
+ require(identities.add(dependency.objectId()), Status.INVALID_REQUEST, "Duplicate task dependency");
+ }
+ require(identities.contains(queue.tableObjectId()), Status.INVALID_REQUEST,
+ "Task omits its target table dependency");
+ }
+
/// Validate current and historical journal records without restoring historical participant authority.
static void validateReferences(CatalogRecord record, View view) throws IOException {
switch (record.kind()) {
@@ -419,6 +753,8 @@ final class TaskCatalog {
storedTable(state.queue(), view);
OwnershipCatalog.validateStoredTableOwner(state.controller(), state.queue().tableName(), view);
}
+ case TASK_SCHEDULE -> validateScheduleReferences(record, view);
+ case TASK_PUBLICATION -> validatePublicationReferences(record, view);
case TASK -> validateTaskReferences(record, view);
case TASK_ATTEMPT -> validateAttemptReferences(record, view);
default -> throw new CoordinationRejection(Status.INVALID_REQUEST, "Unsupported task reference kind");
@@ -438,6 +774,9 @@ final class TaskCatalog {
require(queue != null && !queue.deleted()
&& TaskProtocol.decodeQueue(queue.value()).queue().equals(state.input().queue()), Status.INVALID_REQUEST,
"Task queue is absent or differs");
+ CatalogRecord publication = view.record(RecordKind.TASK_PUBLICATION, state.taskId().toString());
+ require((current.status() == TaskStatus.SUCCEEDED) == (publication != null), Status.INVALID_REQUEST,
+ "Task success and committed publication disagree");
boolean schemaFound = false;
for (ObjectVersion dependency : state.dependencies()) {
CatalogRecord referenced = view.record(dependency.objectId());
@@ -465,6 +804,65 @@ final class TaskCatalog {
}
}
+ private static void validateScheduleReferences(CatalogRecord record, View view) throws IOException {
+ ScheduleState state = TaskProtocol.decodeSchedule(record.value());
+ storedTable(state.input().queue(), view);
+ OwnershipCatalog.validateStoredTableOwner(state.controller(), state.input().queue().tableName(), view);
+ CatalogRecord currentRecord = view.record(record.objectId());
+ ScheduleState current = TaskProtocol.decodeSchedule(currentRecord.value());
+ require(current.input().equals(state.input()) && current.dependencies().equals(state.dependencies())
+ && current.firstRound() == state.firstRound() && current.intervalRounds() == state.intervalRounds()
+ && current.maxOccurrences() == state.maxOccurrences() && current.policyVersion() == state.policyVersion()
+ && current.firedOccurrences() >= state.firedOccurrences(), Status.INVALID_REQUEST,
+ "Schedule configuration or occurrence watermark changed");
+ boolean schemaFound = false;
+ for (ObjectVersion dependency : state.dependencies()) {
+ CatalogRecord referenced = view.record(dependency.objectId());
+ require(referenced != null && supportedDependency(referenced.kind())
+ && referenced.version() >= dependency.version()
+ && (referenced.version() > dependency.version()
+ || referenced.lastModifiedIndex() <= record.lastModifiedIndex()),
+ Status.INVALID_REQUEST, "Schedule references an absent or future input");
+ schemaFound |= referenced.kind() == RecordKind.SCHEMA
+ && referenced.name().equals(TableNameBuilder.extractRawTableName(state.input().queue().tableName()));
+ }
+ require(schemaFound, Status.INVALID_REQUEST, "Schedule omits its table schema dependency");
+ if (record.equals(currentRecord)) {
+ for (long occurrence = 1; occurrence <= state.firedOccurrences(); occurrence++) {
+ UUID taskId = TaskProtocol.occurrenceTaskId(state.historyId(), state.scheduleId(), occurrence);
+ CatalogRecord taskRecord = view.record(RecordKind.TASK, taskId.toString());
+ require(taskRecord != null, Status.INVALID_REQUEST, "Schedule watermark skips a missing task occurrence");
+ TaskState task = TaskProtocol.decodeTask(taskRecord.value());
+ require(task.input().equals(state.input()) && task.dependencies().equals(state.dependencies()),
+ Status.INVALID_REQUEST, "Scheduled task differs from immutable schedule inputs");
+ }
+ }
+ }
+
+ private static void validatePublicationReferences(CatalogRecord record, View view) throws IOException {
+ PublicationState publication = TaskProtocol.decodePublication(record.value());
+ CatalogRecord taskRecord = view.record(RecordKind.TASK, publication.taskId().toString());
+ require(taskRecord != null && taskRecord.lastModifiedIndex() == publication.committedIndex(),
+ Status.INVALID_REQUEST, "Publication has no atomically committed task");
+ TaskState task = TaskProtocol.decodeTask(taskRecord.value());
+ require(task.status() == TaskStatus.SUCCEEDED && task.attempt() == publication.attempt(), Status.INVALID_REQUEST,
+ "Publication task or successful attempt differs");
+ CatalogRecord attemptRecord = attempt(task, view);
+ AttemptState attempt = TaskProtocol.decodeAttempt(attemptRecord.value());
+ require(attempt.status() == AttemptStatus.SUCCEEDED && attempt.minion().equals(publication.minion())
+ && attemptRecord.lastModifiedIndex() == publication.committedIndex(), Status.INVALID_REQUEST,
+ "Publication has no atomically successful exact minion attempt");
+ validateTaskCapability(task.input(), attempt.participantAtClaim());
+ validateOutputs(task, publication.minion(), publication.outputs());
+ for (Content output : publication.outputs()) {
+ CatalogRecord content = view.record(output.segmentObjectId());
+ require(content != null && content.kind() == RecordKind.OFFLINE_SEGMENT_CONTENT
+ && content.lastModifiedIndex() == publication.committedIndex()
+ && output.equals(OfflineSegmentProtocol.decodeContent(content.value())), Status.INVALID_REQUEST,
+ "Published output manifest does not match atomically committed immutable content");
+ }
+ }
+
private static boolean statusMatches(TaskStatus task, AttemptStatus attempt) {
return switch (task) {
case READY -> attempt == AttemptStatus.RETRYABLE_FAILED || attempt == AttemptStatus.FENCED;
@@ -472,7 +870,8 @@ final class TaskCatalog {
case EXECUTION_FINISHED -> attempt == AttemptStatus.EXECUTION_FINISHED;
case RETRYABLE_FAILED -> attempt == AttemptStatus.RETRYABLE_FAILED;
case FATAL_FAILED -> attempt == AttemptStatus.FATAL_FAILED;
- case CANCELLED -> attempt != AttemptStatus.RUNNING;
+ case CANCELLED -> attempt != AttemptStatus.RUNNING && attempt != AttemptStatus.SUCCEEDED;
+ case SUCCEEDED -> attempt == AttemptStatus.SUCCEEDED;
};
}
@@ -515,7 +914,8 @@ final class TaskCatalog {
}
static boolean taskKind(RecordKind kind) {
- return kind == RecordKind.TASK_QUEUE || kind == RecordKind.TASK || kind == RecordKind.TASK_ATTEMPT;
+ return kind == RecordKind.TASK_QUEUE || kind == RecordKind.TASK || kind == RecordKind.TASK_ATTEMPT
+ || kind == RecordKind.TASK_SCHEDULE || kind == RecordKind.TASK_PUBLICATION;
}
private static ObjectVersion version(View view, CatalogRecord record) {
@@ -528,6 +928,14 @@ final class TaskCatalog {
return previous == null ? 1 : previous.version() + 1;
}
+ private static CatalogRecord scheduleRecord(ScheduleState state, long version, long index) {
+ return record(RecordKind.TASK_SCHEDULE, state.scheduleId().toString(), state.historyId(), state, version, index);
+ }
+
+ private static CatalogRecord publicationRecord(PublicationState state, long index) {
+ return record(RecordKind.TASK_PUBLICATION, state.taskId().toString(), state.historyId(), state, 1, index);
+ }
+
private static CatalogRecord queueRecord(QueueState state, long version, long index) {
return record(RecordKind.TASK_QUEUE, TaskProtocol.queueName(state.queue()), state.historyId(), state, version,
index);
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskProtocol.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskProtocol.java
index b3544615ad2..243c4d3c442 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskProtocol.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskProtocol.java
@@ -18,50 +18,134 @@
*/
package org.apache.pinot.coordination.ratis.metadata;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
+import java.net.URI;
import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.coordination.NativeTaskOutputIdentity;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ControlState;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.OwnerToken;
-/// Internal durable task execution ledger. This protocol does not publish task outputs or enable native plugins.
+/// Durable task scheduling, execution and immutable OFFLINE output publication. Legacy plugins are not enabled.
/// Participant tokens identify sessions; a future authenticated transport must derive them from its caller identity.
public final class TaskProtocol {
public static final int MAX_ATTEMPTS = 16;
public static final int MAX_DEPENDENCIES = 32;
public static final int MAX_CONFIGS = 128;
public static final int MAX_DETAIL_LENGTH = 1024;
+ public static final int MAX_OUTPUTS = 16;
+ public static final int MAX_OCCURRENCES = 1024;
+ public static final int MAX_SNAPSHOT_RECORDS = 128;
+ public static final int MIN_SNAPSHOT_BYTES = 1024;
+ public static final int MAX_SNAPSHOT_BYTES = 2 * 1024 * 1024;
+ public static final String NATIVE_SEGMENT_GENERATION_TASK = "NativeSegmentGenerationTask";
private TaskProtocol() {
}
/// EXECUTION_FINISHED records worker execution only. It does not certify output publication or task success.
public enum TaskStatus {
- READY, RUNNING, EXECUTION_FINISHED, RETRYABLE_FAILED, FATAL_FAILED, CANCELLED
+ READY, RUNNING, EXECUTION_FINISHED, RETRYABLE_FAILED, FATAL_FAILED, CANCELLED, SUCCEEDED
}
/// FENCED revokes further reports. It does not assert that a cancelled or unavailable worker stopped executing.
public enum AttemptStatus {
- RUNNING, EXECUTION_FINISHED, RETRYABLE_FAILED, FATAL_FAILED, FENCED
+ RUNNING, EXECUTION_FINISHED, RETRYABLE_FAILED, FATAL_FAILED, FENCED, SUCCEEDED
}
public record QueueKey(String tableName, UUID tableObjectId, String taskType) {
}
- /// Explicit minion identity is a bounded initial assignment policy, not plugin capability discovery or scheduling.
- public record Input(QueueKey queue, String minionId, Map<String, String> configs, int maxAttempts) {
+ /// Explicit assignment remains independent of plugin discovery. Only an explicit typed output policy grants
+ /// publication authority; old inputs remain readable and cannot publish outputs.
+ public record Input(QueueKey queue, String minionId, Map<String, String> configs, int maxAttempts,
+ @Nullable @JsonInclude(JsonInclude.Include.NON_NULL) OutputPolicy outputPolicy) {
public Input {
configs = configs == null ? null : Map.copyOf(configs);
}
+
+ public Input(QueueKey queue, String minionId, Map<String, String> configs, int maxAttempts) {
+ this(queue, minionId, configs, maxAttempts, null);
+ }
+
+ /// Preserve replay of pre-publication commands without weakening missing-field validation for other records.
+ @JsonCreator(mode = JsonCreator.Mode.DELEGATING)
+ public static Input fromJson(JsonNode value) throws IOException {
+ Set<String> fields = Set.of("queue", "minionId", "configs", "maxAttempts", "outputPolicy");
+ if (!value.isObject() || !value.has("queue") || !value.has("minionId") || !value.has("configs")
+ || (!value.path("minionId").isTextual() && !value.path("minionId").isNull())
+ || (!value.path("configs").isObject() && !value.path("configs").isNull())
+ || !value.path("maxAttempts").isIntegralNumber()
+ || !value.path("maxAttempts").canConvertToInt()) {
+ throw new IOException("Invalid task input fields");
+ }
+ Map<String, String> configs = new HashMap<>();
+ for (var entry : value.properties()) {
+ if (!fields.contains(entry.getKey())) {
+ throw new IOException("Unsupported task input field");
+ }
+ }
+ for (var entry : value.path("configs").properties()) {
+ if (!entry.getValue().isTextual()) {
+ throw new IOException("Task config must be a string");
+ }
+ configs.put(entry.getKey(), entry.getValue().textValue());
+ }
+ JsonNode policy = value.get("outputPolicy");
+ return new Input(MetadataProtocol.fromTree(value.get("queue"), QueueKey.class),
+ value.get("minionId").textValue(), value.get("configs").isNull() ? null : configs,
+ value.get("maxAttempts").intValue(),
+ policy == null || policy.isNull() ? null : MetadataProtocol.fromTree(policy, OutputPolicy.class));
+ }
+ }
+
+ /// Controller-approved immutable artifact namespace. Local workers separately enforce their configured I/O roots.
+ public record OutputPolicy(String rootUri, int maxOutputs) {
+ }
+
+ /// A schedule identity is its immutable configuration generation. One FireSchedule creates at most one occurrence;
+ /// finite CATCH_UP scheduling uses committed membership rounds and never wall-clock values supplied by a worker.
+ public record ScheduleState(UUID historyId, UUID scheduleId, Input input, List<ObjectVersion> dependencies,
+ OwnerToken controller, long firstRound, long intervalRounds, long maxOccurrences,
+ long firedOccurrences, long policyVersion) {
+ public ScheduleState {
+ dependencies = dependencies == null ? null : List.copyOf(dependencies);
+ }
+ }
+
+ /// The manifest and successful terminal task/attempt records share one committed transaction with these contents.
+ /// It certifies catalog publication, not installed replicas, serving eligibility or physical file reclamation.
+ public record PublicationState(UUID historyId, UUID taskId, long attempt, ParticipantToken minion,
+ List<Content> outputs, long committedIndex) {
+ public PublicationState {
+ outputs = outputs == null ? null : List.copyOf(outputs);
+ }
+ }
+
+ public record WorkSnapshot(Status status, Revision revision, String detail, ParticipantState participant,
+ List<CatalogRecord> records, ControlState control) {
+ public WorkSnapshot {
+ records = List.copyOf(records);
+ }
}
public record QueueState(UUID historyId, QueueKey queue, boolean paused, OwnerToken controller) {
@@ -90,8 +174,11 @@ public final class TaskProtocol {
@JsonSubTypes({@JsonSubTypes.Type(value = Submit.class, name = "SUBMIT"),
@JsonSubTypes.Type(value = SetQueuePaused.class, name = "SET_QUEUE_PAUSED"),
@JsonSubTypes.Type(value = Cancel.class, name = "CANCEL"),
- @JsonSubTypes.Type(value = Retry.class, name = "RETRY")})
- public sealed interface ControllerOperation permits Submit, SetQueuePaused, Cancel, Retry {
+ @JsonSubTypes.Type(value = Retry.class, name = "RETRY"),
+ @JsonSubTypes.Type(value = CreateSchedule.class, name = "CREATE_SCHEDULE"),
+ @JsonSubTypes.Type(value = FireSchedule.class, name = "FIRE_SCHEDULE")})
+ public sealed interface ControllerOperation
+ permits Submit, SetQueuePaused, Cancel, Retry, CreateSchedule, FireSchedule {
OwnerToken owner();
}
@@ -103,6 +190,18 @@ public final class TaskProtocol {
}
}
+ public record CreateSchedule(OwnerToken owner, UUID scheduleId, Input input, List<ObjectVersion> dependencies,
+ long firstRound, long intervalRounds, long maxOccurrences)
+ implements ControllerOperation {
+ public CreateSchedule {
+ dependencies = dependencies == null ? null : List.copyOf(dependencies);
+ }
+ }
+
+ public record FireSchedule(OwnerToken owner, UUID scheduleId, long expectedVersion)
+ implements ControllerOperation {
+ }
+
/// Version zero creates the queue; existing queues require their exact current version.
public record SetQueuePaused(OwnerToken owner, QueueKey queue, long expectedVersion, boolean paused)
implements ControllerOperation {
@@ -120,8 +219,9 @@ public final class TaskProtocol {
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "operation")
@JsonSubTypes({@JsonSubTypes.Type(value = Claim.class, name = "CLAIM"),
@JsonSubTypes.Type(value = Checkpoint.class, name = "CHECKPOINT"),
- @JsonSubTypes.Type(value = Complete.class, name = "COMPLETE")})
- public sealed interface WorkerOperation permits Claim, Checkpoint, Complete {
+ @JsonSubTypes.Type(value = Complete.class, name = "COMPLETE"),
+ @JsonSubTypes.Type(value = PublishOfflineOutputs.class, name = "PUBLISH_OFFLINE_OUTPUTS")})
+ public sealed interface WorkerOperation permits Claim, Checkpoint, Complete, PublishOfflineOutputs {
UUID taskId();
long expectedVersion();
@@ -139,6 +239,51 @@ public final class TaskProtocol {
implements WorkerOperation {
}
+ public record PublishOfflineOutputs(UUID taskId, long expectedVersion, long attempt, List<Content> outputs,
+ Progress progress) implements WorkerOperation {
+ public PublishOfflineOutputs {
+ outputs = outputs == null ? null : List.copyOf(outputs);
+ }
+ }
+
+ public static ScheduleState decodeSchedule(JsonNode value) throws IOException {
+ return MetadataProtocol.fromTree(value, ScheduleState.class);
+ }
+
+ public static PublicationState decodePublication(JsonNode value) throws IOException {
+ return MetadataProtocol.fromTree(value, PublicationState.class);
+ }
+
+ public static UUID occurrenceTaskId(UUID historyId, UUID scheduleId, long occurrence) {
+ if (historyId == null || scheduleId == null || occurrence <= 0 || occurrence > MAX_OCCURRENCES) {
+ throw new IllegalArgumentException("Invalid scheduled occurrence identity");
+ }
+ return UUID.nameUUIDFromBytes((historyId + "\0TASK_OCCURRENCE\0" + scheduleId + "\0" + occurrence)
+ .getBytes(StandardCharsets.UTF_8));
+ }
+
+ public static String outputSegmentName(UUID taskId, long attempt, int ordinal) {
+ return NativeTaskOutputIdentity.outputSegmentName(taskId, attempt, ordinal);
+ }
+
+ public static UUID outputObjectId(UUID historyId, UUID taskId, long attempt, int ordinal) {
+ return NativeTaskOutputIdentity.outputObjectId(historyId, taskId, attempt, ordinal);
+ }
+
+ public static String outputUri(OutputPolicy policy, UUID historyId, UUID taskId, long attempt, UUID incarnation,
+ int ordinal, String sha256) {
+ outputRoot(policy);
+ return NativeTaskOutputIdentity.outputUri(policy.rootUri(), historyId, taskId, attempt, incarnation, ordinal,
+ sha256);
+ }
+
+ public static URI outputRoot(OutputPolicy policy) {
+ if (policy == null || policy.maxOutputs() <= 0 || policy.maxOutputs() > MAX_OUTPUTS) {
+ throw new IllegalArgumentException("Invalid output publication policy");
+ }
+ return NativeTaskOutputIdentity.outputRoot(policy.rootUri());
+ }
+
public static QueueState decodeQueue(JsonNode value) throws IOException {
return MetadataProtocol.fromTree(value, QueueState.class);
}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java
index c218dcf5efd..8acd3ba10d0 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcServer.java
@@ -47,6 +47,7 @@ import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import javax.net.ssl.SSLSession;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.rpc.MinionTaskServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.ParticipantMembershipServiceGrpc;
import org.apache.pinot.coordination.ratis.rpc.ServerInstallationServiceGrpc;
@@ -61,6 +62,7 @@ public final class MetadataGrpcServer implements AutoCloseable {
private final ThreadPoolExecutor _executor;
private final ThreadPoolExecutor _membershipExecutor;
private final ThreadPoolExecutor _installationExecutor;
+ private final ThreadPoolExecutor _taskExecutor;
private final ThreadPoolExecutor _dispatchExecutor;
private final ScheduledThreadPoolExecutor _deadlines;
@@ -83,6 +85,8 @@ public final class MetadataGrpcServer implements AutoCloseable {
Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
_installationExecutor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(16),
Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
+ _taskExecutor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(16),
+ Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
// gRPC performs method lookup before callExecutor selection. Keep that short dispatch off both handler pools;
// its SerializingExecutor switches to the selected pool before starting the application call.
_dispatchExecutor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
@@ -102,6 +106,9 @@ public final class MetadataGrpcServer implements AutoCloseable {
if (ParticipantMembershipServiceGrpc.SERVICE_NAME.equals(service)) {
return _membershipExecutor;
}
+ if (MinionTaskServiceGrpc.SERVICE_NAME.equals(service)) {
+ return _taskExecutor;
+ }
return ServerInstallationServiceGrpc.SERVICE_NAME.equals(service) ? _installationExecutor : _executor;
}
}).maxInboundMessageSize(MetadataProtocol.MAX_COMMAND_BYTES).maxInboundMetadataSize(8192)
@@ -116,6 +123,11 @@ public final class MetadataGrpcServer implements AutoCloseable {
new InstallationGrpcService(access, installationAccess, participantPolicy, audit, limits.maxRpcMillis()),
new PeerInterceptor(limits.maxRpcMillis(), audit)));
}
+ if (access instanceof TaskAccess taskAccess) {
+ builder.addService(ServerInterceptors.intercept(
+ new TaskGrpcService(access, taskAccess, participantPolicy, audit, limits.maxRpcMillis()),
+ new PeerInterceptor(limits.maxRpcMillis(), audit)));
+ }
}
server = builder.build().start();
_server = server;
@@ -126,6 +138,7 @@ public final class MetadataGrpcServer implements AutoCloseable {
_executor.shutdownNow();
_membershipExecutor.shutdownNow();
_installationExecutor.shutdownNow();
+ _taskExecutor.shutdownNow();
_dispatchExecutor.shutdownNow();
_deadlines.shutdownNow();
throw e;
@@ -143,6 +156,7 @@ public final class MetadataGrpcServer implements AutoCloseable {
_executor.shutdownNow();
_membershipExecutor.shutdownNow();
_installationExecutor.shutdownNow();
+ _taskExecutor.shutdownNow();
_dispatchExecutor.shutdownNow();
_deadlines.shutdownNow();
try {
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataRpcCodec.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataRpcCodec.java
index 140701328c9..c1c82b873dc 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataRpcCodec.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataRpcCodec.java
@@ -184,6 +184,9 @@ final class MetadataRpcCodec {
case TASK_QUEUE -> value.setTaskQueue(TaskRpcCodec.queue(TaskProtocol.decodeQueue(record.value())));
case TASK -> value.setTask(TaskRpcCodec.task(TaskProtocol.decodeTask(record.value())));
case TASK_ATTEMPT -> value.setTaskAttempt(TaskRpcCodec.attempt(TaskProtocol.decodeAttempt(record.value())));
+ case TASK_SCHEDULE -> value.setTaskSchedule(TaskRpcCodec.schedule(TaskProtocol.decodeSchedule(record.value())));
+ case TASK_PUBLICATION -> value.setTaskPublication(
+ TaskRpcCodec.publication(TaskProtocol.decodePublication(record.value())));
default -> throw new IOException("Unsupported catalog record kind");
}
builder.setValue(value);
@@ -207,6 +210,8 @@ final class MetadataRpcCodec {
case TASK_QUEUE -> RecordKind.TASK_QUEUE;
case TASK -> RecordKind.TASK;
case TASK_ATTEMPT -> RecordKind.TASK_ATTEMPT;
+ case TASK_SCHEDULE -> RecordKind.TASK_SCHEDULE;
+ case TASK_PUBLICATION -> RecordKind.TASK_PUBLICATION;
default -> throw new IllegalArgumentException("Unknown catalog record kind");
};
}
@@ -337,6 +342,16 @@ final class MetadataRpcCodec {
return typedValue(TaskRpcCodec.attempt(value.getTaskAttempt()));
}
break;
+ case TASK_SCHEDULE:
+ if (kind == RecordKind.TASK_SCHEDULE) {
+ return typedValue(TaskRpcCodec.schedule(value.getTaskSchedule()));
+ }
+ break;
+ case TASK_PUBLICATION:
+ if (kind == RecordKind.TASK_PUBLICATION) {
+ return typedValue(TaskRpcCodec.publication(value.getTaskPublication()));
+ }
+ break;
default:
break;
}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/ParticipantAccessPolicy.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/ParticipantAccessPolicy.java
index c996466c5b0..b03797316d3 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/ParticipantAccessPolicy.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/ParticipantAccessPolicy.java
@@ -26,7 +26,7 @@ import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
/// Current certificate-to-participant authority, separate from full controller metadata access. Implementations must
/// be thread-safe, nonblocking and return null to deny access. Every operation rechecks the current binding; a prior
-/// token cannot supply its own identity, role or capabilities. Minion membership is outside this implementation slice.
+/// token cannot supply its own identity, role or capabilities.
@FunctionalInterface
public interface ParticipantAccessPolicy {
@Nullable
@@ -37,9 +37,10 @@ public interface ParticipantAccessPolicy {
record ParticipantIdentity(String participantId, Role role, Set<Capability> capabilities) {
public ParticipantIdentity {
if (participantId == null || participantId.isBlank() || participantId.length() > 128 || role == null
- || role == Role.MINION || capabilities == null
+ || capabilities == null
|| (role != Role.BROKER && capabilities.contains(Capability.BROKER_QUERY))
- || (role != Role.SERVER && capabilities.contains(Capability.SERVER_QUERY))) {
+ || (role != Role.SERVER && capabilities.contains(Capability.SERVER_QUERY))
+ || (role != Role.MINION && capabilities.contains(Capability.MINION_TASK))) {
throw new IllegalArgumentException("Invalid supported participant identity, role or capabilities");
}
capabilities = Set.copyOf(capabilities);
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskAccess.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskAccess.java
new file mode 100644
index 00000000000..03846326da1
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskAccess.java
@@ -0,0 +1,32 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.coordination.ratis.transport;
+
+import java.io.IOException;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
+
+
+/// Coherent linearizable projection of one authenticated minion's task ledger and exact input dependencies.
+/// Implementations validate its current live session, include no other minion's task inputs and fail closed when
+/// the complete projection exceeds the supplied bounds. Pre-ready hydration does not authorize a task claim.
+public interface TaskAccess {
+ WorkSnapshot taskSnapshot(ParticipantToken minion, int maxRecords, int maxBytes, long timeoutMillis)
+ throws IOException;
+}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskGrpcService.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskGrpcService.java
new file mode 100644
index 00000000000..38c8c77852c
--- /dev/null
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskGrpcService.java
@@ -0,0 +1,165 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.coordination.ratis.transport;
+
+import com.google.protobuf.Message;
+import io.grpc.Context;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MinionTaskServiceGrpc;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
+
+
+/// MINION-only task work and durable outcomes. A separate executor and admission budget protect membership
+/// renewals from task reads and reports. An RPC timeout never cancels an accepted durable command.
+final class TaskGrpcService extends MinionTaskServiceGrpc.MinionTaskServiceImplBase {
+ static final int MAX_REQUEST_BYTES = 64 * 1024;
+ private final MetadataAccess _metadata;
+ private final TaskAccess _tasks;
+ private final ParticipantAccessPolicy _policy;
+ private final MetadataAudit _audit;
+ private final long _maxRpcMillis;
+ private final Semaphore _calls = new Semaphore(2);
+
+ TaskGrpcService(MetadataAccess metadata, TaskAccess tasks, ParticipantAccessPolicy policy,
+ MetadataAudit audit, long maxRpcMillis) {
+ _metadata = Objects.requireNonNull(metadata);
+ _tasks = Objects.requireNonNull(tasks);
+ _policy = Objects.requireNonNull(policy);
+ _audit = Objects.requireNonNull(audit);
+ _maxRpcMillis = maxRpcMillis;
+ }
+
+ @Override
+ public void execute(MetadataRpc.TaskCommandRequest request,
+ StreamObserver<MetadataRpc.CommandResult> observer) {
+ respond("TaskExecute", request, observer, identity -> MetadataRpcCodec.result(
+ _metadata.execute(TaskRpcCodec.command(request, identity), remainingMillis())));
+ }
+
+ @Override
+ public void outcome(MetadataRpc.TaskOutcomeRequest request,
+ StreamObserver<MetadataRpc.CommandResult> observer) {
+ respond("TaskOutcome", request, observer, identity -> MetadataRpcCodec.result(
+ _metadata.query(TaskRpcCodec.outcome(request, identity), remainingMillis())));
+ }
+
+ @Override
+ public void work(MetadataRpc.TaskWorkRequest request, StreamObserver<MetadataRpc.TaskWorkResult> observer) {
+ respond("TaskWork", request, observer, identity -> {
+ ParticipantToken minion = TaskRpcCodec.work(request, identity);
+ WorkSnapshot snapshot = _tasks.taskSnapshot(minion, request.getMaxRecords(), request.getMaxBytes(),
+ remainingMillis());
+ if (snapshot.status() == MetadataProtocol.Status.OK && (snapshot.participant() == null
+ || !snapshot.participant().participant().equals(minion) || snapshot.control() == null)) {
+ throw Status.INTERNAL.withDescription("Task work identity differs").asRuntimeException();
+ }
+ MetadataRpc.TaskWorkResult response = TaskRpcCodec.work(snapshot);
+ if (response.getSerializedSize() > request.getMaxBytes()) {
+ return MetadataRpc.TaskWorkResult.newBuilder().setStatus(MetadataRpc.ResultStatus.RESPONSE_TOO_LARGE)
+ .setRevision(response.getRevision()).setDetail("Task projection exceeds its wire byte bound").build();
+ }
+ return response;
+ });
+ }
+
+ private <T extends Message> void respond(String operation, Message request, StreamObserver<T> observer,
+ Handler<T> handler) {
+ ParticipantIdentity identity = null;
+ String outcome = "UNAVAILABLE";
+ boolean acquired = false;
+ try {
+ identity = authorize();
+ remainingMillis();
+ if (request.getSerializedSize() > MAX_REQUEST_BYTES || !_calls.tryAcquire()) {
+ throw Status.RESOURCE_EXHAUSTED.withDescription("Task request exceeds admission bounds")
+ .asRuntimeException();
+ }
+ acquired = true;
+ MetadataRpcCodec.rejectUnknown(request);
+ T response = handler.handle(identity);
+ remainingMillis();
+ if (!identity.equals(authorize())) {
+ throw Status.PERMISSION_DENIED.withDescription("Minion authorization changed during task request")
+ .asRuntimeException();
+ }
+ if (response.getSerializedSize() > MetadataProtocol.MAX_CHANGE_BYTES) {
+ throw Status.RESOURCE_EXHAUSTED.withDescription("Task response exceeds transport bounds")
+ .asRuntimeException();
+ }
+ outcome = response.getField(response.getDescriptorForType().findFieldByName("status")).toString();
+ observer.onNext(response);
+ observer.onCompleted();
+ } catch (StatusRuntimeException e) {
+ outcome = e.getStatus().getCode().name();
+ observer.onError(e);
+ } catch (IllegalArgumentException e) {
+ outcome = "INVALID_ARGUMENT";
+ observer.onError(Status.INVALID_ARGUMENT.withDescription("Invalid task request").asRuntimeException());
+ } catch (IOException e) {
+ outcome = "UNAVAILABLE";
+ observer.onError(Status.UNAVAILABLE.withDescription("Task unavailable; write outcome may be unknown")
+ .asRuntimeException());
+ } catch (RuntimeException e) {
+ outcome = "INTERNAL";
+ observer.onError(Status.INTERNAL.withDescription("Task request failed").asRuntimeException());
+ } finally {
+ if (acquired) {
+ _calls.release();
+ }
+ MetadataGrpcService.audit(_audit, identity == null ? "" : identity.participantId(), operation, outcome);
+ }
+ }
+
+ private ParticipantIdentity authorize() {
+ String fingerprint = MetadataGrpcServer.PEER_FINGERPRINT.get();
+ ParticipantIdentity identity = fingerprint == null ? null : _policy.authorize(fingerprint);
+ if (identity == null || identity.role() != Role.MINION
+ || !identity.capabilities().contains(Capability.MINION_TASK)) {
+ throw Status.PERMISSION_DENIED.withDescription("Minion task access is not authorized")
+ .asRuntimeException();
+ }
+ return identity;
+ }
+
+ private long remainingMillis() {
+ Context context = Context.current();
+ long remaining = context.getDeadline() == null ? _maxRpcMillis
+ : context.getDeadline().timeRemaining(TimeUnit.MILLISECONDS);
+ if (context.isCancelled() || remaining <= 0) {
+ throw Status.DEADLINE_EXCEEDED.withDescription("Task request deadline expired").asRuntimeException();
+ }
+ return Math.min(remaining, _maxRpcMillis);
+ }
+
+ private interface Handler<T> {
+ T handle(ParticipantIdentity identity) throws IOException;
+ }
+}
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodec.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodec.java
index 4ec148f27f0..f8f1a73c8f7 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodec.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodec.java
@@ -18,26 +18,149 @@
*/
package org.apache.pinot.coordination.ratis.transport;
+import java.io.IOException;
import java.util.UUID;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.AttemptState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.AttemptStatus;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Input;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.OutputPolicy;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Progress;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.PublicationState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueKey;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueState;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.ScheduleState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskStatus;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkSnapshot;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.WorkerOperation;
import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
-/// Typed controller read/feed projections for task ledgers. This codec grants no MINION identity or mutation API.
+/// Closed task commands and projections. Worker identities and durable request client IDs are derived from the
+/// authenticated MINION policy; payloads cannot select another participant or generic metadata mutation.
final class TaskRpcCodec {
private TaskRpcCodec() {
}
+ static MetadataProtocol.TaskSessionCommand command(MetadataRpc.TaskCommandRequest request,
+ ParticipantIdentity identity) throws IOException {
+ MetadataRpcCodec.rejectUnknown(request);
+ require(request.hasSession(), "Minion session required");
+ UUID history = UUID.fromString(request.getHistoryId());
+ ParticipantToken minion = minion(history, request.getSession(), identity);
+ String clientId = TaskProtocol.clientId(minion);
+ MetadataProtocol.Command command = switch (request.getCommandCase()) {
+ case REGISTER -> {
+ require(minion.incarnation().toString().equals(request.getRegister().getIncarnation()),
+ "Task request registration must use the participant incarnation");
+ yield new MetadataProtocol.RegisterClient(history, clientId, minion.incarnation(),
+ request.getRegister().getExpectedEpoch());
+ }
+ case ACKNOWLEDGE -> new MetadataProtocol.Acknowledge(history, clientId,
+ request.getAcknowledge().getClientEpoch(), request.getAcknowledge().getStreamId(),
+ request.getAcknowledge().getThroughSequence());
+ case OPERATION -> {
+ MetadataRpc.SequencedTaskOperation operation = request.getOperation();
+ yield new MetadataProtocol.NativeTaskCommand(history, clientId, operation.getClientEpoch(),
+ operation.getStreamId(), operation.getSequence(), minion, operation(operation));
+ }
+ default -> throw new IllegalArgumentException("Typed task command required");
+ };
+ MetadataProtocol.TaskSessionCommand guarded = new MetadataProtocol.TaskSessionCommand(minion, command);
+ MetadataProtocol.encode(guarded);
+ return guarded;
+ }
+
+ private static WorkerOperation operation(MetadataRpc.SequencedTaskOperation request) {
+ return switch (request.getOperationCase()) {
+ case CLAIM -> {
+ MetadataRpc.TaskClaim claim = request.getClaim();
+ yield new TaskProtocol.Claim(UUID.fromString(claim.getTaskId()), claim.getExpectedVersion());
+ }
+ case CHECKPOINT -> {
+ MetadataRpc.TaskCheckpoint checkpoint = request.getCheckpoint();
+ require(checkpoint.hasProgress(), "Task progress required");
+ yield new TaskProtocol.Checkpoint(UUID.fromString(checkpoint.getTaskId()), checkpoint.getExpectedVersion(),
+ checkpoint.getAttempt(), progress(checkpoint.getProgress()));
+ }
+ case COMPLETE -> {
+ MetadataRpc.TaskComplete complete = request.getComplete();
+ require(complete.hasProgress(), "Task completion progress required");
+ yield new TaskProtocol.Complete(UUID.fromString(complete.getTaskId()), complete.getExpectedVersion(),
+ complete.getAttempt(), AttemptStatus.valueOf(complete.getResult().name()),
+ progress(complete.getProgress()));
+ }
+ case PUBLISH -> {
+ MetadataRpc.TaskPublishOfflineOutputs publish = request.getPublish();
+ require(publish.hasProgress() && publish.getOutputsCount() > 0
+ && publish.getOutputsCount() <= TaskProtocol.MAX_OUTPUTS,
+ "Bounded task outputs and progress required");
+ yield new TaskProtocol.PublishOfflineOutputs(UUID.fromString(publish.getTaskId()), publish.getExpectedVersion(),
+ publish.getAttempt(), publish.getOutputsList().stream().map(OfflineSegmentRpcCodec::content).toList(),
+ progress(publish.getProgress()));
+ }
+ default -> throw new IllegalArgumentException("Typed task operation required");
+ };
+ }
+
+ static MetadataProtocol.TaskSessionQuery outcome(MetadataRpc.TaskOutcomeRequest request,
+ ParticipantIdentity identity) {
+ MetadataRpcCodec.rejectUnknown(request);
+ require(request.hasSession() && request.hasOutcome(), "Minion session and outcome identity required");
+ UUID history = UUID.fromString(request.getHistoryId());
+ ParticipantToken minion = minion(history, request.getSession(), identity);
+ return new MetadataProtocol.TaskSessionQuery(minion, new MetadataProtocol.GetOutcome(history,
+ TaskProtocol.clientId(minion), request.getOutcome().getClientEpoch(), request.getOutcome().getStreamId(),
+ request.getOutcome().getSequence()));
+ }
+
+ static ParticipantToken work(MetadataRpc.TaskWorkRequest request, ParticipantIdentity identity) {
+ MetadataRpcCodec.rejectUnknown(request);
+ require(request.hasSession(), "Minion session required");
+ require(request.getMaxRecords() > 0 && request.getMaxRecords() <= TaskProtocol.MAX_SNAPSHOT_RECORDS
+ && request.getMaxBytes() >= TaskProtocol.MIN_SNAPSHOT_BYTES
+ && request.getMaxBytes() <= TaskProtocol.MAX_SNAPSHOT_BYTES, "Task projection bounds required");
+ return minion(UUID.fromString(request.getHistoryId()), request.getSession(), identity);
+ }
+
+ static MetadataRpc.TaskWorkResult work(WorkSnapshot snapshot) throws IOException {
+ MetadataRpc.TaskWorkResult.Builder result = MetadataRpc.TaskWorkResult.newBuilder()
+ .setStatus(MetadataRpc.ResultStatus.valueOf(snapshot.status().name())).setDetail(snapshot.detail())
+ .setRevision(MetadataRpc.Revision.newBuilder().setHistoryId(snapshot.revision().historyId().toString())
+ .setIndex(snapshot.revision().index()));
+ if (snapshot.participant() != null) {
+ result.setParticipant(MembershipRpcCodec.participant(snapshot.participant()));
+ }
+ if (snapshot.control() != null) {
+ result.setControl(MembershipRpcCodec.control(snapshot.control()));
+ }
+ for (MetadataProtocol.CatalogRecord record : snapshot.records()) {
+ result.addRecords(MetadataRpcCodec.record(record));
+ }
+ return result.build();
+ }
+
+ private static ParticipantToken minion(UUID history, MetadataRpc.ParticipantSession session,
+ ParticipantIdentity identity) {
+ require(identity.role() == Role.MINION && identity.capabilities().contains(Capability.MINION_TASK)
+ && session.getRegistrationEpoch() > 0, "Authorized MINION task session required");
+ return new ParticipantToken(history, identity.participantId(), Role.MINION,
+ UUID.fromString(session.getIncarnation()), session.getRegistrationEpoch());
+ }
+
+ private static Progress progress(MetadataRpc.TaskProgress progress) {
+ require(progress.getCompletedUnits() >= 0 && progress.getDetail().length() <= TaskProtocol.MAX_DETAIL_LENGTH,
+ "Invalid task progress");
+ return new Progress(progress.getCompletedUnits(), progress.getDetail());
+ }
+
static MetadataRpc.TaskQueueState queue(QueueState state) {
return MetadataRpc.TaskQueueState.newBuilder().setHistoryId(state.historyId().toString())
.setQueue(key(state.queue())).setPaused(state.paused())
@@ -101,6 +224,50 @@ final class TaskRpcCodec {
new Progress(state.getProgress().getCompletedUnits(), state.getProgress().getDetail()));
}
+ static MetadataRpc.TaskScheduleState schedule(ScheduleState state) {
+ return MetadataRpc.TaskScheduleState.newBuilder().setHistoryId(state.historyId().toString())
+ .setScheduleId(state.scheduleId().toString()).setInput(input(state.input()))
+ .addAllDependencies(state.dependencies().stream().map(TaskRpcCodec::version).toList())
+ .setController(ControllerRpcCodec.owner(state.controller())).setFirstRound(state.firstRound())
+ .setIntervalRounds(state.intervalRounds()).setMaxOccurrences(state.maxOccurrences())
+ .setFiredOccurrences(state.firedOccurrences()).setPolicyVersion(state.policyVersion()).build();
+ }
+
+ static ScheduleState schedule(MetadataRpc.TaskScheduleState state) {
+ MetadataRpcCodec.rejectUnknown(state);
+ require(state.hasInput() && state.hasController() && state.getDependenciesCount() >= 2
+ && state.getDependenciesCount() <= TaskProtocol.MAX_DEPENDENCIES && state.getFirstRound() >= 0
+ && state.getIntervalRounds() >= 0 && (state.getMaxOccurrences() == 1 || state.getIntervalRounds() > 0)
+ && state.getMaxOccurrences() > 0
+ && state.getMaxOccurrences() <= TaskProtocol.MAX_OCCURRENCES && state.getFiredOccurrences() >= 0
+ && state.getFiredOccurrences() <= state.getMaxOccurrences() && state.getPolicyVersion() > 0,
+ "Invalid task schedule projection");
+ return new ScheduleState(UUID.fromString(state.getHistoryId()), UUID.fromString(state.getScheduleId()),
+ input(state.getInput()), state.getDependenciesList().stream().map(TaskRpcCodec::version).toList(),
+ ControllerRpcCodec.owner(state.getController()), state.getFirstRound(), state.getIntervalRounds(),
+ state.getMaxOccurrences(), state.getFiredOccurrences(), state.getPolicyVersion());
+ }
+
+ static MetadataRpc.TaskPublicationState publication(PublicationState state) {
+ return MetadataRpc.TaskPublicationState.newBuilder().setHistoryId(state.historyId().toString())
+ .setTaskId(state.taskId().toString()).setAttempt(state.attempt())
+ .setMinion(MembershipRpcCodec.participant(state.minion()))
+ .addAllOutputs(state.outputs().stream().map(OfflineSegmentRpcCodec::content).toList())
+ .setCommittedIndex(state.committedIndex()).build();
+ }
+
+ static PublicationState publication(MetadataRpc.TaskPublicationState state) {
+ MetadataRpcCodec.rejectUnknown(state);
+ require(state.hasMinion() && state.getAttempt() > 0 && state.getAttempt() <= TaskProtocol.MAX_ATTEMPTS
+ && state.getOutputsCount() > 0 && state.getOutputsCount() <= TaskProtocol.MAX_OUTPUTS
+ && state.getCommittedIndex() > 0, "Invalid task publication projection");
+ ParticipantToken minion = MembershipRpcCodec.participant(state.getMinion());
+ require(minion.role() == Role.MINION, "Task publication requires a MINION session");
+ return new PublicationState(UUID.fromString(state.getHistoryId()), UUID.fromString(state.getTaskId()),
+ state.getAttempt(), minion, state.getOutputsList().stream().map(OfflineSegmentRpcCodec::content).toList(),
+ state.getCommittedIndex());
+ }
+
private static MetadataRpc.TaskQueueKey key(QueueKey key) {
return MetadataRpc.TaskQueueKey.newBuilder().setTableName(key.tableName())
.setTableObjectId(key.tableObjectId().toString()).setTaskType(key.taskType()).build();
@@ -113,8 +280,13 @@ final class TaskRpcCodec {
}
private static MetadataRpc.TaskInput input(Input input) {
- return MetadataRpc.TaskInput.newBuilder().setQueue(key(input.queue())).setMinionId(input.minionId())
- .putAllConfigs(input.configs()).setMaxAttempts(input.maxAttempts()).build();
+ MetadataRpc.TaskInput.Builder result = MetadataRpc.TaskInput.newBuilder().setQueue(key(input.queue()))
+ .setMinionId(input.minionId()).putAllConfigs(input.configs()).setMaxAttempts(input.maxAttempts());
+ if (input.outputPolicy() != null) {
+ result.setOutputPolicy(MetadataRpc.TaskOutputPolicy.newBuilder().setRootUri(input.outputPolicy().rootUri())
+ .setMaxOutputs(input.outputPolicy().maxOutputs()));
+ }
+ return result.build();
}
private static Input input(MetadataRpc.TaskInput input) {
@@ -124,7 +296,13 @@ final class TaskRpcCodec {
input.getConfigsMap().forEach((key, value) -> require(
!key.isBlank() && key.length() <= 256 && value.length() <= 16_384,
"Task configuration exceeds supported bounds"));
- return new Input(key(input.getQueue()), input.getMinionId(), input.getConfigsMap(), input.getMaxAttempts());
+ OutputPolicy output = null;
+ if (input.hasOutputPolicy()) {
+ require(!input.getOutputPolicy().getRootUri().isBlank() && input.getOutputPolicy().getMaxOutputs() > 0
+ && input.getOutputPolicy().getMaxOutputs() <= TaskProtocol.MAX_OUTPUTS, "Invalid task output policy");
+ output = new OutputPolicy(input.getOutputPolicy().getRootUri(), input.getOutputPolicy().getMaxOutputs());
+ }
+ return new Input(key(input.getQueue()), input.getMinionId(), input.getConfigsMap(), input.getMaxAttempts(), output);
}
private static MetadataRpc.ObjectVersion version(ObjectVersion version) {
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MembershipStateTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MembershipStateTest.java
index bf91df49c28..aa681efda5a 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MembershipStateTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MembershipStateTest.java
@@ -314,7 +314,7 @@ public class MembershipStateTest {
v2.remove("membership");
empty.restoreSnapshot(v2);
assertNull(empty.membershipControl().controlEpoch());
- assertEquals(empty.captureSnapshot().path("formatVersion").asInt(), 9);
+ assertEquals(empty.captureSnapshot().path("formatVersion").asInt(), 10);
}
@Test
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalogTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalogTest.java
index 6d21087cab7..8b7624f30ff 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalogTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataCatalogTest.java
@@ -497,7 +497,7 @@ public class MetadataCatalogTest {
assertEquals(catalog.captureSnapshot(), before, "A rejected old membership log cannot advance apply");
assertEquals(MembershipProtocol.decodeResponse(catalog.execute(current, 1, 2)).status(), Status.OK);
JsonNode initialized = catalog.captureSnapshot();
- assertEquals(initialized.path("formatVersion").asInt(), 9);
+ assertEquals(initialized.path("formatVersion").asInt(), 10);
for (int version = 3; version <= 7; version++) {
ObjectNode legacy = initialized.deepCopy();
legacy.put("formatVersion", version);
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataChangeFeedTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataChangeFeedTest.java
index addb9421731..f79b10fb164 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataChangeFeedTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/MetadataChangeFeedTest.java
@@ -240,7 +240,7 @@ public class MetadataChangeFeedTest {
}
MetadataCatalog restored = new MetadataCatalog(fixture._cluster, fixture._history);
restored.restoreSnapshot(prior);
- assertEquals(restored.captureSnapshot().path("formatVersion").asInt(), 9);
+ assertEquals(restored.captureSnapshot().path("formatVersion").asInt(), 10);
assertEquals(restored.captureImage(ALL), fixture._catalog.captureImage(ALL));
assertEquals(restored.captureSnapshot().path("journal"), current.path("journal"));
assertEquals(restored.captureSnapshot().path("journalFloor"), current.path("journalFloor"));
@@ -250,7 +250,7 @@ public class MetadataChangeFeedTest {
legacy.remove(List.of("journalPolicyVersion", "journalFloor", "journal", "membership"));
MetadataCatalog recovered = new MetadataCatalog(fixture._cluster, fixture._history);
recovered.restoreSnapshot(legacy);
- assertEquals(recovered.captureSnapshot().path("formatVersion").asInt(), 9);
+ assertEquals(recovered.captureSnapshot().path("formatVersion").asInt(), 10);
assertEquals(recovered.captureImage(ALL), fixture._catalog.captureImage(ALL));
FeedCursor before = new FeedCursor(fixture._history, 1, ALL.scopeId());
assertEquals(recovered.changes(ALL, before, 128, 128, MetadataProtocol.MAX_CHANGE_BYTES).status(),
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java
index 1a042a4359f..8d33ef9adb8 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java
@@ -21,6 +21,7 @@ package org.apache.pinot.coordination.ratis.metadata;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -30,12 +31,14 @@ import java.util.function.LongFunction;
import java.util.stream.Collectors;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ActivateControlEpoch;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.BeginDrain;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ConfigurePolicy;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Register;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.SetReady;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Tick;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Unregister;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Acknowledge;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Action;
@@ -47,6 +50,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipC
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Mutation;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.NativeTaskCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.OfflineSegmentCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.OwnershipCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RegisterClient;
@@ -56,6 +60,9 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.TaskSessionCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Transaction;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.PublishContent;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.AssignOwner;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.ConfigureLayout;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.OwnerToken;
@@ -64,8 +71,12 @@ import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Cancel;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Checkpoint;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Claim;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Complete;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.CreateSchedule;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.FireSchedule;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Input;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.OutputPolicy;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Progress;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.PublishOfflineOutputs;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueKey;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Retry;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.SetQueuePaused;
@@ -79,6 +90,7 @@ import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
@@ -247,7 +259,7 @@ public class TaskCatalogTest {
old.put("formatVersion", 8);
MetadataCatalog upgraded = new MetadataCatalog(f._cluster, f._history);
upgraded.restoreSnapshot(old);
- assertEquals(upgraded.captureSnapshot().path("formatVersion").asInt(), 9);
+ assertEquals(upgraded.captureSnapshot().path("formatVersion").asInt(), 10);
assertEquals(f.submit().status(), Status.OK);
assertEquals(f.worker(new Claim(f._task, f.version())).status(), Status.OK);
long claimIndex = f._index;
@@ -282,7 +294,200 @@ public class TaskCatalogTest {
assertEquals(upgraded.captureSnapshot(), current);
}
+ @Test
+ public void testOccurrenceWatermarkAndPublicationCommitAtomicallyAcrossRecovery() throws Exception {
+ Fixture f = new Fixture(true);
+ UUID scheduleId = UUID.randomUUID();
+ CreateSchedule create = new CreateSchedule(f._owner, scheduleId, f.input(), List.of(), 1, 1, 2);
+ assertEquals(f.task(create).status(), Status.OK);
+ assertEquals(f.task(create).status(), Status.OK);
+ assertEquals(f.task(new CreateSchedule(f._owner, scheduleId, f.input(), List.of(), 2, 1, 2)).status(),
+ Status.CONFLICT);
+ assertEquals(f.task(new FireSchedule(f._owner, scheduleId, 1)).status(), Status.NOT_READY);
+ var control = f._catalog.membershipControl();
+ assertEquals(f.member(new Tick(control.controlEpoch(), 0, control.policy().version())).status(), Status.OK);
+ assertEquals(f.task(new FireSchedule(f._owner, scheduleId, 1)).status(), Status.OK);
+ UUID taskId = TaskProtocol.occurrenceTaskId(f._history, scheduleId, 1);
+ assertEquals(f.worker(new Claim(taskId, 1)).status(), Status.OK);
+ Content output = f.output(taskId, 1);
+ long before = f._index;
+ assertEquals(f.worker(new PublishOfflineOutputs(taskId, 2, 1, List.of(output), new Progress(1, "published")))
+ .status(), Status.OK);
+ long published = f._lastWorkerResult.revision().index();
+ assertTrue(published > before);
+ for (CatalogRecord record : List.of(f.record(RecordKind.TASK, taskId.toString()),
+ f.record(RecordKind.TASK_ATTEMPT, TaskProtocol.attemptName(taskId, 1)),
+ f.record(RecordKind.TASK_PUBLICATION, taskId.toString()),
+ f.record(RecordKind.OFFLINE_SEGMENT_CONTENT, OfflineSegmentProtocol.contentName(f._table,
+ output.segmentName())))) {
+ assertEquals(record.lastModifiedIndex(), published);
+ }
+ assertEquals(TaskProtocol.decodeTask(f.record(RecordKind.TASK, taskId.toString()).value()).status(),
+ TaskStatus.SUCCEEDED);
+ f.restore();
+ assertEquals(f.apply(f._lastWorker), f._lastWorkerResult);
+ assertTrue(f._catalog.taskSnapshot(f._minion, 128, 100_000).records().isEmpty());
+ assertEquals(f.task(new Cancel(f._owner, taskId, 3, "too late")).status(), Status.CONFLICT);
+ assertEquals(f.task(new FireSchedule(f._owner, scheduleId, 1)).status(), Status.CONFLICT);
+ assertEquals(f.task(new FireSchedule(f._owner, scheduleId, 2)).status(), Status.NOT_READY);
+ assertEquals(f.member(new Tick(control.controlEpoch(), 1, control.policy().version())).status(), Status.OK);
+ assertEquals(f.task(new FireSchedule(f._owner, scheduleId, 2)).status(), Status.OK);
+ assertEquals(f.task(new FireSchedule(f._owner, scheduleId, 3)).status(), Status.CONFLICT);
+ f.restore();
+ ObjectNode corrupt = f._catalog.captureSnapshot().deepCopy();
+ for (JsonNode records : corrupt.findValues("records")) {
+ for (JsonNode record : records) {
+ if (record.path("kind").asText().equals(RecordKind.TASK_PUBLICATION.name())) {
+ ((ObjectNode) record.at("/value/outputs/0")).put("contentSha256", "b".repeat(64));
+ }
+ }
+ }
+ assertThrows(IOException.class, () -> new MetadataCatalog(f._cluster, f._history).restoreSnapshot(corrupt));
+ }
+
+ @Test
+ public void testPublicationRejectsUnownedArtifactsAndFencedAttemptsWithoutPartialEffects() throws Exception {
+ Fixture f = new Fixture(true);
+ assertEquals(f.submit().status(), Status.OK);
+ assertEquals(f.worker(new Claim(f._task, f.version())).status(), Status.OK);
+ Content output = f.output(f._task, 1);
+ Content wrongPath = new Content(output.tableName(), output.tableObjectId(), output.segmentName(),
+ output.segmentObjectId(), output.contentSha256(), "file:///outside/result.tar.gz", output.loader());
+ assertEquals(f.worker(new PublishOfflineOutputs(f._task, f.version(), 1, List.of(wrongPath),
+ new Progress(1, "bad path"))).status(), Status.INVALID_REQUEST);
+ assertEquals(f.state().status(), TaskStatus.RUNNING);
+ assertEquals(f.worker(new Complete(f._task, f.version(), 1, AttemptStatus.EXECUTION_FINISHED,
+ new Progress(1, "no publication"))).status(), Status.INVALID_REQUEST);
+ UUID second = UUID.randomUUID();
+ assertEquals(f.task(new Submit(f._owner, second, f.input(), List.of())).status(), Status.OK);
+ assertEquals(f.worker(new Claim(second, 1)).status(), Status.NOT_READY);
+ ParticipantToken old = f._minion;
+ assertEquals(f.member(new Unregister(old)).status(), Status.OK);
+ f._minion = f.registerMinion("minion", 1);
+ assertEquals(f.worker(new Claim(second, 1)).status(), Status.NOT_READY);
+ assertEquals(f.task(new Retry(f._owner, f._task, f.version())).status(), Status.OK);
+ assertEquals(f.worker(new Claim(f._task, f.version())).status(), Status.OK);
+ assertEquals(f.worker(new PublishOfflineOutputs(f._task, f.version(), 1, List.of(output),
+ new Progress(1, "stale attempt"))).status(), Status.FENCED);
+ assertEquals(f.task(new Cancel(f._owner, f._task, f.version(), "cancel")).status(), Status.OK);
+ assertEquals(f.worker(new PublishOfflineOutputs(f._task, f.version(), 2, List.of(f.output(f._task, 2)),
+ new Progress(1, "late"))).status(), Status.FENCED);
+ assertFalse(f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK_PUBLICATION,
+ RecordKind.OFFLINE_SEGMENT_CONTENT), "outputs"), 10, 100_000).records().iterator().hasNext());
+ f.restore();
+ }
+
+ @Test
+ public void testPublicationCollisionAndInputChangeCannotCertifySuccess() throws Exception {
+ Fixture f = new Fixture(true);
+ assertEquals(f.submit().status(), Status.OK);
+ assertEquals(f.worker(new Claim(f._task, f.version())).status(), Status.OK);
+ Content output = f.output(f._task, 1);
+ ObjectVersion table = f.objectVersion(f.record(RecordKind.TABLE_CONFIG, TABLE));
+ assertEquals(f.controller(sequence -> new OfflineSegmentCommand(f._history, "controller", 1, "controller",
+ sequence, new PublishContent(f._owner, table, output))).status(), Status.OK);
+ assertEquals(f.worker(new PublishOfflineOutputs(f._task, f.version(), 1, List.of(output),
+ new Progress(1, "collision"))).status(), Status.CONFLICT);
+ assertEquals(f.state().status(), TaskStatus.RUNNING);
+ assertEquals(f.mutate(new Mutation(Action.UPDATE, RecordKind.TABLE_CONFIG, TABLE, table,
+ MetadataProtocol.toTree(f.table()))).status(), Status.OK);
+ assertEquals(f.worker(new PublishOfflineOutputs(f._task, f.version(), 1, List.of(output),
+ new Progress(1, "changed input"))).status(), Status.CONFLICT);
+ assertEquals(f._catalog.taskSnapshot(f._minion, 1, 100_000).status(), Status.RESPONSE_TOO_LARGE);
+ assertEquals(f._catalog.taskSnapshot(f._minion, 128, 1024).status(), Status.RESPONSE_TOO_LARGE);
+ var image = f._catalog.taskSnapshot(f._minion, 128, 100_000);
+ assertEquals(image.status(), Status.OK);
+ assertTrue(image.records().stream().noneMatch(record -> record.kind() == RecordKind.PARTICIPANT));
+ f.restore();
+ }
+
+ @Test
+ public void testLegacyInputEncodingAndSnapshotRemainReplayable() throws Exception {
+ Fixture f = new Fixture();
+ TaskCommand command = new TaskCommand(f._history, "controller", 1, "controller", f._sequence + 1,
+ new Submit(f._owner, f._task, f.input(), List.of()));
+ byte[] legacy = MetadataProtocol.encode(command);
+ assertFalse(new String(legacy, StandardCharsets.UTF_8).contains("outputPolicy"));
+ Command decoded = MetadataProtocol.decodeCommand(legacy);
+ assertEquals(MetadataProtocol.encode(decoded), legacy);
+ assertEquals(MetadataProtocol.digest(decoded), MetadataProtocol.digest(command));
+ for (Input invalid : List.of(new Input(null, "minion", Map.of(), 2),
+ new Input(f.input().queue(), null, Map.of(), 2), new Input(f.input().queue(), "minion", null, 2))) {
+ TaskCommand rejected = new TaskCommand(f._history, "controller", 1, "controller", f._sequence + 1,
+ new Submit(f._owner, f._task, invalid, List.of()));
+ byte[] rejectedBytes = MetadataProtocol.encode(rejected);
+ assertEquals(MetadataProtocol.encode(MetadataProtocol.decodeCommand(rejectedBytes)), rejectedBytes);
+ assertEquals(f.task(new Submit(f._owner, f._task, invalid, List.of())).status(), Status.INVALID_REQUEST);
+ }
+ assertEquals(f.submit().status(), Status.OK);
+ ObjectNode old = f._catalog.captureSnapshot().deepCopy();
+ old.put("formatVersion", 9);
+ MetadataCatalog restored = new MetadataCatalog(f._cluster, f._history);
+ restored.restoreSnapshot(old);
+ assertEquals(restored.captureSnapshot().path("formatVersion").asInt(), 10);
+ }
+
+ @Test
+ public void testActionableAdmissionPreservesWatermarkAndReservesClaimCapacity() throws Exception {
+ for (boolean largeInput : List.of(false, true)) {
+ Fixture f = new Fixture(true);
+ Input input = f.input();
+ if (largeInput) {
+ input = new Input(input.queue(), input.minionId(),
+ Map.of("sourceUri", "file:///" + "a".repeat(16_000), "sourceSha256", "a".repeat(64)),
+ input.maxAttempts(), input.outputPolicy());
+ }
+ UUID schedule = UUID.randomUUID();
+ assertEquals(f.task(new CreateSchedule(f._owner, schedule, input, List.of(), 0, 0, 1)).status(), Status.OK);
+ int admitted = 0;
+ while (admitted < TaskProtocol.MAX_SNAPSHOT_RECORDS) {
+ UUID id = admitted == 0 ? f._task : UUID.randomUUID();
+ Result result = f.task(new Submit(f._owner, id, input, List.of()));
+ if (result.status() == Status.NOT_READY) {
+ break;
+ }
+ assertEquals(result.status(), Status.OK);
+ admitted++;
+ }
+ assertTrue(admitted > 0 && admitted < TaskProtocol.MAX_SNAPSHOT_RECORDS);
+ var image = f._catalog.taskSnapshot(f._minion, TaskProtocol.MAX_SNAPSHOT_RECORDS,
+ TaskProtocol.MAX_SNAPSHOT_BYTES);
+ assertEquals(image.status(), Status.OK);
+ if (largeInput) {
+ assertTrue(image.records().size() < TaskProtocol.MAX_SNAPSHOT_RECORDS - 1,
+ "Byte admission must reject before the record bound is exhausted");
+ }
+ // Configuration creation adds no actionable work, even while this minion is full.
+ assertEquals(f.task(new CreateSchedule(f._owner, UUID.randomUUID(), input, List.of(), 0, 0, 1)).status(),
+ Status.OK);
+ assertEquals(f.task(new FireSchedule(f._owner, schedule, 1)).status(), Status.NOT_READY);
+ CatalogRecord retained = f.record(RecordKind.TASK_SCHEDULE, schedule.toString());
+ assertEquals(retained.version(), 1L);
+ assertEquals(TaskProtocol.decodeSchedule(retained.value()).firedOccurrences(), 0L);
+ assertEquals(f.worker(new Claim(f._task, 1)).status(), Status.OK);
+ assertEquals(f.worker(new Checkpoint(f._task, f.version(), 1,
+ new Progress(1, "\u0000".repeat(TaskProtocol.MAX_DETAIL_LENGTH)))).status(), Status.OK);
+ assertEquals(f._catalog.taskSnapshot(f._minion, TaskProtocol.MAX_SNAPSHOT_RECORDS,
+ TaskProtocol.MAX_SNAPSHOT_BYTES).status(), Status.OK);
+ // A failed first attempt can return to READY at the full queue; historical attempt records do not consume
+ // actionable capacity, and the second claim still has its reserved slot.
+ assertEquals(f.worker(new Complete(f._task, f.version(), 1, AttemptStatus.RETRYABLE_FAILED,
+ new Progress(1, "retry"))).status(), Status.OK);
+ assertEquals(f.task(new Retry(f._owner, f._task, f.version())).status(), Status.OK);
+ assertEquals(f.worker(new Claim(f._task, f.version())).status(), Status.OK);
+ assertEquals(f._catalog.taskSnapshot(f._minion, TaskProtocol.MAX_SNAPSHOT_RECORDS,
+ TaskProtocol.MAX_SNAPSHOT_BYTES).status(), Status.OK);
+ assertEquals(f.worker(new Complete(f._task, f.version(), 2, AttemptStatus.FATAL_FAILED,
+ new Progress(0, "execution failed"))).status(), Status.OK);
+ assertEquals(f.task(new FireSchedule(f._owner, schedule, 1)).status(), Status.OK);
+ assertEquals(TaskProtocol.decodeSchedule(f.record(RecordKind.TASK_SCHEDULE, schedule.toString()).value())
+ .firedOccurrences(), 1L);
+ f.restore();
+ }
+ }
+
private static final class Fixture {
+ private final boolean _nativeTask;
private final UUID _cluster = UUID.randomUUID();
private final UUID _history = UUID.randomUUID();
private final UUID _table = UUID.randomUUID();
@@ -297,6 +502,11 @@ public class TaskCatalogTest {
private Result _lastWorkerResult;
private Fixture() throws Exception {
+ this(false);
+ }
+
+ private Fixture(boolean nativeTask) throws Exception {
+ _nativeTask = nativeTask;
assertEquals(apply(new RegisterClient(_history, "controller", UUID.randomUUID(), 0)).status(), Status.OK);
assertEquals(member(new ActivateControlEpoch(1, "controller", UUID.randomUUID())).status(), Status.OK);
assertEquals(member(new ConfigurePolicy(_catalog.membershipControl().controlEpoch(), 0,
@@ -320,15 +530,29 @@ public class TaskCatalogTest {
}
private Input input() {
+ if (_nativeTask) {
+ return new Input(new QueueKey(TABLE, _table, TaskProtocol.NATIVE_SEGMENT_GENERATION_TASK), "minion",
+ Map.of("sourceUri", "file:///inputs/source.json", "sourceSha256", "a".repeat(64)), 2,
+ new OutputPolicy("file:///outputs", 1));
+ }
return new Input(new QueueKey(TABLE, _table, "TestTask"), "minion", Map.of("tableName", TABLE), 2);
}
+ private Content output(UUID taskId, long attempt) {
+ String digest = "a".repeat(64);
+ return new Content(TABLE, _table, TaskProtocol.outputSegmentName(taskId, attempt, 0),
+ TaskProtocol.outputObjectId(_history, taskId, attempt, 0), digest,
+ TaskProtocol.outputUri(input().outputPolicy(), _history, taskId, attempt, _minion.incarnation(), 0, digest),
+ new LoaderDescriptor(1, 0, false, null, null, 1, 1, Map.of()));
+ }
+
private Result submit() throws IOException {
return task(new Submit(_owner, _task, input(), List.of()));
}
private ParticipantToken register(String id, Role role, long epoch) throws IOException {
- var response = member(new Register(id, role, UUID.randomUUID(), epoch, Map.of(), Set.of()));
+ var response = member(new Register(id, role, UUID.randomUUID(), epoch, Map.of(),
+ _nativeTask && role == Role.MINION ? Set.of(Capability.MINION_TASK) : Set.of()));
assertEquals(response.status(), Status.OK);
ParticipantToken token = response.participant().participant();
assertEquals(member(new SetReady(token, new Revision(_history, _index), true)).status(), Status.OK);
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java
index 78ffa4d1378..95968775247 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/MetadataGrpcTransportTest.java
@@ -44,6 +44,7 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -85,6 +86,7 @@ import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.HashVersion;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.Layout;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.TableShard;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol;
import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
import org.apache.pinot.coordination.ratis.rpc.MetadataServiceGrpc;
import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy.ControllerIdentity;
@@ -1382,6 +1384,152 @@ public class MetadataGrpcTransportTest {
throw new AssertionError("Native runtime did not advance its applied metadata heartbeat: " + server.failure());
}
+ @Test(timeOut = 90_000)
+ public void testMinionTaskIdentityOutcomesAndProtectedRenewal() throws Exception {
+ ParticipantIdentity minion = new ParticipantIdentity("Minion_worker", Role.MINION, Set.of(Capability.MINION_TASK));
+ AtomicReference<ParticipantIdentity> binding = new AtomicReference<>(minion);
+ ParticipantAccessPolicy policy = fingerprint -> {
+ if (fingerprint.equals(_certificates.get("worker").fingerprint())) {
+ return binding.get();
+ }
+ if (fingerprint.equals(_certificates.get("controller-a").fingerprint())) {
+ return new ParticipantIdentity("Controller_tasks", Role.CONTROLLER, Set.of());
+ }
+ return fingerprint.equals(_certificates.get("controller-b").fingerprint())
+ ? new ParticipantIdentity("Minion_other", Role.MINION, Set.of(Capability.MINION_TASK)) : null;
+ };
+ try (Quorum quorum = new Quorum()) {
+ quorum.startMembership();
+ BlockingAccess access = new BlockingAccess(quorum._access);
+ try (Fixture fixture = new Fixture(MetadataGrpcService.Limits.defaults(), quorum._history, access,
+ quorum._access, policy);
+ MetadataGrpcClient controller = fixture.client("controller-a");
+ MetadataGrpcClient worker = fixture.client("worker");
+ MetadataGrpcClient other = fixture.client("controller-b")) {
+ UUID history = quorum._history;
+ MetadataRpc.ParticipantToken controllerSession = registerReady(controller, history);
+ MetadataRpc.ParticipantToken workerSession = registerReady(worker, history);
+ MetadataRpc.ParticipantToken otherSession = registerReady(other, history);
+ assertEquals(workerSession.getRole(), MetadataRpc.ParticipantRole.MINION);
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.read(fixture.revision()));
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.openSnapshot(fixture.open()));
+ assertEquals(quorum._access.execute(new MetadataProtocol.RegisterClient(history, "native-test",
+ UUID.randomUUID(), 0), 5000).status(), MetadataProtocol.Status.OK);
+ MembershipProtocol.ControlToken control = quorum._access.query(new MetadataProtocol.MembershipQuery(history,
+ new MembershipProtocol.GetControl()), 5000).control().controlEpoch();
+ Layout layout = new Layout(1, 24, HashVersion.MURMUR2_UTF8_MASK_POSITIVE_V1);
+ commitNativeTest(quorum, new MetadataProtocol.OwnershipCommand(history, "native-test", 1, "ownership", 1,
+ new OwnershipProtocol.ConfigureLayout(control, layout)));
+ String tableName = "tasks_OFFLINE";
+ UUID tableId = UUID.randomUUID();
+ TableShard scope = OwnershipProtocol.tableScope(layout, tableName);
+ commitNativeTest(quorum, new MetadataProtocol.OwnershipCommand(history, "native-test", 1, "ownership", 2,
+ new OwnershipProtocol.AssignOwner(control, 1, scope, 0,
+ MembershipRpcCodec.participant(controllerSession))));
+ OwnershipProtocol.OwnerToken owner = new OwnershipProtocol.OwnerToken(history, 1, scope, 1,
+ MembershipRpcCodec.participant(controllerSession));
+ commitNativeTest(quorum, new MetadataProtocol.Transaction(history, "native-test", 1, "config", 1,
+ List.of(MetadataProtocol.Mutation.createTableConfig(history, tableId,
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(tableName).build()),
+ MetadataProtocol.Mutation.createSchema(history, UUID.randomUUID(),
+ new Schema.SchemaBuilder().setSchemaName("tasks").addSingleValueDimension("id", DataType.INT)
+ .build()))));
+ TaskProtocol.QueueKey queue = new TaskProtocol.QueueKey(tableName, tableId,
+ TaskProtocol.NATIVE_SEGMENT_GENERATION_TASK);
+ UUID ownTask = UUID.randomUUID();
+ UUID otherTask = UUID.randomUUID();
+ commitNativeTest(quorum, new MetadataProtocol.TaskCommand(history, "native-test", 1, "tasks", 1,
+ new TaskProtocol.Submit(owner, ownTask, new TaskProtocol.Input(queue, minion.participantId(), Map.of(), 2),
+ List.of())));
+ commitNativeTest(quorum, new MetadataProtocol.TaskCommand(history, "native-test", 1, "tasks", 2,
+ new TaskProtocol.Submit(owner, otherTask,
+ new TaskProtocol.Input(queue, otherSession.getParticipantId(), Map.of(), 2), List.of())));
+ MetadataRpc.TaskWorkRequest work = MetadataRpc.TaskWorkRequest.newBuilder().setHistoryId(history.toString())
+ .setSession(session(workerSession)).setMaxRecords(128).setMaxBytes(TaskProtocol.MAX_SNAPSHOT_BYTES).build();
+ MetadataRpc.TaskWorkResult image = worker.taskWork(work, 5000);
+ assertEquals(image.getStatus(), MetadataRpc.ResultStatus.OK);
+ assertEquals(image.getParticipant().getParticipant(), workerSession);
+ assertEquals(image.getRecordsList().stream().filter(r -> r.getKind() == MetadataRpc.RecordKind.TASK)
+ .map(r -> r.getValue().getTask().getTaskId()).toList(), List.of(ownTask.toString()));
+ assertEquals(other.taskWork(work, 5000).getStatus(), MetadataRpc.ResultStatus.FENCED);
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> controller.taskWork(work, 5000));
+ MetadataRpc.TaskWorkResult oversized = worker.taskWork(work.toBuilder().setMaxRecords(1).build(), 5000);
+ assertEquals(oversized.getStatus(), MetadataRpc.ResultStatus.RESPONSE_TOO_LARGE);
+ assertEquals(oversized.getRecordsCount(), 0);
+ MetadataRpc.TaskCommandRequest registration = MetadataRpc.TaskCommandRequest.newBuilder()
+ .setHistoryId(history.toString()).setSession(session(workerSession)).setRegister(
+ MetadataRpc.RegisterClient.newBuilder().setIncarnation(workerSession.getIncarnation())).build();
+ assertEquals(worker.taskExecute(registration, 5000).getStatus(), MetadataRpc.ResultStatus.OK);
+ assertStatus(Status.Code.INVALID_ARGUMENT, () -> worker.taskExecute(registration.toBuilder().setRegister(
+ registration.getRegister().toBuilder().setIncarnation(UUID.randomUUID().toString())).build(), 5000));
+ MetadataRpc.TaskCommandRequest claim = registration.toBuilder().setOperation(
+ MetadataRpc.SequencedTaskOperation.newBuilder().setClientEpoch(1).setStreamId("tasks").setSequence(1)
+ .setClaim(MetadataRpc.TaskClaim.newBuilder().setTaskId(ownTask.toString()).setExpectedVersion(1)))
+ .build();
+ MetadataRpc.CommandResult accepted = worker.taskExecute(claim, 5000);
+ assertEquals(accepted.getStatus(), MetadataRpc.ResultStatus.OK);
+ assertEquals(worker.taskExecute(claim, 5000), accepted, "Lost reply retries retain the committed outcome");
+ MetadataRpc.TaskOutcomeRequest outcome = MetadataRpc.TaskOutcomeRequest.newBuilder()
+ .setHistoryId(history.toString()).setSession(session(workerSession)).setOutcome(
+ MetadataRpc.OutcomeQuery.newBuilder().setClientEpoch(1).setStreamId("tasks").setSequence(1)).build();
+ assertEquals(worker.taskOutcome(outcome, 5000), accepted);
+ assertEquals(other.taskOutcome(outcome, 5000).getStatus(), MetadataRpc.ResultStatus.FENCED);
+ assertStatus(Status.Code.INVALID_ARGUMENT, () -> worker.taskExecute(claim.toBuilder().setUnknownFields(
+ UnknownFieldSet.newBuilder().addField(100, UnknownFieldSet.Field.newBuilder()
+ .addLengthDelimited(ByteString.copyFromUtf8("generic mutation")).build()).build()).build(), 5000));
+ binding.set(new ParticipantIdentity(minion.participantId(), Role.MINION, Set.of()));
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.taskWork(work, 5000));
+ assertStatus(Status.Code.PERMISSION_DENIED, () -> worker.taskExecute(claim, 5000));
+ binding.set(minion);
+ CountDownLatch entered = new CountDownLatch(2);
+ CountDownLatch release = new CountDownLatch(1);
+ access._beforeWork.set(() -> {
+ entered.countDown();
+ try {
+ if (!release.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Task test gate was not released");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ });
+ ExecutorService requests = Executors.newFixedThreadPool(2);
+ try {
+ Future<MetadataRpc.TaskWorkResult> first = requests.submit(() -> worker.taskWork(work, 5000));
+ Future<MetadataRpc.TaskWorkResult> second = requests.submit(() -> worker.taskWork(work, 5000));
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ MetadataRpc.MembershipResponse renewed = worker.membershipExecute(memberRenewal(history,
+ session(workerSession), 1, image.getRevision()));
+ assertEquals(renewed.getRenewals(0).getStatus(), MetadataRpc.ResultStatus.OK);
+ assertEquals(renewed.getRenewals(0).getPermit().getCapabilitiesList(),
+ List.of(MetadataRpc.ServingCapability.MINION_TASK));
+ assertEquals(release.getCount(), 1L, "Membership renewal must not wait for task handlers");
+ binding.set(new ParticipantIdentity(minion.participantId(), Role.MINION, Set.of()));
+ release.countDown();
+ for (Future<MetadataRpc.TaskWorkResult> pending : List.of(first, second)) {
+ ExecutionException denied = expectThrows(ExecutionException.class, () -> pending.get(5, TimeUnit.SECONDS));
+ assertEquals(Status.fromThrowable(denied.getCause()).getCode(), Status.Code.PERMISSION_DENIED);
+ }
+ } finally {
+ release.countDown();
+ requests.shutdownNow();
+ access._beforeWork.set(() -> { });
+ binding.set(minion);
+ }
+ MetadataRpc.TaskCommandRequest acknowledge = registration.toBuilder().setAcknowledge(
+ MetadataRpc.Acknowledge.newBuilder().setClientEpoch(1).setStreamId("tasks").setThroughSequence(1)).build();
+ assertEquals(worker.taskExecute(acknowledge, 5000).getStatus(), MetadataRpc.ResultStatus.OK);
+ assertEquals(worker.taskOutcome(outcome, 5000).getStatus(), MetadataRpc.ResultStatus.RESULT_EXPIRED);
+ assertEquals(worker.membershipExecute(MetadataRpc.MembershipCommandRequest.newBuilder()
+ .setHistoryId(history.toString()).setUnregister(session(workerSession)).build()).getStatus(),
+ MetadataRpc.ResultStatus.OK);
+ assertEquals(worker.taskExecute(claim, 5000).getStatus(), MetadataRpc.ResultStatus.FENCED);
+ assertEquals(worker.taskOutcome(outcome, 5000).getStatus(), MetadataRpc.ResultStatus.FENCED);
+ }
+ }
+ }
+
private static MetadataRpc.ParticipantToken registerReady(MetadataGrpcClient client, UUID history)
throws IOException {
return registerReady(client, history, 0);
@@ -1662,7 +1810,7 @@ public class MetadataGrpcTransportTest {
}
/// Blocks the four catalog handler threads without blocking membership access to the same real quorum.
- private static final class BlockingAccess implements MetadataAccess, InstallationAccess {
+ private static final class BlockingAccess implements MetadataAccess, InstallationAccess, TaskAccess {
private final MetadataAccess _delegate;
private final AtomicReference<Runnable> _afterExecute = new AtomicReference<>(() -> { });
private final AtomicReference<Runnable> _beforeWork = new AtomicReference<>(() -> { });
@@ -1689,6 +1837,13 @@ public class MetadataGrpcTransportTest {
return ((InstallationAccess) _delegate).query(query, timeoutMillis);
}
+ @Override
+ public TaskProtocol.WorkSnapshot taskSnapshot(MembershipProtocol.ParticipantToken minion, int maxRecords,
+ int maxBytes, long timeoutMillis) throws IOException {
+ _beforeWork.get().run();
+ return ((TaskAccess) _delegate).taskSnapshot(minion, maxRecords, maxBytes, timeoutMillis);
+ }
+
@Override
public OfflineInstallationProtocol.InstallationSnapshot installationSnapshot(
MembershipProtocol.ParticipantToken server, int maxRecords, int maxBytes, long timeoutMillis)
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodecTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodecTest.java
index 82003b62cc6..8c0872b66e5 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodecTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/transport/TaskRpcCodecTest.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantState;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ParticipantToken;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
@@ -29,14 +30,19 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogReco
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.OwnerToken;
import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.TableShard;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.AttemptState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.AttemptStatus;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Input;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.OutputPolicy;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.Progress;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.PublicationState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueKey;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.QueueState;
+import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.ScheduleState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskState;
import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskStatus;
import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
@@ -48,7 +54,7 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertThrows;
-/// Task read/feed values preserve their fences without exposing task commands or enabling MINION transport access.
+/// Typed task values preserve their fences; worker commands use only authenticated MINION session identities.
public class TaskRpcCodecTest {
@Test
public void testTypedTaskFeedRoundTrip() throws Exception {
@@ -61,13 +67,21 @@ public class TaskRpcCodecTest {
ParticipantToken minion = new ParticipantToken(history, "minion", Role.MINION, UUID.randomUUID(), 1);
ParticipantState participant = new ParticipantState(minion, Map.of(), Set.of(), 3, true, false, false,
new Revision(history, 2), 0, 3);
+ Input input = new Input(key, "minion", Map.of("key", "v"), 2, new OutputPolicy("file:/tmp/task-outputs/", 1));
+ List<ObjectVersion> dependencies =
+ List.of(new ObjectVersion(history, table, 1), new ObjectVersion(history, UUID.randomUUID(), 1));
+ Content output = new Content("orders_OFFLINE", table, "segment", UUID.randomUUID(), "a".repeat(64),
+ "file:/tmp/task-outputs/segment.tar.gz", new LoaderDescriptor(1, -1, false, null, null, -1, -1, Map.of()));
List<CatalogRecord> records = List.of(
record(RecordKind.TASK_QUEUE, "queue", new QueueState(history, key, false, owner)),
- record(RecordKind.TASK, "task", new TaskState(history, taskId, new Input(key, "minion", Map.of("key", "v"), 2),
- List.of(new ObjectVersion(history, table, 1), new ObjectVersion(history, UUID.randomUUID(), 1)), owner,
- TaskStatus.RUNNING, 1)),
+ record(RecordKind.TASK, "task",
+ new TaskState(history, taskId, input, dependencies, owner, TaskStatus.SUCCEEDED, 1)),
record(RecordKind.TASK_ATTEMPT, "attempt", new AttemptState(history, taskId, 1, participant,
- new ObjectVersion(history, UUID.randomUUID(), 2), 4, AttemptStatus.RUNNING, new Progress(0, ""))));
+ new ObjectVersion(history, UUID.randomUUID(), 2), 4, AttemptStatus.SUCCEEDED, new Progress(1, ""))),
+ record(RecordKind.TASK_SCHEDULE, "schedule", new ScheduleState(history, UUID.randomUUID(), input, dependencies,
+ owner, 0, 1, 2, 1, 1)),
+ record(RecordKind.TASK_PUBLICATION, "publication",
+ new PublicationState(history, taskId, 1, minion, List.of(output), 4)));
MetadataRpc.ChangeTransaction.Builder transaction = MetadataRpc.ChangeTransaction.newBuilder()
.setPreviousIndex(3).setIndex(4);
for (CatalogRecord record : records) {
@@ -86,10 +100,14 @@ public class TaskRpcCodecTest {
}
@Test
- public void testReadTypesDoNotAuthorizeTaskWritesOrMinionMembership() {
- assertThrows(IllegalArgumentException.class, () -> new ParticipantIdentity("minion", Role.MINION, Set.of()));
+ public void testReadTypesDoNotAuthorizeGenericTaskWritesOrQueryCapabilities() {
+ assertThrows(IllegalArgumentException.class,
+ () -> new ParticipantIdentity("minion", Role.MINION, Set.of(Capability.SERVER_QUERY)));
+ assertThrows(IllegalArgumentException.class,
+ () -> new ParticipantIdentity("server", Role.SERVER, Set.of(Capability.MINION_TASK)));
for (MetadataRpc.RecordKind kind : List.of(MetadataRpc.RecordKind.TASK_QUEUE, MetadataRpc.RecordKind.TASK,
- MetadataRpc.RecordKind.TASK_ATTEMPT)) {
+ MetadataRpc.RecordKind.TASK_ATTEMPT, MetadataRpc.RecordKind.TASK_SCHEDULE,
+ MetadataRpc.RecordKind.TASK_PUBLICATION)) {
assertThrows(IllegalArgumentException.class, () -> MetadataRpcCodec.mutation(
MetadataRpc.Mutation.newBuilder().setKind(kind).setAction(MetadataRpc.Mutation.Action.CREATE).build()));
}
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java b/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
index 05ccbd4f6eb..bd11a6e378f 100644
--- a/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
@@ -40,6 +40,7 @@ import org.apache.pinot.common.Utils;
import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
import org.apache.pinot.common.config.TlsConfig;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.common.metrics.MinionGauge;
import org.apache.pinot.common.metrics.MinionMeter;
import org.apache.pinot.common.metrics.MinionMetrics;
@@ -56,6 +57,7 @@ import org.apache.pinot.common.version.PinotVersion;
import org.apache.pinot.core.transport.ListenerConfig;
import org.apache.pinot.core.util.ListenerConfigUtil;
import org.apache.pinot.core.util.trace.ContinuousJfrStarter;
+import org.apache.pinot.minion.coordination.NativeMinionRuntime;
import org.apache.pinot.minion.event.DefaultMinionTaskObserverStorageManager;
import org.apache.pinot.minion.event.EventObserverFactoryRegistry;
import org.apache.pinot.minion.event.MinionEventObserverFactory;
@@ -88,6 +90,7 @@ public abstract class BaseMinionStarter implements ServiceStartable {
private static final String HTTPS_ENABLED = "enabled";
+ private NativeMinionRuntime _nativeRuntime;
protected MinionConf _config;
protected String _hostname;
protected int _port;
@@ -105,6 +108,12 @@ public abstract class BaseMinionStarter implements ServiceStartable {
public void init(PinotConfiguration config)
throws Exception {
_config = new MinionConf(config.toMap());
+ if (CoordinationBackend.from(_config) == CoordinationBackend.RATIS) {
+ applyCustomConfigs(_config);
+ _nativeRuntime = new NativeMinionRuntime(_config);
+ _instanceId = _nativeRuntime.instanceId();
+ return;
+ }
String zkAddress = _config.getZkAddress();
String helixClusterName = _config.getHelixClusterName();
ServiceStartableUtils.applyClusterConfig(_config, zkAddress, helixClusterName, ServiceRole.MINION);
@@ -161,6 +170,7 @@ public abstract class BaseMinionStarter implements ServiceStartable {
///
/// This is for pluggable task executor factories.
public void registerTaskExecutorFactory(PinotTaskExecutorFactory taskExecutorFactory) {
+ Preconditions.checkState(_nativeRuntime == null, "Native minions require qualified fenced task adapters");
_taskExecutorFactoryRegistry.registerTaskExecutorFactory(taskExecutorFactory);
}
@@ -168,6 +178,7 @@ public abstract class BaseMinionStarter implements ServiceStartable {
///
/// This is for pluggable event observer factories.
public void registerEventObserverFactory(MinionEventObserverFactory eventObserverFactory) {
+ Preconditions.checkState(_nativeRuntime == null, "Native minions do not register legacy task observers");
_eventObserverFactoryRegistry.registerEventObserverFactory(eventObserverFactory);
}
@@ -212,6 +223,10 @@ public abstract class BaseMinionStarter implements ServiceStartable {
@Override
public void start()
throws Exception {
+ if (_nativeRuntime != null) {
+ _nativeRuntime.start();
+ return;
+ }
LOGGER.info("Starting Pinot minion: {} (Version: {})", _instanceId, PinotVersion.VERSION);
LOGGER.info("Minion configs: {}", new PinotAppConfigs(getConfig()).toJSONString());
long startTimeMs = System.currentTimeMillis();
@@ -373,6 +388,14 @@ public abstract class BaseMinionStarter implements ServiceStartable {
/// Stops the Pinot Minion instance.
@Override
public void stop() {
+ if (_nativeRuntime != null) {
+ try {
+ _nativeRuntime.close();
+ } catch (Exception e) {
+ throw new IllegalStateException("Failed to stop native minion", e);
+ }
+ return;
+ }
try {
LOGGER.info("Closing PinotFS classes");
PinotFSFactory.shutdown();
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeMinionRuntime.java b/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeMinionRuntime.java
new file mode 100644
index 00000000000..064878fb3bc
--- /dev/null
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeMinionRuntime.java
@@ -0,0 +1,465 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.minion.coordination;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.coordination.MetadataRpcConnection;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus;
+import org.apache.pinot.minion.MinionConf;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+
+/// Native task participant behind the normal minion starter. The single worker consumes a complete own-session
+/// projection, stages immutable attempt outputs, and publishes only through consensus fencing. Membership uses
+/// independent channels/threads. No Helix connection, legacy plugin, singleton context, or query listener is created.
+public final class NativeMinionRuntime implements AutoCloseable, ServiceStatus.ServiceStatusCallback {
+ public static final String PREFIX = NativeMetadataConfig.PREFIX + "minion.";
+ private static final Logger LOGGER = LoggerFactory.getLogger(NativeMinionRuntime.class);
+ private final NativeMetadataConfig _metadataConfig;
+ private final Path _dataDirectory;
+ private final Path _sourceRoot;
+ private final Path _outputRoot;
+ private final long _maxSourceBytes;
+ private final int _maxRows;
+ private final long _pollMillis;
+ private final UUID _incarnation = UUID.randomUUID();
+ private final Map<String, Long> _executedAttempts = new HashMap<>();
+ private final ScheduledExecutorService _heartbeats = Executors.newSingleThreadScheduledExecutor(
+ runnable -> new Thread(runnable, "native-minion-membership"));
+ private final ScheduledExecutorService _worker = Executors.newSingleThreadScheduledExecutor(
+ runnable -> new Thread(runnable, "native-minion-task"));
+ @Nullable
+ private Rpc _membership;
+ @Nullable
+ private Rpc _work;
+ @Nullable
+ private FileChannel _lockFile;
+ @Nullable
+ private FileLock _directoryLock;
+ @Nullable
+ private NativeSegmentGenerationTask _generator;
+ @Nullable
+ private NativeTaskOutbox _outbox;
+ @Nullable
+ private volatile MetadataRpc.ParticipantToken _participant;
+ private volatile MetadataRpc.Revision _revision;
+ private volatile boolean _healthy;
+ private volatile boolean _hydrated;
+ private volatile boolean _closed;
+ private volatile boolean _fenced;
+ @Nullable
+ private volatile String _failure = "Not started";
+ private boolean _started;
+ private boolean _statusRegistered;
+ private long _renewal = 1;
+ @Nullable
+ private MetadataRpc.MembershipCommandRequest _pendingRenewal;
+
+ /// The transport must preserve payloads across retries and bound each invocation by the supplied timeout.
+ interface Rpc extends AutoCloseable {
+ MetadataRpc.MembershipResponse membershipRead(MetadataRpc.MembershipReadRequest request, long timeout);
+ MetadataRpc.MembershipResponse membershipExecute(MetadataRpc.MembershipCommandRequest request, long timeout);
+ MetadataRpc.TaskWorkResult taskWork(MetadataRpc.TaskWorkRequest request, long timeout);
+ MetadataRpc.CommandResult taskExecute(MetadataRpc.TaskCommandRequest request, long timeout);
+ @Override
+ void close();
+ }
+
+ private static final class Connection extends MetadataRpcConnection implements Rpc {
+ private Connection(NativeMetadataConfig config) throws IOException {
+ super(config.endpoints(), config.certificate(), config.privateKey(), config.trustCertificates(),
+ config.rpcTimeoutMillis());
+ }
+ }
+
+ public NativeMinionRuntime(PinotConfiguration config) {
+ _metadataConfig = new NativeMetadataConfig(config);
+ String instanceId = new MinionConf(config.toMap()).getInstanceId();
+ checkArgument(_metadataConfig.participantId().equals(instanceId) && instanceId.startsWith("Minion_"),
+ "Native minion requires an explicit Minion_ instance ID matching its participant ID");
+ _dataDirectory = directory(config, CommonConstants.Helix.Instance.DATA_DIR_KEY);
+ _sourceRoot = directory(config, PREFIX + "sourceRoot");
+ _outputRoot = directory(config, PREFIX + "outputRoot");
+ _maxSourceBytes = config.getProperty(PREFIX + "maxSourceBytes", 16L * 1024 * 1024);
+ _maxRows = config.getProperty(PREFIX + "maxRows", 100_000);
+ _pollMillis = config.getProperty(PREFIX + "pollMillis", 1000L);
+ checkArgument(_pollMillis > 0 && _pollMillis <= 30_000, "Native task poll interval must be between 1 and 30000 ms");
+ _revision = MetadataRpc.Revision.newBuilder().setHistoryId(_metadataConfig.historyId().toString()).build();
+ }
+
+ @VisibleForTesting
+ NativeMinionRuntime(PinotConfiguration config, Rpc membership, Rpc work) {
+ this(config);
+ checkArgument(membership != work, "Membership requires independent transport admission");
+ _membership = membership;
+ _work = work;
+ }
+
+ private static Path directory(PinotConfiguration config, String key) {
+ String value = config.getProperty(key);
+ checkArgument(value != null && !value.isBlank() && Path.of(value).isAbsolute(),
+ "Native minion requires an explicit absolute directory: %s", key);
+ return Path.of(value).normalize();
+ }
+
+ public String instanceId() {
+ return _metadataConfig.participantId();
+ }
+
+ public synchronized void start() throws Exception {
+ checkState(!_started && !_closed, "Native minion is already started or closed");
+ _started = true;
+ try {
+ Files.createDirectories(_dataDirectory);
+ _lockFile = FileChannel.open(_dataDirectory.resolve(".native-minion.lock"), StandardOpenOption.CREATE,
+ StandardOpenOption.WRITE);
+ _directoryLock = _lockFile.tryLock();
+ checkState(_directoryLock != null, "Native minion directory is already owned");
+ _generator = new NativeSegmentGenerationTask(_sourceRoot, _outputRoot, _maxSourceBytes, _maxRows);
+ if (_membership == null) {
+ _membership = new Connection(_metadataConfig);
+ }
+ if (_work == null) {
+ _work = new Connection(_metadataConfig);
+ }
+ MetadataRpc.MembershipResponse current = _membership.membershipRead(MetadataRpc.MembershipReadRequest
+ .newBuilder().setHistoryId(_revision.getHistoryId()).build(), timeout());
+ checkState(current.getStatus() == ResultStatus.OK || current.getStatus() == ResultStatus.NOT_FOUND,
+ "Cannot inspect native minion registration: %s", current.getStatus());
+ long previousEpoch = current.hasParticipant()
+ ? current.getParticipant().getParticipant().getRegistrationEpoch() : 0;
+ MetadataRpc.MembershipResponse registered = _membership.membershipExecute(command()
+ .setRegister(MetadataRpc.ParticipantRegistration.newBuilder().setIncarnation(_incarnation.toString())
+ .setExpectedEpoch(previousEpoch)).build(), timeout());
+ checkState(registered.getStatus() == ResultStatus.OK && registered.hasParticipant(),
+ "Cannot register native minion: %s", registered.getStatus());
+ _participant = registered.getParticipant().getParticipant();
+ checkState(_participant.getHistoryId().equals(_revision.getHistoryId())
+ && _participant.getParticipantId().equals(instanceId())
+ && _participant.getIncarnation().equals(_incarnation.toString())
+ && _participant.getRole() == MetadataRpc.ParticipantRole.MINION, "Native minion registration differs");
+ snapshot();
+ MetadataRpc.MembershipResponse ready = _membership.membershipExecute(command().setReady(
+ MetadataRpc.ParticipantReadiness.newBuilder().setSession(session()).setAppliedRevision(_revision)
+ .setReady(true)).build(), timeout());
+ checkState(ready.getStatus() == ResultStatus.OK, "Native minion readiness was rejected: %s", ready.getStatus());
+ snapshot();
+ Path sessions = _dataDirectory.resolve("task-sessions");
+ Files.createDirectories(sessions);
+ try (var entries = Files.list(sessions)) {
+ checkState(entries.limit(128).count() < 128, "Native minion recovery evidence requires explicit archival");
+ }
+ _outbox = new NativeTaskOutbox(_work, _participant, sessions.resolve(_incarnation.toString()), timeout());
+ heartbeat();
+ checkState(_healthy, "Native minion initial heartbeat failed: %s", _failure);
+ ServiceStatus.setServiceStatusCallback(instanceId(), this);
+ _statusRegistered = true;
+ _heartbeats.scheduleWithFixedDelay(this::heartbeat, _metadataConfig.heartbeatMillis(),
+ _metadataConfig.heartbeatMillis(), TimeUnit.MILLISECONDS);
+ _worker.scheduleWithFixedDelay(this::poll, 0, _pollMillis, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ try {
+ close();
+ } catch (Exception cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ throw e;
+ }
+ }
+
+ private long timeout() {
+ return _metadataConfig.rpcTimeoutMillis();
+ }
+
+ private MetadataRpc.MembershipCommandRequest.Builder command() {
+ return MetadataRpc.MembershipCommandRequest.newBuilder().setHistoryId(_metadataConfig.historyId().toString());
+ }
+
+ private MetadataRpc.ParticipantSession session() {
+ checkState(_participant != null, "Native minion is unregistered");
+ return MetadataRpc.ParticipantSession.newBuilder().setIncarnation(_participant.getIncarnation())
+ .setRegistrationEpoch(_participant.getRegistrationEpoch()).build();
+ }
+
+ private MetadataRpc.TaskWorkResult snapshot() {
+ MetadataRpc.TaskWorkResult image = _work.taskWork(MetadataRpc.TaskWorkRequest.newBuilder()
+ .setHistoryId(_metadataConfig.historyId().toString()).setSession(session()).setMaxRecords(128)
+ .setMaxBytes(2 * 1024 * 1024).build(), timeout());
+ if (image.getStatus() == ResultStatus.FENCED || image.getStatus() == ResultStatus.HISTORY_MISMATCH) {
+ _fenced = true;
+ }
+ checkState(image.getStatus() == ResultStatus.OK && image.hasParticipant() && image.hasControl()
+ && image.getParticipant().getParticipant().equals(_participant)
+ && image.getRevision().getHistoryId().equals(_participant.getHistoryId())
+ && image.getRevision().getIndex() >= _revision.getIndex() && image.getRecordsCount() <= 128
+ && image.getSerializedSize() <= 2 * 1024 * 1024, "Native minion work snapshot is unavailable or inconsistent");
+ checkState(!image.getParticipant().getRetired() && !image.getParticipant().getDraining()
+ && image.getParticipant().getExpiryRound() > image.getControl().getRound()
+ && image.getParticipant().getCapabilitiesList().equals(List.of(MetadataRpc.ServingCapability.MINION_TASK)),
+ "Native minion task capability or live registration is unavailable");
+ _revision = image.getRevision();
+ _hydrated = true;
+ return image;
+ }
+
+ private void heartbeat() {
+ if (_closed || _fenced) {
+ return;
+ }
+ try {
+ if (_pendingRenewal == null) {
+ _pendingRenewal = command().setRenew(MetadataRpc.ParticipantRenewal.newBuilder().setSession(session())
+ .setSequence(_renewal).setAppliedRevision(_revision)).build();
+ }
+ MetadataRpc.MembershipResponse response = _membership.membershipExecute(_pendingRenewal, timeout());
+ if (response.getStatus() == ResultStatus.FENCED || response.getStatus() == ResultStatus.HISTORY_MISMATCH) {
+ _fenced = true;
+ }
+ checkState(response.getStatus() == ResultStatus.OK && response.getRenewalsCount() == 1,
+ "Native minion renewal rejected: %s", response.getStatus());
+ MetadataRpc.RenewalResult renewal = response.getRenewals(0);
+ if (renewal.getStatus() == ResultStatus.FENCED) {
+ _fenced = true;
+ }
+ checkState(renewal.getParticipant().equals(_participant) && renewal.getSequence() == _renewal,
+ "Native minion renewal identity differs");
+ checkState((renewal.getStatus() == ResultStatus.OK || renewal.getStatus() == ResultStatus.NOT_READY)
+ && _renewal < Long.MAX_VALUE, "Native minion renewal has an unusable durable outcome");
+ // NOT_READY is a committed consumed sequence too; retrying it can never become a lease grant.
+ _renewal++;
+ _pendingRenewal = null;
+ _healthy = renewal.getStatus() == ResultStatus.OK;
+ if (!_healthy) {
+ _failure = "Native minion heartbeat awaits a current metadata image";
+ } else if (_hydrated) {
+ _failure = null;
+ }
+ } catch (Exception e) {
+ _healthy = false;
+ _failure = "Native minion heartbeat is unavailable";
+ LOGGER.debug("Native minion heartbeat failed", e);
+ }
+ }
+
+ private void poll() {
+ if (_closed || _fenced) {
+ return;
+ }
+ try {
+ // Hydrate even when renewal is not ready, so advancing applied revision can recover a lagging heartbeat.
+ MetadataRpc.TaskWorkResult image = snapshot();
+ if (!_healthy) {
+ return;
+ }
+ if (_outbox.pending()) {
+ MetadataRpc.CommandResult result = _outbox.resolve();
+ if (result.getStatus() != ResultStatus.OK) {
+ _failure = "Native task transition rejected: " + result.getStatus();
+ }
+ return;
+ }
+ Map<String, MetadataRpc.CatalogRecord> records = new HashMap<>();
+ Map<String, Long> running = new HashMap<>();
+ for (MetadataRpc.CatalogRecord record : image.getRecordsList()) {
+ checkState(records.putIfAbsent(record.getObjectId() + ":" + record.getVersion(), record) == null,
+ "Duplicate task projection record");
+ if (record.getKind() == MetadataRpc.RecordKind.TASK
+ && record.getValue().getTask().getState() == MetadataRpc.TaskExecutionState.State.RUNNING) {
+ running.put(record.getValue().getTask().getTaskId(), record.getValue().getTask().getAttempt());
+ }
+ }
+ _executedAttempts.entrySet().removeIf(entry -> !entry.getValue().equals(running.get(entry.getKey())));
+ for (MetadataRpc.CatalogRecord record : image.getRecordsList()) {
+ if (record.getKind() != MetadataRpc.RecordKind.TASK || record.getDeleted()) {
+ continue;
+ }
+ MetadataRpc.TaskExecutionState task = record.getValue().getTask();
+ if (!task.getInput().getMinionId().equals(instanceId())) {
+ throw new IllegalStateException("Task belongs to another minion");
+ }
+ if (task.getState() == MetadataRpc.TaskExecutionState.State.RUNNING && ownAttempt(image, task)
+ && !_executedAttempts.containsKey(task.getTaskId())) {
+ execute(task, record.getVersion(), records);
+ _executedAttempts.put(task.getTaskId(), task.getAttempt());
+ return;
+ }
+ }
+ // A READY task must not starve a later RUNNING attempt by repeatedly failing the one-active-task check.
+ if (!running.isEmpty()) {
+ return;
+ }
+ for (MetadataRpc.CatalogRecord record : image.getRecordsList()) {
+ if (record.getKind() == MetadataRpc.RecordKind.TASK && !record.getDeleted()) {
+ MetadataRpc.TaskExecutionState task = record.getValue().getTask();
+ if (task.getState() == MetadataRpc.TaskExecutionState.State.READY && !queuePaused(image, task)) {
+ _outbox.enqueue(MetadataRpc.SequencedTaskOperation.newBuilder().setClaim(MetadataRpc.TaskClaim.newBuilder()
+ .setTaskId(task.getTaskId()).setExpectedVersion(record.getVersion())));
+ return;
+ }
+ }
+ }
+ } catch (Exception e) {
+ _hydrated = false;
+ _failure = "Native minion work is unavailable";
+ LOGGER.debug("Native minion task poll failed", e);
+ }
+ }
+
+ private static boolean queuePaused(MetadataRpc.TaskWorkResult image, MetadataRpc.TaskExecutionState task) {
+ return image.getRecordsList().stream().filter(record -> record.getKind() == MetadataRpc.RecordKind.TASK_QUEUE)
+ .map(record -> record.getValue().getTaskQueue())
+ .filter(queue -> queue.getQueue().equals(task.getInput().getQueue()))
+ .findFirst().map(MetadataRpc.TaskQueueState::getPaused).orElse(true);
+ }
+
+ private boolean ownAttempt(MetadataRpc.TaskWorkResult image, MetadataRpc.TaskExecutionState task) {
+ return image.getRecordsList().stream().filter(record -> record.getKind() == MetadataRpc.RecordKind.TASK_ATTEMPT)
+ .map(record -> record.getValue().getTaskAttempt())
+ .anyMatch(attempt -> attempt.getTaskId().equals(task.getTaskId()) && attempt.getAttempt() == task.getAttempt()
+ && attempt.getState() == MetadataRpc.TaskAttemptState.State.RUNNING
+ && attempt.getParticipantAtClaim().getParticipant().equals(_participant));
+ }
+
+ private void execute(MetadataRpc.TaskExecutionState task, long version,
+ Map<String, MetadataRpc.CatalogRecord> records) throws Exception {
+ try {
+ TableConfig table = null;
+ Schema schema = null;
+ for (MetadataRpc.ObjectVersion dependency : task.getDependenciesList()) {
+ MetadataRpc.CatalogRecord record = records.get(dependency.getObjectId() + ":" + dependency.getVersion());
+ checkState(record != null && dependency.getHistoryId().equals(_participant.getHistoryId()),
+ "Captured task dependency is absent");
+ if (record.getKind() == MetadataRpc.RecordKind.TABLE_CONFIG) {
+ checkState(table == null && record.getObjectId().equals(task.getInput().getQueue().getTableObjectId()),
+ "Captured task table differs");
+ table = JsonUtils.bytesToObject(record.getValue().getTableConfigJson().toByteArray(), TableConfig.class);
+ } else if (record.getKind() == MetadataRpc.RecordKind.SCHEMA) {
+ checkState(schema == null, "Captured task schema is ambiguous");
+ schema = Schema.fromString(record.getValue().getSchemaJson().toStringUtf8());
+ }
+ }
+ checkState(table != null && schema != null, "Captured task table and schema are required");
+ MetadataRpc.OfflineSegmentContent output = _generator.generate(_participant, task, table, schema,
+ _dataDirectory.resolve("working"), this::checkRunning);
+ checkRunning();
+ _outbox.enqueue(MetadataRpc.SequencedTaskOperation.newBuilder().setPublish(
+ MetadataRpc.TaskPublishOfflineOutputs.newBuilder().setTaskId(task.getTaskId()).setExpectedVersion(version)
+ .setAttempt(task.getAttempt()).addOutputs(output).setProgress(MetadataRpc.TaskProgress.newBuilder()
+ .setCompletedUnits(1).setDetail("Generated immutable segment"))));
+ } catch (CancellationException e) {
+ if (_closed || _fenced || Thread.currentThread().isInterrupted()) {
+ throw e;
+ }
+ // Heartbeat uncertainty stops local work. Resolve this attempt after recovery instead of regenerating it.
+ _outbox.enqueue(MetadataRpc.SequencedTaskOperation.newBuilder().setComplete(MetadataRpc.TaskComplete.newBuilder()
+ .setTaskId(task.getTaskId()).setExpectedVersion(version).setAttempt(task.getAttempt())
+ .setResult(MetadataRpc.TaskAttemptState.State.RETRYABLE_FAILED)
+ .setProgress(MetadataRpc.TaskProgress.newBuilder()
+ .setDetail("Native segment generation stopped during heartbeat uncertainty"))));
+ } catch (Exception e) {
+ checkRunning();
+ _outbox.enqueue(MetadataRpc.SequencedTaskOperation.newBuilder().setComplete(MetadataRpc.TaskComplete.newBuilder()
+ .setTaskId(task.getTaskId()).setExpectedVersion(version).setAttempt(task.getAttempt())
+ .setResult(MetadataRpc.TaskAttemptState.State.FATAL_FAILED).setProgress(MetadataRpc.TaskProgress.newBuilder()
+ .setDetail("Native segment generation failed"))));
+ LOGGER.warn("Native task generation failed for {}", task.getTaskId(), e);
+ }
+ }
+
+ private void checkRunning() {
+ if (_closed || _fenced || !_healthy || Thread.currentThread().isInterrupted()) {
+ throw new CancellationException("Native minion execution was cancelled or its live session is uncertain");
+ }
+ }
+
+ @Override
+ public ServiceStatus.Status getServiceStatus() {
+ return _closed ? ServiceStatus.Status.NOT_STARTED
+ : _healthy && _hydrated && !_fenced ? ServiceStatus.Status.GOOD : ServiceStatus.Status.BAD;
+ }
+
+ @Override
+ public String getStatusDescription() {
+ return _failure == null ? "Native task worker ready; legacy plugins and query listeners disabled" : _failure;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ if (_closed) {
+ return;
+ }
+ _closed = true;
+ _healthy = false;
+ _worker.shutdownNow();
+ _heartbeats.shutdownNow();
+ if (_statusRegistered) {
+ ServiceStatus.removeServiceStatusCallback(instanceId(), this);
+ _statusRegistered = false;
+ }
+ if (_participant != null && _membership != null) {
+ try {
+ _membership.membershipExecute(command().setUnregister(session()).build(), timeout());
+ } catch (Exception e) {
+ LOGGER.debug("Native minion retirement could not be confirmed; lease expiry fences the session", e);
+ }
+ }
+ if (_work != null) {
+ _work.close();
+ }
+ if (_membership != null) {
+ _membership.close();
+ }
+ // Keep ownership if an uninterruptible plugin/IO action is still running; a replacement must not overlap it.
+ checkState(_worker.awaitTermination(30, TimeUnit.SECONDS), "Native minion worker is still stopping; lock retained");
+ checkState(_heartbeats.awaitTermination(5, TimeUnit.SECONDS), "Native minion heartbeat is still stopping");
+ if (_directoryLock != null) {
+ _directoryLock.release();
+ }
+ if (_lockFile != null) {
+ _lockFile.close();
+ }
+ }
+}
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeSegmentGenerationTask.java b/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeSegmentGenerationTask.java
new file mode 100644
index 00000000000..a6d803838d5
--- /dev/null
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeSegmentGenerationTask.java
@@ -0,0 +1,256 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.minion.coordination;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.util.HexFormat;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.coordination.NativeTaskOutputIdentity;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.apache.pinot.spi.data.readers.RecordReaderConfig;
+import org.apache.pinot.spi.utils.JsonUtils;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// Generates one genuine OFFLINE segment from a digest-pinned JSON-lines file. Local root approvals constrain
+/// input reads and immutable staging. No legacy upload, replacement, deletion, or metadata publication is invoked.
+public final class NativeSegmentGenerationTask {
+ private final Path _sourceRoot;
+ private final Path _outputRoot;
+ private final long _maxSourceBytes;
+ private final int _maxRows;
+
+ public NativeSegmentGenerationTask(Path sourceRoot, Path outputRoot, long maxSourceBytes, int maxRows)
+ throws IOException {
+ _sourceRoot = sourceRoot.toRealPath();
+ Files.createDirectories(outputRoot);
+ _outputRoot = outputRoot.toRealPath();
+ checkArgument(Files.isDirectory(_sourceRoot) && Files.isDirectory(_outputRoot), "Task roots must be directories");
+ checkArgument(maxSourceBytes > 0 && maxSourceBytes <= 64 * 1024 * 1024 && maxRows > 0 && maxRows <= 1_000_000,
+ "Native generation requires bounded source bytes and rows");
+ _maxSourceBytes = maxSourceBytes;
+ _maxRows = maxRows;
+ }
+
+ public MetadataRpc.OfflineSegmentContent generate(MetadataRpc.ParticipantToken participant,
+ MetadataRpc.TaskExecutionState task, TableConfig table, Schema schema, Path workingDirectory, Runnable guard)
+ throws Exception {
+ checkArgument(NativeTaskOutputIdentity.TASK_TYPE.equals(task.getInput().getQueue().getTaskType())
+ && task.getInput().hasOutputPolicy() && task.getInput().getOutputPolicy().getMaxOutputs() == 1,
+ "Only one-output native segment generation is supported");
+ checkArgument(table.getTableType() == TableType.OFFLINE
+ && table.getTableName().equals(task.getInput().getQueue().getTableName()), "Task requires its OFFLINE table");
+ checkArgument(Path.of(NativeTaskOutputIdentity.outputRoot(task.getInput().getOutputPolicy().getRootUri()))
+ .toRealPath().equals(_outputRoot), "Task output root is not locally approved");
+ Map<String, String> configs = task.getInput().getConfigsMap();
+ checkArgument(configs.keySet().equals(Set.of("sourceUri", "sourceSha256")), "Unsupported generation options");
+ String expectedDigest = configs.get("sourceSha256");
+ checkArgument(expectedDigest.matches("[a-f0-9]{64}"), "Source SHA-256 is required");
+ URI sourceUri = URI.create(configs.get("sourceUri"));
+ checkArgument("file".equals(sourceUri.getScheme()) && sourceUri.getAuthority() == null
+ && sourceUri.getQuery() == null && sourceUri.getFragment() == null,
+ "Only approved local sources are supported");
+ Path source = Path.of(sourceUri).toRealPath();
+ checkArgument(source.startsWith(_sourceRoot) && Files.isRegularFile(source, LinkOption.NOFOLLOW_LINKS),
+ "Task source is outside the approved root");
+ Files.createDirectories(workingDirectory);
+ Path temporary = Files.createTempDirectory(workingDirectory, "generation-");
+ try {
+ guard.run();
+ Path captured = temporary.resolve("input.json");
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ try (InputStream input = Files.newInputStream(source, LinkOption.NOFOLLOW_LINKS);
+ var output = Files.newOutputStream(captured, StandardOpenOption.CREATE_NEW)) {
+ byte[] buffer = new byte[8192];
+ long size = 0;
+ for (int count; (count = input.read(buffer)) != -1;) {
+ guard.run();
+ size += count;
+ checkArgument(size <= _maxSourceBytes, "Native task source exceeds its byte limit");
+ digest.update(buffer, 0, count);
+ output.write(buffer, 0, count);
+ }
+ }
+ checkArgument(HexFormat.of().formatHex(digest.digest()).equals(expectedDigest), "Source digest differs");
+ UUID taskId = UUID.fromString(task.getTaskId());
+ UUID history = UUID.fromString(participant.getHistoryId());
+ String segmentName = NativeTaskOutputIdentity.outputSegmentName(taskId, task.getAttempt(), 0);
+ Path generated = Files.createDirectory(temporary.resolve("generated"));
+ SegmentGeneratorConfig generator = new SegmentGeneratorConfig(table, schema);
+ generator.setOutDir(generated.toString());
+ generator.setSegmentName(segmentName);
+ generator.setContinueOnError(false);
+ generator.setFailOnEmptySegment(true);
+ try (JsonLinesReader reader = new JsonLinesReader(captured, _maxRows, guard)) {
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(generator, reader);
+ driver.build();
+ }
+ guard.run();
+ Path segment = generated.resolve(segmentName);
+ SegmentMetadataImpl metadata = new SegmentMetadataImpl(segment.toFile());
+ Path archive = temporary.resolve("output.tar.gz");
+ TarCompressionUtils.createCompressedTarFile(segment.toFile(), archive.toFile());
+ digest.reset();
+ try (InputStream input = Files.newInputStream(archive)) {
+ byte[] buffer = new byte[8192];
+ for (int count; (count = input.read(buffer)) != -1;) {
+ guard.run();
+ digest.update(buffer, 0, count);
+ }
+ }
+ String sha256 = HexFormat.of().formatHex(digest.digest());
+ String uri = NativeTaskOutputIdentity.outputUri(task.getInput().getOutputPolicy().getRootUri(), history,
+ taskId, task.getAttempt(), UUID.fromString(participant.getIncarnation()), 0, sha256);
+ Path policyRoot = Path.of(NativeTaskOutputIdentity.outputRoot(task.getInput().getOutputPolicy().getRootUri()));
+ Path destination = _outputRoot.resolve(policyRoot.relativize(Path.of(URI.create(uri))));
+ createOutputDirectories(destination.getParent());
+ // CREATE_NEW prevents a retry or stale process from overwriting any attempt's artifact.
+ try (FileChannel output = FileChannel.open(destination, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
+ InputStream input = Files.newInputStream(archive)) {
+ byte[] buffer = new byte[8192];
+ for (int count; (count = input.read(buffer)) != -1;) {
+ guard.run();
+ ByteBuffer bytes = ByteBuffer.wrap(buffer, 0, count);
+ while (bytes.hasRemaining()) {
+ output.write(bytes);
+ }
+ }
+ output.force(true);
+ }
+ // Persist directory entries through the approved root before metadata can make the archive reachable.
+ for (Path directory = destination.getParent(); directory.startsWith(_outputRoot);
+ directory = directory.getParent()) {
+ try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) {
+ channel.force(true);
+ }
+ }
+ guard.run();
+ return MetadataRpc.OfflineSegmentContent.newBuilder().setTableName(table.getTableName())
+ .setTableObjectId(task.getInput().getQueue().getTableObjectId()).setSegmentName(segmentName)
+ .setSegmentObjectId(NativeTaskOutputIdentity.outputObjectId(history, taskId, task.getAttempt(), 0).toString())
+ .setContentSha256(sha256).setDownloadUri(uri).setLoader(MetadataRpc.SegmentLoaderDescriptor.newBuilder()
+ .setCrc(metadata.getCrc()).setCreationTimeMs(metadata.getIndexCreationTime()).setPushTimeMs(-1)).build();
+ } finally {
+ FileUtils.deleteDirectory(temporary.toFile());
+ }
+ }
+
+ private void createOutputDirectories(Path parent) throws IOException {
+ checkArgument(parent.normalize().startsWith(_outputRoot), "Output escaped approved root");
+ Path current = _outputRoot;
+ for (Path component : _outputRoot.relativize(parent)) {
+ current = current.resolve(component);
+ if (!Files.exists(current, LinkOption.NOFOLLOW_LINKS)) {
+ Files.createDirectory(current);
+ }
+ checkArgument(Files.isDirectory(current, LinkOption.NOFOLLOW_LINKS)
+ && current.toRealPath().startsWith(_outputRoot), "Output directory contains a symbolic link");
+ }
+ }
+
+ /// The driver may rewind for dictionary/index construction. Each pass validates the same captured input bounds.
+ private static final class JsonLinesReader implements RecordReader {
+ private final Path _file;
+ private final int _maxRows;
+ private final Runnable _guard;
+ private BufferedReader _reader;
+ @Nullable
+ private String _next;
+ private int _rows;
+
+ private JsonLinesReader(Path file, int maxRows, Runnable guard) throws IOException {
+ _file = file;
+ _maxRows = maxRows;
+ _guard = guard;
+ rewind();
+ }
+
+ @Override
+ public void init(File file, Set<String> fields, RecordReaderConfig config) {
+ throw new UnsupportedOperationException("Native reader is initialized with its captured file");
+ }
+
+ @Override
+ public boolean hasNext() {
+ _guard.run();
+ if (_next == null) {
+ try {
+ _next = _reader.readLine();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ return _next != null;
+ }
+
+ @Override
+ public GenericRow next(GenericRow reuse) throws IOException {
+ _guard.run();
+ checkArgument(_next != null && ++_rows <= _maxRows, "Native task exceeds its row limit");
+ Map<String, Object> row = JsonUtils.stringToMap(_next);
+ _next = null;
+ reuse.clear();
+ reuse.putValues(row);
+ return reuse;
+ }
+
+ @Override
+ public void rewind() throws IOException {
+ close();
+ _reader = Files.newBufferedReader(_file, StandardCharsets.UTF_8);
+ _next = null;
+ _rows = 0;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (_reader != null) {
+ _reader.close();
+ }
+ }
+ }
+}
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeTaskOutbox.java b/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeTaskOutbox.java
new file mode 100644
index 00000000000..5ba1e20d5d6
--- /dev/null
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/coordination/NativeTaskOutbox.java
@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.minion.coordination;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import javax.annotation.Nullable;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus;
+
+import static com.google.common.base.Preconditions.checkState;
+
+
+/// One serial durable request stream for an exact MINION incarnation. A pending payload and its known outcome are
+/// persisted before acknowledgement; unknown replies reuse identical bytes. Old-session journals are evidence only
+/// and are never replayed as a new process. Published artifacts remain independent of this journal's lifetime.
+final class NativeTaskOutbox {
+ private static final String STREAM = "tasks";
+ private final NativeMinionRuntime.Rpc _rpc;
+ private final MetadataRpc.TaskCommandRequest _envelope;
+ private final Path _directory;
+ private final long _timeoutMillis;
+ private long _sequence = 1;
+ @Nullable
+ private MetadataRpc.TaskCommandRequest _pending;
+ @Nullable
+ private MetadataRpc.CommandResult _result;
+
+ NativeTaskOutbox(NativeMinionRuntime.Rpc rpc, MetadataRpc.ParticipantToken token, Path directory, long timeoutMillis)
+ throws Exception {
+ _rpc = rpc;
+ _directory = directory;
+ _timeoutMillis = timeoutMillis;
+ Files.createDirectory(directory);
+ write(directory.resolve("session.pb"), token.toByteArray());
+ _envelope = MetadataRpc.TaskCommandRequest.newBuilder().setHistoryId(token.getHistoryId())
+ .setSession(MetadataRpc.ParticipantSession.newBuilder().setIncarnation(token.getIncarnation())
+ .setRegistrationEpoch(token.getRegistrationEpoch())).build();
+ MetadataRpc.TaskCommandRequest registration = _envelope.toBuilder().setRegister(MetadataRpc.RegisterClient
+ .newBuilder().setIncarnation(token.getIncarnation())).build();
+ write(directory.resolve("registration.pb"), registration.toByteArray());
+ MetadataRpc.CommandResult registered = _rpc.taskExecute(registration, timeoutMillis);
+ require(registered);
+ checkState(registered.getClientEpoch() == 1, "Native task session client epoch differs");
+ }
+
+ boolean pending() {
+ return _pending != null;
+ }
+
+ void enqueue(MetadataRpc.SequencedTaskOperation.Builder operation) throws IOException {
+ checkState(_pending == null && _sequence < Long.MAX_VALUE, "Task stream has pending work or is exhausted");
+ MetadataRpc.TaskCommandRequest request = _envelope.toBuilder().setOperation(operation
+ .setClientEpoch(1).setStreamId(STREAM).setSequence(_sequence)).build();
+ checkState(request.getSerializedSize() <= 64 * 1024, "Task command exceeds its durable bound");
+ write(_directory.resolve("pending.pb"), request.toByteArray());
+ _pending = request;
+ _result = null;
+ }
+
+ MetadataRpc.CommandResult resolve() throws Exception {
+ checkState(_pending != null, "Task stream has no pending operation");
+ if (_result == null) {
+ MetadataRpc.CommandResult result = _rpc.taskExecute(_pending, _timeoutMillis);
+ checkState(result.hasRevision() && result.getRevision().getHistoryId().equals(_envelope.getHistoryId()),
+ "Task command response history differs");
+ // A retained domain rejection is acknowledged too. Session rejection cannot advance the stream because
+ // acknowledgement is independently fenced by that same exact participant session.
+ write(_directory.resolve("result.pb"), result.toByteArray());
+ _result = result;
+ }
+ MetadataRpc.CommandResult acknowledgement = _rpc.taskExecute(_envelope.toBuilder()
+ .setAcknowledge(MetadataRpc.Acknowledge.newBuilder().setClientEpoch(1).setStreamId(STREAM)
+ .setThroughSequence(_sequence)).build(), _timeoutMillis);
+ require(acknowledgement);
+ MetadataRpc.CommandResult result = _result;
+ // Retain the last request/result as bounded evidence; the next exact payload replaces them after this ack.
+ _pending = null;
+ _result = null;
+ _sequence++;
+ return result;
+ }
+
+ private static void require(MetadataRpc.CommandResult result) {
+ checkState(result.getStatus() == ResultStatus.OK, "Task stream request rejected: %s", result.getStatus());
+ }
+
+ private static void write(Path destination, byte[] bytes) throws IOException {
+ Path temporary = Files.createTempFile(destination.getParent(), ".write-", ".tmp");
+ try {
+ try (FileChannel output = FileChannel.open(temporary, StandardOpenOption.WRITE)) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ while (buffer.hasRemaining()) {
+ output.write(buffer);
+ }
+ output.force(true);
+ }
+ Files.move(temporary, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
+ try (FileChannel directory = FileChannel.open(destination.getParent(), StandardOpenOption.READ)) {
+ directory.force(true);
+ }
+ } finally {
+ Files.deleteIfExists(temporary);
+ }
+ }
+}
diff --git a/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeMinionRuntimeTest.java b/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeMinionRuntimeTest.java
new file mode 100644
index 00000000000..a5c5d5d7923
--- /dev/null
+++ b/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeMinionRuntimeTest.java
@@ -0,0 +1,126 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.minion.coordination;
+
+import java.nio.channels.OverlappingFileLockException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// A consumed semantic heartbeat rejection must advance, and failed startup must preserve the live instance.
+public class NativeMinionRuntimeTest {
+ @Test
+ public void testHeartbeatRecoveryAndFailedDirectoryOwnership() throws Exception {
+ Path directory = Files.createTempDirectory("native-minion-lifecycle-");
+ String history = UUID.randomUUID().toString();
+ String instanceId = "Minion_" + UUID.randomUUID();
+ try {
+ Path certificate = Files.writeString(directory.resolve("test.pem"), "injected transport");
+ PinotConfiguration config = new PinotConfiguration();
+ String prefix = NativeMetadataConfig.PREFIX;
+ config.setProperty(prefix + "historyId", history);
+ config.setProperty(prefix + "participantId", instanceId);
+ config.setProperty(prefix + "metadata.endpoints", "localhost:12345");
+ config.setProperty(prefix + "tls.certChain", certificate.toString());
+ config.setProperty(prefix + "tls.privateKey", certificate.toString());
+ config.setProperty(prefix + "tls.trustCerts", certificate.toString());
+ config.setProperty(prefix + "heartbeatMillis", 20L);
+ config.setProperty(NativeMinionRuntime.PREFIX + "pollMillis", 5L);
+ config.setProperty(NativeMinionRuntime.PREFIX + "sourceRoot", directory.toString());
+ config.setProperty(NativeMinionRuntime.PREFIX + "outputRoot", directory.resolve("outputs").toString());
+ config.setProperty(CommonConstants.Minion.CONFIG_OF_MINION_ID, instanceId);
+ config.setProperty(CommonConstants.Helix.Instance.DATA_DIR_KEY, directory.resolve("data").toString());
+ NativeMinionRuntime.Rpc membership = mock(NativeMinionRuntime.Rpc.class);
+ NativeMinionRuntime.Rpc work = mock(NativeMinionRuntime.Rpc.class);
+ MetadataRpc.Revision revision = MetadataRpc.Revision.newBuilder().setHistoryId(history).setIndex(10).build();
+ AtomicReference<MetadataRpc.ParticipantToken> participant = new AtomicReference<>();
+ CountDownLatch recovered = new CountDownLatch(1);
+ when(membership.membershipRead(any(), anyLong())).thenReturn(MetadataRpc.MembershipResponse.newBuilder()
+ .setStatus(ResultStatus.NOT_FOUND).setRevision(revision).build());
+ when(membership.membershipExecute(any(), anyLong())).thenAnswer(invocation -> {
+ MetadataRpc.MembershipCommandRequest request = invocation.getArgument(0);
+ MetadataRpc.MembershipResponse.Builder response = MetadataRpc.MembershipResponse.newBuilder()
+ .setStatus(ResultStatus.OK).setRevision(revision);
+ if (request.hasRegister()) {
+ participant.set(MetadataRpc.ParticipantToken.newBuilder().setHistoryId(history).setParticipantId(instanceId)
+ .setRole(MetadataRpc.ParticipantRole.MINION).setIncarnation(request.getRegister().getIncarnation())
+ .setRegistrationEpoch(1).build());
+ return response.setParticipant(state(participant.get())).build();
+ }
+ if (request.hasRenew()) {
+ long sequence = request.getRenew().getSequence();
+ if (sequence >= 3) {
+ recovered.countDown();
+ }
+ return response.addRenewals(MetadataRpc.RenewalResult.newBuilder().setParticipant(participant.get())
+ .setSequence(sequence).setStatus(sequence == 2 ? ResultStatus.NOT_READY : ResultStatus.OK)).build();
+ }
+ return response.build();
+ });
+ when(work.taskWork(any(), anyLong())).thenAnswer(invocation -> {
+ MetadataRpc.TaskWorkRequest request = invocation.getArgument(0);
+ assertEquals(request.getMaxRecords(), 128);
+ return MetadataRpc.TaskWorkResult.newBuilder().setStatus(ResultStatus.OK).setRevision(revision)
+ .setParticipant(state(participant.get())).setControl(MetadataRpc.MembershipControl.newBuilder().setRound(1))
+ .build();
+ });
+ when(work.taskExecute(any(), anyLong())).thenReturn(MetadataRpc.CommandResult.newBuilder()
+ .setStatus(ResultStatus.OK).setRevision(revision).setClientEpoch(1).build());
+ try (NativeMinionRuntime runtime = new NativeMinionRuntime(config, membership, work)) {
+ runtime.start();
+ assertTrue(recovered.await(5, TimeUnit.SECONDS), "A known NOT_READY renewal must not replay forever");
+ try (NativeMinionRuntime competing = new NativeMinionRuntime(config, mock(NativeMinionRuntime.Rpc.class),
+ mock(NativeMinionRuntime.Rpc.class))) {
+ assertThrows(OverlappingFileLockException.class, competing::start);
+ assertNotEquals(ServiceStatus.getServiceStatus(instanceId), ServiceStatus.Status.NOT_STARTED,
+ "Failed data-directory ownership must not unregister the existing minion status");
+ }
+ }
+ assertEquals(ServiceStatus.getServiceStatus(instanceId), ServiceStatus.Status.NOT_STARTED);
+ } finally {
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ private static MetadataRpc.ParticipantState state(MetadataRpc.ParticipantToken participant) {
+ return MetadataRpc.ParticipantState.newBuilder().setParticipant(participant).setExpiryRound(100)
+ .addCapabilities(MetadataRpc.ServingCapability.MINION_TASK).build();
+ }
+}
diff --git a/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeSegmentGenerationTaskTest.java b/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeSegmentGenerationTaskTest.java
new file mode 100644
index 00000000000..1582f211e31
--- /dev/null
+++ b/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeSegmentGenerationTaskTest.java
@@ -0,0 +1,139 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.minion.coordination;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.HexFormat;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.coordination.NativeTaskOutputIdentity;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Real generation verifies immutable artifact identity, captured source integrity and local filesystem boundaries.
+public class NativeSegmentGenerationTaskTest {
+ @Test
+ public void testRealSegmentArtifactAndBounds() throws Exception {
+ Path directory = Files.createTempDirectory("native-minion-generation-");
+ try {
+ Path sources = Files.createDirectory(directory.resolve("sources"));
+ Path outputs = Files.createDirectory(directory.resolve("outputs"));
+ Path source = sources.resolve("orders.json");
+ Files.writeString(source, "{\"id\":1,\"label\":\"one\"}\n{\"id\":2,\"label\":\"two\"}\n");
+ String hash = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(source)));
+ MetadataRpc.ParticipantToken participant = MetadataRpc.ParticipantToken.newBuilder()
+ .setHistoryId(UUID.randomUUID().toString()).setParticipantId("Minion_test")
+ .setIncarnation(UUID.randomUUID().toString()).setRegistrationEpoch(1)
+ .setRole(MetadataRpc.ParticipantRole.MINION).build();
+ MetadataRpc.TaskExecutionState task = task(participant, source, outputs, hash);
+ TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").build();
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("orders").addSingleValueDimension("id", DataType.LONG)
+ .addSingleValueDimension("label", DataType.STRING).build();
+ NativeSegmentGenerationTask generator = new NativeSegmentGenerationTask(sources, outputs, 1024, 2);
+ MetadataRpc.OfflineSegmentContent result = generator.generate(participant, task, table, schema,
+ directory.resolve("work"), () -> { });
+ Path archive = Path.of(java.net.URI.create(result.getDownloadUri()));
+ assertTrue(Files.isRegularFile(archive));
+ assertEquals(HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(archive))),
+ result.getContentSha256());
+ assertEquals(result.getSegmentName(),
+ NativeTaskOutputIdentity.outputSegmentName(UUID.fromString(task.getTaskId()), 1, 0));
+ Path extracted = Files.createDirectory(directory.resolve("extracted"));
+ TarCompressionUtils.untar(archive.toFile(), extracted.toFile());
+ SegmentMetadataImpl metadata = new SegmentMetadataImpl(extracted.resolve(result.getSegmentName()).toFile());
+ assertEquals(metadata.getTotalDocs(), 2);
+ assertEquals(metadata.getCrc(), result.getLoader().getCrc());
+ assertFalse(Files.exists(directory.resolve("work").resolve(result.getSegmentName())));
+ byte[] original = Files.readAllBytes(archive);
+ MetadataRpc.TaskExecutionState wrongDigest = task.toBuilder().setInput(task.getInput().toBuilder()
+ .putConfigs("sourceSha256", "0".repeat(64))).build();
+ assertThrows(IllegalArgumentException.class,
+ () -> generator.generate(participant, wrongDigest, table, schema, directory.resolve("work"), () -> { }));
+ assertThrows(IllegalArgumentException.class, () -> new NativeSegmentGenerationTask(sources, outputs, 1, 2)
+ .generate(participant, task, table, schema, directory.resolve("work"), () -> { }));
+ assertThrows(Exception.class, () -> new NativeSegmentGenerationTask(sources, outputs, 1024, 1)
+ .generate(participant, task, table, schema, directory.resolve("work"), () -> { }));
+ assertEquals(Files.readAllBytes(archive), original, "Failed attempts cannot replace a staged artifact");
+ } finally {
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ @Test
+ public void testSourceAndOutputSymlinksCannotEscapeApprovedRoots() throws Exception {
+ Path directory = Files.createTempDirectory("native-minion-boundaries-");
+ try {
+ Path sources = Files.createDirectory(directory.resolve("sources"));
+ Path outputs = Files.createDirectory(directory.resolve("outputs"));
+ Path outside = Files.createDirectory(directory.resolve("outside"));
+ Path source = outside.resolve("input.json");
+ Files.writeString(source, "{\"id\":1}\n");
+ String hash = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(source)));
+ MetadataRpc.ParticipantToken participant = MetadataRpc.ParticipantToken.newBuilder()
+ .setHistoryId(UUID.randomUUID().toString()).setParticipantId("Minion_test")
+ .setIncarnation(UUID.randomUUID().toString()).setRegistrationEpoch(1)
+ .setRole(MetadataRpc.ParticipantRole.MINION).build();
+ Path linked = Files.createSymbolicLink(sources.resolve("input.json"), source);
+ TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").build();
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("orders").addSingleValueDimension("id", DataType.LONG)
+ .build();
+ NativeSegmentGenerationTask generator = new NativeSegmentGenerationTask(sources, outputs, 1024, 10);
+ assertThrows(IllegalArgumentException.class, () -> generator.generate(participant,
+ task(participant, linked, outputs, hash), table, schema, directory.resolve("work"), () -> { }));
+ Files.delete(linked);
+ Files.copy(source, linked);
+ Files.createSymbolicLink(outputs.resolve(participant.getHistoryId()), outside);
+ assertThrows(IllegalArgumentException.class, () -> generator.generate(participant,
+ task(participant, linked, outputs, hash), table, schema, directory.resolve("work"), () -> { }));
+ try (var files = Files.list(outside)) {
+ assertEquals(files.count(), 1L, "No attempt artifacts may be created outside the approved root");
+ }
+ } finally {
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+
+ private static MetadataRpc.TaskExecutionState task(MetadataRpc.ParticipantToken participant, Path source,
+ Path outputs, String digest) {
+ return MetadataRpc.TaskExecutionState.newBuilder().setHistoryId(participant.getHistoryId())
+ .setTaskId(UUID.randomUUID().toString()).setAttempt(1).setState(MetadataRpc.TaskExecutionState.State.RUNNING)
+ .setInput(MetadataRpc.TaskInput.newBuilder().setMinionId(participant.getParticipantId()).setMaxAttempts(2)
+ .setQueue(MetadataRpc.TaskQueueKey.newBuilder().setTableName("orders_OFFLINE")
+ .setTableObjectId(UUID.randomUUID().toString()).setTaskType(NativeTaskOutputIdentity.TASK_TYPE))
+ .putAllConfigs(Map.of("sourceUri", source.toUri().toString(), "sourceSha256", digest))
+ .setOutputPolicy(MetadataRpc.TaskOutputPolicy.newBuilder().setRootUri(outputs.toUri().toString())
+ .setMaxOutputs(1))).build();
+ }
+}
diff --git a/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeTaskOutboxTest.java b/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeTaskOutboxTest.java
new file mode 100644
index 00000000000..2a71a699038
--- /dev/null
+++ b/pinot-minion/src/test/java/org/apache/pinot/minion/coordination/NativeTaskOutboxTest.java
@@ -0,0 +1,100 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.minion.coordination;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.coordination.ratis.rpc.MetadataRpc;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Lost mutation and acknowledgement replies must retain the same task attempt and durable command sequence.
+public class NativeTaskOutboxTest {
+ @Test
+ public void testUnknownOutcomeAndAckRetryRetainExactRequest() throws Exception {
+ Path directory = Files.createTempDirectory("native-task-outbox-");
+ try {
+ Path journal = directory.resolve("session");
+ MetadataRpc.ParticipantToken participant = MetadataRpc.ParticipantToken.newBuilder()
+ .setHistoryId(UUID.randomUUID().toString()).setParticipantId("Minion_test")
+ .setIncarnation(UUID.randomUUID().toString()).setRegistrationEpoch(1)
+ .setRole(MetadataRpc.ParticipantRole.MINION).build();
+ MetadataRpc.CommandResult ok = MetadataRpc.CommandResult.newBuilder().setStatus(MetadataRpc.ResultStatus.OK)
+ .setRevision(MetadataRpc.Revision.newBuilder().setHistoryId(participant.getHistoryId()).setIndex(10))
+ .setClientEpoch(1).build();
+ NativeMinionRuntime.Rpc rpc = mock(NativeMinionRuntime.Rpc.class);
+ List<MetadataRpc.TaskCommandRequest> operations = new ArrayList<>();
+ AtomicInteger acknowledgements = new AtomicInteger();
+ when(rpc.taskExecute(any(), anyLong())).thenAnswer(invocation -> {
+ MetadataRpc.TaskCommandRequest request = invocation.getArgument(0);
+ if (request.hasRegister()) {
+ assertTrue(Files.exists(journal.resolve("registration.pb")));
+ return ok;
+ }
+ if (request.hasOperation()) {
+ operations.add(request);
+ assertEquals(MetadataRpc.TaskCommandRequest.parseFrom(Files.readAllBytes(journal.resolve("pending.pb"))),
+ request);
+ if (operations.size() == 1) {
+ throw new IOException("Reply lost after commit");
+ }
+ return ok;
+ }
+ assertEquals(MetadataRpc.CommandResult.parseFrom(Files.readAllBytes(journal.resolve("result.pb"))), ok);
+ if (acknowledgements.incrementAndGet() == 1) {
+ throw new IOException("Acknowledgement reply lost");
+ }
+ return ok;
+ });
+ NativeTaskOutbox outbox = new NativeTaskOutbox(rpc, participant, journal, 1000);
+ outbox.enqueue(MetadataRpc.SequencedTaskOperation.newBuilder().setClaim(MetadataRpc.TaskClaim.newBuilder()
+ .setTaskId(UUID.randomUUID().toString()).setExpectedVersion(1)));
+ assertThrows(IOException.class, outbox::resolve);
+ assertTrue(outbox.pending());
+ assertThrows(IOException.class, outbox::resolve);
+ assertTrue(outbox.pending());
+ assertEquals(outbox.resolve(), ok);
+ assertFalse(outbox.pending());
+ assertEquals(operations.size(), 2);
+ assertEquals(operations.get(0), operations.get(1));
+ assertEquals(acknowledgements.get(), 2);
+ outbox.enqueue(MetadataRpc.SequencedTaskOperation.newBuilder().setClaim(MetadataRpc.TaskClaim.newBuilder()
+ .setTaskId(UUID.randomUUID().toString()).setExpectedVersion(1)));
+ assertEquals(outbox.resolve(), ok);
+ assertEquals(operations.get(2).getOperation().getSequence(), 2L);
+ } finally {
+ FileUtils.deleteDirectory(directory.toFile());
+ }
+ }
+}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java b/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
index 1bb4af364a9..36c976d2d21 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
@@ -174,14 +174,12 @@ public class PinotServiceManager {
public String startMinion(String minionStarterClassName, PinotConfiguration minionConf)
throws Exception {
- if (CoordinationBackend.from(minionConf) == CoordinationBackend.RATIS) {
- throw new IllegalArgumentException("Ratis minion startup is not supported");
- }
LOGGER.info("Trying to start Pinot Minion...");
if (!minionConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
minionConf.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
}
- if (!minionConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
+ if (CoordinationBackend.from(minionConf) == CoordinationBackend.HELIX
+ && !minionConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER)) {
minionConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPER_SERVER, _zkAddress);
}
ServiceStartable minionStarter = getServiceStartable(minionStarterClassName);
diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
index 5a42b42d39c..682d3b7ed38 100644
--- a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
+++ b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java
@@ -58,6 +58,7 @@ import org.apache.pinot.controller.coordination.RatisControllerProvider;
import org.apache.pinot.coordination.ratis.RatisMetadataStore;
import org.apache.pinot.coordination.ratis.RatisMetadataStore.MembershipClockSettings;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Capability;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol;
@@ -173,9 +174,11 @@ public class RatisStartersTest {
CertificateFiles controller = certificate(directory, "controller");
CertificateFiles broker = certificate(directory, "broker");
CertificateFiles server = certificate(directory, "server");
+ CertificateFiles minion = certificate(directory, "minion");
Path trust = directory.resolve("clients.pem");
Files.writeString(trust, Files.readString(controller.certificate()) + Files.readString(broker.certificate())
- + Files.readString(server.certificate()));
+ + Files.readString(server.certificate()) + Files.readString(minion.certificate()));
+ Path taskOutput = Files.createDirectories(directory.resolve("task-output"));
int[] ports = ports(12);
String history = UUID.randomUUID().toString();
String cluster = UUID.randomUUID().toString();
@@ -215,6 +218,9 @@ public class RatisStartersTest {
config.put("test.controllerFingerprint", controller.fingerprint());
config.put("test.brokerFingerprint", broker.fingerprint());
config.put("test.serverFingerprint", server.fingerprint());
+ config.put("test.minionFingerprint", minion.fingerprint());
+ config.put(PREFIX + "controller.tasks.outputRoot", taskOutput.toString());
+ config.put(PREFIX + "controller.tasks.pollMillis", 100L);
controllers.add(config);
starts.add(starters.submit(() -> manager.startRole(ServiceRole.CONTROLLER, config)));
}
@@ -412,6 +418,8 @@ public class RatisStartersTest {
awaitSize(http, controllerUrl + "/tables/orders/size", "reportedSizeInBytes", expectedBytes);
assertEquals(ServiceStatus.getServiceStatus(BROKER), ServiceStatus.Status.BAD);
assertEquals(ServiceStatus.getServiceStatus(SERVER), ServiceStatus.Status.GOOD);
+ runNativeTask(manager, directory, history, endpoints, minion, controller, http, controllerUrl, taskOutput,
+ new PinotConfiguration(controllers.getFirst()), serverState.getParticipant(), serverAdmin);
for (int i = 6; i < 8; i++) {
try (ServerSocket unused = new ServerSocket(ports[i])) {
assertTrue(unused.isBound(), "Native query listener must remain closed");
@@ -466,11 +474,132 @@ public class RatisStartersTest {
// Participants retire while a quorum remains available; the manager's arbitrary iteration order is unsuitable.
manager.stopPinotInstanceById(BROKER);
manager.stopPinotInstanceById(SERVER);
+ manager.stopPinotInstanceById("Minion_native_test");
manager.stopAll();
FileUtils.deleteDirectory(directory.toFile());
}
}
+ private static void runNativeTask(PinotServiceManager manager, Path directory, String history, String endpoints,
+ CertificateFiles minion, CertificateFiles controller, HttpClient http, String controllerUrl, Path output,
+ PinotConfiguration voterConfig, MetadataRpc.ParticipantToken server, String serverAdmin) throws Exception {
+ Path sourceRoot = Files.createDirectories(directory.resolve("task-source"));
+ Path source = sourceRoot.resolve("rows.json");
+ Files.writeString(source, "{\"id\":1}\n{\"id\":2}\n");
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("nativeTasks")
+ .addSingleValueDimension("id", DataType.LONG).build();
+ Map<String, String> headers = Map.of("If-None-Match", "*", "Idempotency-Key", UUID.randomUUID().toString());
+ HttpResponse<String> created = write(http, controllerUrl + "/schemas", "POST", schema.toSingleLineJsonString(),
+ headers);
+ assertEquals(created.statusCode(), 200, created.body());
+ TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName("nativeTasks").setNumReplicas(1).build();
+ created = write(http, controllerUrl + "/tables", "POST", JsonUtils.objectToString(table),
+ Map.of("If-None-Match", "*", "Idempotency-Key", UUID.randomUUID().toString()));
+ assertEquals(created.statusCode(), 200, created.body());
+ Map<String, Object> config = tls(history, minion, controller.certificate());
+ config.put(PREFIX + "participantId", "Minion_native_test");
+ config.put(PREFIX + "metadata.endpoints", endpoints);
+ config.put("pinot.minion.instance.id", "Minion_native_test");
+ config.put("dataDir", directory.resolve("minion-data").toString());
+ config.put(PREFIX + "minion.sourceRoot", sourceRoot.toString());
+ config.put(PREFIX + "minion.outputRoot", output.toString());
+ config.put(PREFIX + "minion.pollMillis", 100L);
+ config.put(PREFIX + "heartbeatMillis", 100L);
+ assertEquals(manager.startRole(ServiceRole.MINION, config), "Minion_native_test");
+ try {
+ String url = controllerUrl + "/tasks/native/nativeTasks";
+ String schedule = url + "/schedules/" + UUID.randomUUID();
+ String sourceSha256 = HexFormat.of().formatHex(
+ MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(source)));
+ String body = JsonUtils.objectToString(Map.of("minionId", "Minion_native_test", "sourceUri", source.toUri(),
+ "sourceSha256", sourceSha256,
+ "firstRound", 0, "intervalRounds", 0, "occurrences", 1, "maxAttempts", 2));
+ HttpResponse<String> submitted = write(http, schedule, "PUT", body, Map.of());
+ assertEquals(submitted.statusCode(), 200, submitted.body());
+ submitted = write(http, schedule, "PUT", body, Map.of());
+ assertEquals(submitted.statusCode(), 200, "Repeating the immutable schedule must not create another occurrence");
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+ HttpResponse<String> state;
+ do {
+ state = get(http, url, true);
+ if (state.statusCode() == 200
+ && JsonUtils.stringToJsonNode(state.body()).path("publications").size() == 1) {
+ break;
+ }
+ Thread.sleep(100);
+ } while (System.nanoTime() < deadline);
+ assertEquals(state.statusCode(), 200, state.body());
+ var snapshot = JsonUtils.stringToJsonNode(state.body());
+ assertEquals(snapshot.path("tasks").size(), 1, state.body());
+ assertEquals(snapshot.path("tasks").get(0).path("value").path("status").asText(), "SUCCEEDED", state.body());
+ assertEquals(snapshot.path("schedules").get(0).path("value").path("firedOccurrences").asLong(), 1L);
+ var content = snapshot.path("publications").get(0).path("value").path("outputs").get(0);
+ Path archive = Path.of(URI.create(content.path("downloadUri").asText()));
+ assertTrue(archive.toRealPath().startsWith(output.toRealPath()));
+ assertTrue(Files.size(archive) > 0);
+ assertEquals(HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(archive))),
+ content.path("contentSha256").asText());
+ Path unpacked = Files.createDirectories(directory.resolve("task-unpacked"));
+ TarCompressionUtils.untar(archive.toFile(), unpacked.toFile());
+ SegmentMetadataImpl generated = new SegmentMetadataImpl(unpacked.resolve(content.path("segmentName").asText())
+ .toFile());
+ assertEquals(generated.getTotalDocs(), 2);
+ installTaskOutput(voterConfig, server, OfflineSegmentProtocol.decodeContent(content));
+ long bytes = FileUtils.sizeOfDirectory(unpacked.resolve(content.path("segmentName").asText()).resolve("v3")
+ .toFile());
+ awaitSize(http, serverAdmin + "/tables/nativeTasks_OFFLINE/size", "diskSizeInBytes", bytes);
+ assertTrue(manager.stopPinotInstanceById("Minion_native_test"));
+ assertEquals(manager.startRole(ServiceRole.MINION, config), "Minion_native_test");
+ snapshot = JsonUtils.stringToJsonNode(get(http, url, true).body());
+ assertEquals(snapshot.path("tasks").size(), 1);
+ assertEquals(snapshot.path("publications").size(), 1);
+ } finally {
+ manager.stopPinotInstanceById("Minion_native_test");
+ }
+ }
+
+ /// Placement uses the existing trusted test protocol; the minion publishes content without acquiring controller
+ /// ownership or installation authority. This verifies the output is installable by the normal native server.
+ private static void installTaskOutput(PinotConfiguration config, MetadataRpc.ParticipantToken serverSession,
+ OfflineSegmentProtocol.Content content) throws Exception {
+ UUID history = UUID.fromString(serverSession.getHistoryId());
+ try (RatisMetadataStore.Client client = RatisMetadataStore.connect(config)) {
+ assertEquals(MetadataProtocol.decodeResult(client.execute(MetadataProtocol.encode(
+ new MetadataProtocol.RegisterClient(history, "task-install-test", UUID.randomUUID(), 0)))).status(),
+ MetadataProtocol.Status.OK);
+ CatalogRecord table = record(client, history, MetadataProtocol.RecordKind.TABLE_CONFIG, content.tableName());
+ CatalogRecord schema = record(client, history, MetadataProtocol.RecordKind.SCHEMA, "nativeTasks");
+ var layout = OwnershipProtocol.decodeLayout(record(client, history, MetadataProtocol.RecordKind.OWNER_LAYOUT,
+ OwnershipProtocol.LAYOUT_NAME).value());
+ var scope = OwnershipProtocol.tableScope(layout, content.tableName());
+ var owner = OwnershipProtocol.decodeAssignment(record(client, history,
+ MetadataProtocol.RecordKind.OWNER_ASSIGNMENT, OwnershipProtocol.name(scope)).value());
+ var token =
+ new OwnershipProtocol.OwnerToken(history, layout.generation(), scope, owner.ownerEpoch(), owner.owner());
+ var server = new MembershipProtocol.ParticipantToken(history, serverSession.getParticipantId(), Role.SERVER,
+ UUID.fromString(serverSession.getIncarnation()), serverSession.getRegistrationEpoch());
+ commit(client, new MetadataProtocol.OfflineSegmentCommand(history, "task-install-test", 1, "segments", 1,
+ new OfflineSegmentProtocol.AssignReplica(token, content.segmentObjectId(), 0, server,
+ new OfflineSegmentProtocol.Compatibility(new ObjectVersion(history, table.objectId(), table.version()),
+ new ObjectVersion(history, schema.objectId(), schema.version())), true)));
+ var assignment = OfflineSegmentProtocol.decodeAssignment(record(client, history,
+ MetadataProtocol.RecordKind.OFFLINE_REPLICA_ASSIGNMENT,
+ OfflineSegmentProtocol.assignmentName(content.segmentObjectId(), serverSession.getParticipantId())).value());
+ UUID operationId = UUID.randomUUID();
+ var intent = new OperationProtocol.OfflineSegmentReload(content.tableName(), content.tableObjectId(),
+ content.segmentName(), content.segmentObjectId(), content.contentSha256(),
+ List.of(new OperationProtocol.TargetMaterialization(server, assignment.assignmentEpoch(),
+ assignment.materializationId())), true, true, false);
+ commit(client, new MetadataProtocol.NativeOperationCommand(history, "task-install-test", 1, "load", 1,
+ new OperationProtocol.Submit(operationId, intent)));
+ var operation = OperationProtocol.decodeOperation(record(client, history,
+ MetadataProtocol.RecordKind.CONTROLLER_OPERATION, operationId.toString()).value());
+ commit(client, new MetadataProtocol.NativeOperationCommand(history, "task-install-test", 1, "load", 2,
+ new OperationProtocol.Claim(operationId, 1, token,
+ ((OperationProtocol.AssignedOfflineSegmentReload) operation.input()).dependencies())));
+ }
+ }
+
/// Uses the existing trusted consensus test API to publish content; public segment upload remains a separate API.
private static long installSizeTestSegment(Path directory, PinotConfiguration config, MetadataRpcClient metadata,
MetadataRpc.ParticipantToken serverSession, TableConfig table, Schema schema) throws Exception {
@@ -700,6 +829,7 @@ public class RatisStartersTest {
private String _controller;
private String _broker;
private String _server;
+ private String _minion;
private long _permitMillis;
@Override
@@ -707,6 +837,7 @@ public class RatisStartersTest {
_controller = config.getProperty("test.controllerFingerprint");
_broker = config.getProperty("test.brokerFingerprint");
_server = config.getProperty("test.serverFingerprint");
+ _minion = config.getProperty("test.minionFingerprint");
_permitMillis = config.getProperty("test.bootstrapPermitMillis", 1000L);
}
@@ -730,6 +861,10 @@ public class RatisStartersTest {
if (fingerprint.equals(_server)) {
return new ParticipantIdentity(SERVER, Role.SERVER, Set.of());
}
+ if (fingerprint.equals(_minion)) {
+ return new ParticipantIdentity("Minion_native_test", Role.MINION,
+ Set.of(Capability.MINION_TASK));
+ }
return fingerprint.equals(_controller) ? new ParticipantIdentity("Controller_test", Role.CONTROLLER, Set.of())
: null;
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org