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 01:15:07 UTC
(pinot) branch xiangfu0/codex2/ratis-controller-rest-ui created (now 701da8dda10)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a change to branch xiangfu0/codex2/ratis-controller-rest-ui
in repository https://gitbox.apache.org/repos/asf/pinot.git
at 701da8dda10 Expose Ratis controller metadata REST APIs and UI
This branch includes the following new commits:
new 701da8dda10 Expose Ratis controller metadata REST APIs and UI
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Expose Ratis controller metadata REST APIs and UI
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-controller-rest-ui
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 701da8dda10101efe3fe6d3a790fae6db894fc96
Author: Xiang Fu <xi...@apache.org>
AuthorDate: Mon Oct 19 18:14:49 2026 -0700
Expose Ratis controller metadata REST APIs and UI
---
.../pinot/common/audit/AuditConfigManager.java | 2 +-
.../pinot/common/audit/AuditServiceBinder.java | 14 +-
.../api/ControllerAdminApiApplication.java | 49 +-
.../api/access/AccessControlFactory.java | 7 +
.../api/access/AllowAllAccessFactory.java | 12 +
.../api/access/BasicAuthAccessControlFactory.java | 9 +
.../api/resources/PinotControllerAuthResource.java | 19 +-
.../coordination/NativeControllerMetadata.java | 496 +++++++++++++++
.../coordination/NativeControllerResource.java | 701 +++++++++++++++++++++
.../coordination/RatisControllerRuntime.java | 83 ++-
pinot-controller/src/main/resources/app/App.tsx | 18 +-
.../src/main/resources/app/components/Layout.tsx | 10 +-
.../app/components/NativeMetadataEditor.tsx | 152 +++++
.../resources/app/components/auth/AuthProvider.tsx | 86 +--
.../resources/app/pages/NativeMetadataPage.tsx | 289 +++++++++
.../src/main/resources/app/requests/index.ts | 2 +-
.../main/resources/app/requests/nativeMetadata.ts | 138 ++++
.../main/resources/app/utils/PinotMethodUtils.ts | 5 +-
.../src/main/resources/app/utils/axios-config.ts | 2 +
pinot-controller/src/main/resources/vite.config.ts | 2 +-
.../access/BasicAuthAccessControlFactoryTest.java | 27 +
.../resources/PinotControllerAuthResourceTest.java | 54 ++
.../coordination/NativeControllerResourceTest.java | 291 +++++++++
.../coordination/NativeControllerWriteTest.java | 334 ++++++++++
pinot-coordination-ratis/README.md | 86 ++-
.../coordination/ratis/RatisMetadataAccess.java | 12 +-
.../ratis/metadata/MetadataCatalog.java | 67 +-
.../ratis/metadata/MetadataProtocol.java | 35 +-
.../ratis/transport/MetadataAccess.java | 7 +
.../coordination/ratis/RatisMetadataStoreTest.java | 7 +
.../ratis/metadata/ControllerCoordinationTest.java | 8 +
.../ratis/metadata/MetadataCatalogTest.java | 117 ++++
.../segment/local/utils/TableConfigUtils.java | 41 +-
.../segment/local/utils/TableConfigUtilsTest.java | 28 +
.../pinot/tools/service/RatisStartersTest.java | 210 +++++-
35 files changed, 3310 insertions(+), 110 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditConfigManager.java b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditConfigManager.java
index f92bcd9c41e..621610d4562 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditConfigManager.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditConfigManager.java
@@ -43,7 +43,7 @@ public final class AuditConfigManager implements PinotClusterConfigChangeListene
Map.of(ServiceRole.CONTROLLER, "controller", ServiceRole.SERVER, "server", ServiceRole.BROKER, "broker");
private final String _configPrefix;
- private AuditConfig _currentConfig = new AuditConfig();
+ private volatile AuditConfig _currentConfig = new AuditConfig();
@Inject
public AuditConfigManager(ServiceRole serviceRole) {
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditServiceBinder.java b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditServiceBinder.java
index 1a9de50d8ce..224df34d42f 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditServiceBinder.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditServiceBinder.java
@@ -21,6 +21,7 @@ package org.apache.pinot.common.audit;
import javax.inject.Singleton;
import org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
import org.apache.pinot.common.metrics.AbstractMetrics;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigProvider;
import org.apache.pinot.spi.services.ServiceRole;
import org.glassfish.hk2.utilities.binding.AbstractBinder;
@@ -28,13 +29,18 @@ import org.glassfish.hk2.utilities.binding.AbstractBinder;
/// HK2 Binder for audit-related services.
/// Handles initialization and registration of audit components.
public class AuditServiceBinder extends AbstractBinder {
- private final DefaultClusterConfigChangeHandler _clusterConfigChangeHandler;
+ private final PinotClusterConfigProvider _clusterConfigProvider;
private final ServiceRole _serviceRole;
private final AbstractMetrics<?, ?, ?, ?> _componentMetrics;
public AuditServiceBinder(DefaultClusterConfigChangeHandler clusterConfigChangeHandler, ServiceRole serviceRole,
AbstractMetrics<?, ?, ?, ?> componentMetrics) {
- _clusterConfigChangeHandler = clusterConfigChangeHandler;
+ this((PinotClusterConfigProvider) clusterConfigChangeHandler, serviceRole, componentMetrics);
+ }
+
+ public AuditServiceBinder(PinotClusterConfigProvider clusterConfigProvider, ServiceRole serviceRole,
+ AbstractMetrics<?, ?, ?, ?> componentMetrics) {
+ _clusterConfigProvider = clusterConfigProvider;
_serviceRole = serviceRole;
_componentMetrics = componentMetrics;
}
@@ -46,8 +52,8 @@ public class AuditServiceBinder extends AbstractBinder {
AuditConfigManager auditConfigManager = new AuditConfigManager(_serviceRole);
- // Register with cluster config change handler
- _clusterConfigChangeHandler.registerClusterConfigChangeListener(auditConfigManager);
+ // Register with the committed cluster config provider.
+ _clusterConfigProvider.registerClusterConfigChangeListener(auditConfigManager);
bind(auditConfigManager).to(AuditConfigManager.class);
bindAsContract(AuditMetrics.class).in(Singleton.class);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/ControllerAdminApiApplication.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/ControllerAdminApiApplication.java
index aad9e5e67ac..b4599d10b2e 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/ControllerAdminApiApplication.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/ControllerAdminApiApplication.java
@@ -23,6 +23,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.swagger.jaxrs.listing.SwaggerSerializers;
import java.io.IOException;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
@@ -37,6 +38,7 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.ext.ContextResolver;
import javax.ws.rs.ext.Provider;
import org.apache.pinot.common.audit.AuditLogFilter;
+import org.apache.pinot.common.coordination.CoordinationBackend;
import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.swagger.SwaggerApiListingResource;
@@ -75,6 +77,8 @@ public class ControllerAdminApiApplication extends ResourceConfig {
public static final String START_TIME = "controllerStartTime";
+ public static final String COORDINATION_BACKEND = "controllerCoordinationBackend";
+
private final String _controllerResourcePackages;
private final boolean _useHttps;
private final boolean _enableSwagger;
@@ -82,24 +86,43 @@ public class ControllerAdminApiApplication extends ResourceConfig {
private final ThreadPoolExecutor _managedAsyncExecutor;
public ControllerAdminApiApplication(ControllerConf conf) {
+ this(conf, true, Set.of());
+ }
+
+ /// Registers an explicit HTTP resource set without legacy package scanning, service discovery or multipart upload
+ /// providers. Swagger is disabled on this path because the existing package scanner uses a shared global registry
+ /// and cannot safely describe an isolated resource set alongside other controllers in the same JVM.
+ public ControllerAdminApiApplication(ControllerConf conf, Set<Class<?>> resourceClasses) {
+ this(conf, false, Set.copyOf(resourceClasses));
+ }
+
+ private ControllerAdminApiApplication(ControllerConf conf, boolean scanLegacyPackages,
+ Set<Class<?>> resourceClasses) {
super();
property(PINOT_CONFIGURATION, conf);
+ property(COORDINATION_BACKEND, CoordinationBackend.from(conf));
_controllerResourcePackages = conf.getControllerResourcePackages();
- packages(_controllerResourcePackages);
+ if (scanLegacyPackages) {
+ packages(_controllerResourcePackages);
+ } else {
+ registerClasses(resourceClasses);
+ }
// TODO See ControllerResponseFilter
// register(new LoggingFeature());
_useHttps = Boolean.parseBoolean(conf.getProperty(ControllerConf.CONSOLE_SWAGGER_USE_HTTPS));
- _enableSwagger = conf.isEnableSwagger();
- if (conf.getProperty(CommonConstants.Controller.CONTROLLER_SERVICE_AUTO_DISCOVERY, false)) {
+ _enableSwagger = scanLegacyPackages && conf.isEnableSwagger();
+ if (scanLegacyPackages && conf.getProperty(CommonConstants.Controller.CONTROLLER_SERVICE_AUTO_DISCOVERY, false)) {
register(ServiceAutoDiscoveryFeature.class);
}
register(JacksonFeature.class);
- register(MultiPartFeature.class);
- register(new MultiPartTempDirResolver());
- register(new MultiPartTempDirGuard());
- register(SwaggerApiListingResource.class);
- register(SwaggerSerializers.class);
+ if (scanLegacyPackages) {
+ register(MultiPartFeature.class);
+ register(new MultiPartTempDirResolver());
+ register(new MultiPartTempDirGuard());
+ register(SwaggerApiListingResource.class);
+ register(SwaggerSerializers.class);
+ }
register(new CorsFilter());
register(AuthenticationFilter.class);
register(AuditLogFilter.class);
@@ -145,11 +168,13 @@ public class ControllerAdminApiApplication extends ResourceConfig {
}
public void stop() {
- if (!_httpServer.isStarted()) {
- return;
+ try {
+ if (_httpServer != null) {
+ _httpServer.shutdownNow();
+ }
+ } finally {
+ _managedAsyncExecutor.shutdown();
}
- _httpServer.shutdownNow();
- _managedAsyncExecutor.shutdown();
}
private class CorsFilter implements ContainerResponseFilter {
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AccessControlFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AccessControlFactory.java
index 32d2f7c4a82..3f35a0c136e 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AccessControlFactory.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AccessControlFactory.java
@@ -43,5 +43,12 @@ public interface AccessControlFactory {
init(configuration);
}
+ /// Initializes a factory that explicitly supports operation without Helix services. Implementations must opt in;
+ /// falling back to the configuration-only initializer could silently omit a factory's metadata-backed policies.
+ default void initWithoutHelix(PinotConfiguration configuration) throws IOException {
+ throw new IOException("Access control factory does not support initialization without Helix: "
+ + getClass().getName());
+ }
+
AccessControl create();
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AllowAllAccessFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AllowAllAccessFactory.java
index a454f028dfa..e920a416632 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AllowAllAccessFactory.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/AllowAllAccessFactory.java
@@ -18,10 +18,22 @@
*/
package org.apache.pinot.controller.api.access;
+import java.io.IOException;
+import org.apache.pinot.spi.env.PinotConfiguration;
+
+
public class AllowAllAccessFactory implements AccessControlFactory {
private static final AccessControl ALLOW_ALL_ACCESS = new AccessControl() {
};
+ @Override
+ public void initWithoutHelix(PinotConfiguration configuration) throws IOException {
+ if (getClass() != AllowAllAccessFactory.class) {
+ AccessControlFactory.super.initWithoutHelix(configuration);
+ }
+ init(configuration);
+ }
+
@Override
public AccessControl create() {
return ALLOW_ALL_ACCESS;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactory.java
index 9d8d3c615e5..2fcb6f98681 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactory.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactory.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.api.access;
+import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -59,6 +60,14 @@ public class BasicAuthAccessControlFactory implements AccessControlFactory {
BasicAuthPrincipalUtils.extractBasicAuthPrincipals(configuration, PREFIX));
}
+ @Override
+ public void initWithoutHelix(PinotConfiguration configuration) throws IOException {
+ if (getClass() != BasicAuthAccessControlFactory.class) {
+ AccessControlFactory.super.initWithoutHelix(configuration);
+ }
+ init(configuration);
+ }
+
@Override
public AccessControl create() {
return _accessControl;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResource.java
index fc6fb0eee2c..fd3b0fc5479 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResource.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.api.resources;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
import io.swagger.annotations.ApiOperation;
@@ -34,15 +35,19 @@ import javax.ws.rs.NotAuthorizedException;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.Application;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
+import org.apache.pinot.common.coordination.CoordinationBackend;
+import org.apache.pinot.controller.api.ControllerAdminApiApplication;
import org.apache.pinot.controller.api.access.AccessControl;
import org.apache.pinot.controller.api.access.AccessControlFactory;
import org.apache.pinot.controller.api.access.AccessType;
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.utils.JsonUtils;
import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_KEY;
@@ -60,6 +65,9 @@ public class PinotControllerAuthResource {
@Context
HttpHeaders _httpHeaders;
+ @Context
+ private Application _application;
+
/// Verify a token is both authenticated and authorized to perform an operation.
///
/// @param tableName table name (optional)
@@ -113,7 +121,8 @@ public class PinotControllerAuthResource {
}
/// Provide the auth workflow configuration for the Pinot UI to perform user authentication. Currently supports NONE
- /// (no auth) and BASIC (basic auth with username and password)
+ /// (no auth) and BASIC (basic auth with username and password). The backend is captured from startup configuration,
+ /// independently of mutable cluster configuration and authentication-provider extensions.
///
/// @return auth workflow info/configuration
@GET
@@ -122,7 +131,11 @@ public class PinotControllerAuthResource {
@Produces(MediaType.APPLICATION_JSON)
@ApiOperation(value = "Retrieve auth workflow info")
@ApiResponses(value = {@ApiResponse(code = 200, message = "Auth workflow info provided")})
- public AccessControl.AuthWorkflowInfo info() {
- return _accessControlFactory.create().getAuthWorkflowInfo();
+ public ObjectNode info() {
+ ObjectNode info = (ObjectNode) JsonUtils.objectToJsonNode(_accessControlFactory.create().getAuthWorkflowInfo());
+ CoordinationBackend backend =
+ (CoordinationBackend) _application.getProperties().get(ControllerAdminApiApplication.COORDINATION_BACKEND);
+ info.put("coordinationBackend", backend.name());
+ return info;
}
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java
new file mode 100644
index 00000000000..cf4703543dc
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerMetadata.java
@@ -0,0 +1,496 @@
+/**
+ * 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.util.ArrayList;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+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 javax.annotation.Nullable;
+import org.apache.pinot.common.config.provider.RecoverableClusterConfigCache;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.evaluator.FunctionEvaluatorFactory;
+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.MetadataProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Acknowledge;
+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.ConfigTransaction;
+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.Mutation;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RegisterClient;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+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.MetadataProtocol.StreamTipResult;
+import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol;
+import org.apache.pinot.coordination.ratis.metadata.OperationProtocol;
+import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
+import org.apache.pinot.segment.local.utils.SchemaUtils;
+import org.apache.pinot.segment.local.utils.TableConfigUtils;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigProvider;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableConfigValidatorRegistry;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+
+/// Bounded, quorum-backed controller reads. Each REST read captures a new coherent image; the separately refreshed
+/// cluster config provider exists only for local listeners such as audit configuration, never as a REST fallback.
+/// Conditional writes serialize one durable controller stream and validate one coherent image. Native configurations
+/// are literal: environment/system-property templates are unsupported. Lifecycle and provider refreshes are
+/// serialized. The owner closes this facade before closing its MetadataAccess.
+public final class NativeControllerMetadata implements PinotClusterConfigProvider, AutoCloseable {
+ private static final FeedScope CONFIG_SCOPE =
+ new FeedScope(Set.of(RecordKind.CLUSTER_CONFIG), "controller-http-config");
+ private static final FeedScope BROWSE_SCOPE = new FeedScope(EnumSet.of(RecordKind.CLUSTER_CONFIG, RecordKind.SCHEMA,
+ RecordKind.TABLE_CONFIG, RecordKind.MEMBERSHIP_CONTROL, RecordKind.PARTICIPANT,
+ RecordKind.OFFLINE_SEGMENT_CONTENT, RecordKind.OFFLINE_REPLICA_ASSIGNMENT, RecordKind.CONTROLLER_OPERATION),
+ "controller-http-browse");
+ private final MetadataAccess _access;
+ private final UUID _history;
+ private final String _clusterName;
+ private final int _maxRecords;
+ private final int _maxBytes;
+ private final long _timeoutMillis;
+ private final long _refreshMillis;
+ private final RecoverableClusterConfigCache _clusterConfig = new RecoverableClusterConfigCache();
+ private final Object _writeLock = new Object();
+ private final String _clientId;
+ private final UUID _clientIncarnation;
+ private final boolean _disableGroovy;
+ private final boolean _enforcePoolBasedAssignment;
+ private final int _minReplicas;
+ @Nullable
+ private ScheduledExecutorService _refresher;
+ private volatile boolean _started;
+ private volatile boolean _closed;
+ @Nullable
+ private volatile String _clusterConfigFailure;
+ private long _configRevision = -1;
+
+ public NativeControllerMetadata(MetadataAccess access, UUID history, ControllerConf conf) {
+ _access = requireNonNull(access, "access");
+ _history = requireNonNull(history, "history");
+ _clientId = "controller-http/" + NativeMetadataConfig.required(conf, "id");
+ _clientIncarnation = UUID.nameUUIDFromBytes((_history + "/" + _clientId).getBytes(StandardCharsets.UTF_8));
+ _disableGroovy = conf.isDisableIngestionGroovy();
+ _enforcePoolBasedAssignment = conf.isEnforcePoolBasedAssignmentEnabled();
+ _minReplicas = conf.getDefaultTableMinReplicas();
+ String name = conf.getHelixClusterName();
+ _clusterName = name == null || name.isBlank() ? NativeMetadataConfig.required(conf, "clusterId") : name;
+ String prefix = NativeMetadataConfig.PREFIX + "controller.http.";
+ _maxRecords = conf.getProperty(prefix + "maxRecords", 4096);
+ _maxBytes = conf.getProperty(prefix + "maxBytes", 2 * MetadataProtocol.MAX_COMMAND_BYTES);
+ _timeoutMillis = conf.getProperty(prefix + "timeoutMs", 5000L);
+ _refreshMillis = conf.getProperty(prefix + "configRefreshMs", 1000L);
+ checkArgument(_maxRecords > 0 && _maxRecords <= MetadataProtocol.MAX_IMAGE_RECORDS,
+ "Native HTTP record limit is outside supported bounds");
+ checkArgument(_maxBytes >= 1024 && _maxBytes <= MetadataProtocol.MAX_IMAGE_BYTES,
+ "Native HTTP byte limit is outside supported bounds");
+ checkArgument(_timeoutMillis > 0 && _timeoutMillis <= 60_000 && _refreshMillis > 0 && _refreshMillis <= 60_000,
+ "Native HTTP read and config refresh intervals must be between 1 and 60000 milliseconds");
+ }
+
+ public synchronized void start() throws IOException {
+ checkState(!_started && !_closed, "Native controller metadata has already started or closed");
+ refreshClusterConfig();
+ _refresher = Executors.newSingleThreadScheduledExecutor(runnable -> {
+ Thread thread = new Thread(runnable, "native-controller-cluster-config");
+ thread.setDaemon(true);
+ return thread;
+ });
+ _started = true;
+ _refresher.scheduleWithFixedDelay(this::refreshSafely, _refreshMillis, _refreshMillis, TimeUnit.MILLISECONDS);
+ }
+
+ private void refreshSafely() {
+ try {
+ refreshClusterConfig();
+ } catch (Exception e) {
+ _clusterConfigFailure = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
+ }
+ }
+
+ /// Serialized complete refreshes preserve changed/unset listener retries after a partial callback failure.
+ synchronized void refreshClusterConfig() throws IOException {
+ checkState(!_closed, "Native controller metadata is closed");
+ CatalogImage image = image(CONFIG_SCOPE);
+ if (image.cursor().index() < _configRevision) {
+ throw new IOException("Native cluster config image regressed");
+ }
+ Map<String, String> values = clusterConfig(image.records());
+ try {
+ if (_configRevision < 0 || _clusterConfigFailure != null || !values.equals(_clusterConfig.getClusterConfigs())) {
+ _clusterConfig.updateClusterConfigs(values);
+ }
+ _configRevision = image.cursor().index();
+ _clusterConfigFailure = null;
+ } catch (RuntimeException e) {
+ _clusterConfigFailure = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
+ throw e;
+ }
+ }
+
+ @Nullable
+ public String clusterConfigFailure() {
+ return _clusterConfigFailure;
+ }
+
+ public String clusterName() {
+ return _clusterName;
+ }
+
+ public Snapshot snapshot() throws IOException {
+ if (!_started || _closed) {
+ throw new IOException("Native controller metadata is not running");
+ }
+ CatalogImage image = image(BROWSE_SCOPE);
+ Map<String, CatalogRecord> schemas = new HashMap<>();
+ Map<String, CatalogRecord> tables = new HashMap<>();
+ Map<String, ParticipantState> participants = new HashMap<>();
+ ControlState control = null;
+ Set<String> nativeTables = new HashSet<>();
+ CatalogRecord clusterRecord = null;
+ try {
+ for (CatalogRecord record : image.records()) {
+ switch (record.kind()) {
+ case SCHEMA -> schemas.put(record.name(), record);
+ case TABLE_CONFIG -> tables.put(record.name(), record);
+ case MEMBERSHIP_CONTROL -> {
+ if (!MembershipProtocol.CONTROL_NAME.equals(record.name()) || control != null) {
+ throw new IOException("Invalid native membership control record");
+ }
+ control = MembershipProtocol.decodeControl(record.value());
+ }
+ case PARTICIPANT -> {
+ ParticipantState participant = MembershipProtocol.decodeParticipant(record.value());
+ if (participant.participant() == null || !_history.equals(participant.participant().historyId())
+ || !record.name().equals(participant.participant().participantId())) {
+ throw new IOException("Invalid native participant record");
+ }
+ participants.put(record.name(), participant);
+ }
+ case CLUSTER_CONFIG -> clusterRecord = record;
+ case OFFLINE_SEGMENT_CONTENT -> nativeTables.add(
+ OfflineSegmentProtocol.decodeContent(record.value()).tableName());
+ case OFFLINE_REPLICA_ASSIGNMENT -> nativeTables.add(
+ OfflineSegmentProtocol.decodeAssignment(record.value()).tableName());
+ case CONTROLLER_OPERATION -> nativeTables.add(
+ OperationProtocol.decodeOperation(record.value()).input().tableName());
+ default -> throw new IOException("Unexpected native HTTP record family");
+ }
+ }
+ if (control == null || control.controlEpoch() == null || control.policy() == null
+ || !_history.equals(control.controlEpoch().historyId())) {
+ throw new IOException("Native membership control is not initialized");
+ }
+ return new Snapshot(new Revision(_history, image.cursor().index()), clusterConfig(image.records()), schemas,
+ tables, control, participants, clusterRecord, nativeTables);
+ } catch (IllegalArgumentException | NullPointerException e) {
+ throw new IOException("Malformed native controller metadata image", e);
+ }
+ }
+
+ /// Resolves the bounded durable controller stream before issuing one conditional transaction. An IOException from
+ /// execute remains an unknown outcome: this method never changes its precondition or automatically resubmits it.
+ public Result commit(Snapshot snapshot, Mutation mutation) throws IOException {
+ synchronized (_writeLock) {
+ if (!_started || _closed) {
+ throw new IOException("Native controller metadata is not running");
+ }
+ Result registration = _access.execute(new RegisterClient(_history, _clientId, _clientIncarnation, 0),
+ _timeoutMillis);
+ if (registration.status() != Status.OK || registration.clientEpoch() == null || registration.clientEpoch() != 1) {
+ throw new IOException("Native HTTP writer registration is unavailable or fenced");
+ }
+ StreamTipResult tip = _access.query(new GetStreamTip(_history, _clientId, 1, "config"), _timeoutMillis);
+ if (tip.status() != Status.OK || tip.clientEpoch() != 1 || tip.sequence() < 0 || tip.floor() < 0
+ || tip.floor() > tip.sequence() || tip.sequence() == Long.MAX_VALUE) {
+ throw new IOException("Native HTTP writer stream is unavailable or exhausted");
+ }
+ if (tip.sequence() > tip.floor()) {
+ if (tip.outcome() == null) {
+ throw new IOException("Native HTTP writer has an unresolved outcome");
+ }
+ Result acknowledged = _access.execute(new Acknowledge(_history, _clientId, 1, "config", tip.sequence()),
+ _timeoutMillis);
+ if (acknowledged.status() != Status.OK) {
+ throw new IOException("Native HTTP writer could not acknowledge its resolved outcome");
+ }
+ }
+ return _access.execute(new ConfigTransaction(_history, _clientId, 1, "config", tip.sequence() + 1,
+ snapshot.revision(), List.of(mutation)), _timeoutMillis);
+ }
+ }
+
+ /// Validates only metadata changes whose native segment lifecycle has not started. Existing assignments retain
+ /// table/schema versions, so changing those versions requires reconciliation that this HTTP surface does not run.
+ public void validateSchema(Snapshot snapshot, Schema schema, @Nullable Schema previous) throws IOException {
+ requireLiteralConfig(schema);
+ String name = schema.getSchemaName();
+ checkArgument(name != null && !name.isBlank() && TableNameBuilder.getTableTypeFromTableName(name) == null
+ && !name.contains(TableConfig.TABLE_NAME_FORBIDDEN_SUBSTRING),
+ "Schema requires a valid name");
+ List<TableConfig> tables = new ArrayList<>();
+ for (TableType type : TableType.values()) {
+ String typed = TableNameBuilder.forType(type).tableNameWithType(name);
+ requireNoNativeState(snapshot, typed);
+ CatalogRecord table = snapshot.tables().get(typed);
+ if (table != null) {
+ if (type != TableType.OFFLINE) {
+ throw new UnsupportedOperationException("Schema changes for realtime tables are not supported");
+ }
+ tables.add(JsonUtils.jsonNodeToObject(table.value(), TableConfig.class));
+ }
+ }
+ validateSchemaTransforms(schema);
+ validateNameCollision(snapshot.schemas(), name, ignoreCase(snapshot.clusterConfig()));
+ // The SchemaUtils table-compatibility overload reads legacy process-global validation switches. Validate the
+ // schema separately, then each affected table using the explicit native component policy instead.
+ SchemaUtils.validate(schema, List.of(), ignoreCase(snapshot.clusterConfig()), previous);
+ for (TableConfig table : tables) {
+ TableConfigUtils.validate(table, schema, null, table, _disableGroovy, _enforcePoolBasedAssignment);
+ }
+ checkArgument(previous == null || schema.isBackwardCompatibleWith(previous),
+ "Schema update is not backward compatible");
+ }
+
+ public void validateTable(Snapshot snapshot, TableConfig table, @Nullable TableConfig previous) throws IOException {
+ requireLiteralConfig(table);
+ String name = table.getTableName();
+ checkArgument(name != null && !name.isBlank()
+ && TableNameBuilder.getTableTypeFromTableName(TableNameBuilder.extractRawTableName(name)) == null,
+ "Table requires an unambiguous raw name");
+ checkArgument(table.getValidationConfig() != null, "Table validation configuration is required");
+ requireNoNativeState(snapshot, name);
+ if (table.getTableType() != TableType.OFFLINE || table.isDimTable() || table.isUpsertEnabled()
+ || table.getDedupConfig() != null || table.getTaskConfig() != null
+ || table.getTierConfigsList() != null && !table.getTierConfigsList().isEmpty()
+ || table.getTunerConfigsList() != null && !table.getTunerConfigsList().isEmpty()
+ || table.getInstanceAssignmentConfigMap() != null && !table.getInstanceAssignmentConfigMap().isEmpty()
+ || table.getInstancePartitionsMap() != null && !table.getInstancePartitionsMap().isEmpty()
+ || table.getSegmentAssignmentConfigMap() != null && !table.getSegmentAssignmentConfigMap().isEmpty()
+ || table.getMaterializedViewTaskConfigs() != null && !table.getMaterializedViewTaskConfigs().isEmpty()
+ || table.getValidationConfig().getPeerSegmentDownloadScheme() != null) {
+ throw new UnsupportedOperationException("Native HTTP supports ordinary OFFLINE table configurations without "
+ + "tasks, tiers, upsert, dedup, tuning, peer download or Helix assignment options");
+ }
+ String raw = TableNameBuilder.extractRawTableName(name);
+ if (snapshot.tables().containsKey(TableNameBuilder.REALTIME.tableNameWithType(raw))) {
+ throw new UnsupportedOperationException("Native HTTP does not manage hybrid tables");
+ }
+ CatalogRecord schemaRecord = snapshot.schemas().get(raw);
+ checkArgument(schemaRecord != null, "Schema not found for table: %s", name);
+ Schema schema = JsonUtils.jsonNodeToObject(schemaRecord.value(), Schema.class);
+ validateSchema(snapshot, schema, schema);
+ validateNameCollision(snapshot.tables(), name, ignoreCase(snapshot.clusterConfig()));
+ TableConfigUtils.validateTableName(table);
+ TableConfigUtils.ensureMinReplicas(table, _minReplicas);
+ TableConfigUtils.validate(table, schema, null, previous, _disableGroovy, _enforcePoolBasedAssignment);
+ TableConfigValidatorRegistry.validate(table, schema);
+ checkArgument(previous == null || TableConfigUtils.validateBackwardCompatibility(table, previous).isEmpty(),
+ "Table configuration update is not backward compatible");
+ }
+
+ public void validateSchemaDeletion(Snapshot snapshot, String name) {
+ for (TableType type : TableType.values()) {
+ String typed = TableNameBuilder.forType(type).tableNameWithType(name);
+ requireNoNativeState(snapshot, typed);
+ checkState(!snapshot.tables().containsKey(typed), "Schema is still referenced by table: %s", typed);
+ }
+ }
+
+ public void validateClusterConfig(Snapshot snapshot, Map<String, String> values) throws IOException {
+ requireLiteralConfig(values);
+ if (!snapshot.nativeTables().isEmpty()) {
+ throw new UnsupportedOperationException("Cluster config changes require native table reconciliation");
+ }
+ String ignoreCase = values.get(CommonConstants.Helix.ENABLE_CASE_INSENSITIVE_KEY);
+ checkArgument(ignoreCase == null || "true".equalsIgnoreCase(ignoreCase) || "false".equalsIgnoreCase(ignoreCase),
+ "Case-insensitive configuration must be true or false");
+ if (ignoreCase(values)) {
+ for (CatalogRecord record : snapshot.schemas().values()) {
+ validateNameCollision(snapshot.schemas(), record.name(), true);
+ Schema schema = JsonUtils.jsonNodeToObject(record.value(), Schema.class);
+ validateSchemaTransforms(schema);
+ SchemaUtils.validate(schema, List.of(), true, schema);
+ }
+ for (CatalogRecord record : snapshot.tables().values()) {
+ validateNameCollision(snapshot.tables(), record.name(), true);
+ }
+ }
+ }
+
+ private static void requireLiteralConfig(Object value) throws IOException {
+ // Unlike the Helix provider, native participants do not resolve host-local templates on config reads.
+ // Even a template with a default can differ across hosts and cannot be validated as committed configuration.
+ if (JsonUtils.objectToString(value).contains("${")) {
+ throw new UnsupportedOperationException("Native metadata requires literal configuration; environment and "
+ + "system-property templates are not supported");
+ }
+ }
+
+ // Inspect legacy schema transforms before SchemaUtils can compile them, even though the field is deprecated.
+ @SuppressWarnings("deprecation")
+ private void validateSchemaTransforms(Schema schema) {
+ if (_disableGroovy) {
+ schema.getAllFieldSpecs().forEach(field -> {
+ String expression = field.getTransformFunction();
+ checkArgument(expression == null || !FunctionEvaluatorFactory.isGroovyExpression(expression),
+ "Groovy schema transforms are disabled");
+ });
+ }
+ }
+
+ private static boolean ignoreCase(Map<String, String> values) {
+ return Boolean.parseBoolean(values.getOrDefault(CommonConstants.Helix.ENABLE_CASE_INSENSITIVE_KEY,
+ Boolean.toString(CommonConstants.Helix.DEFAULT_ENABLE_CASE_INSENSITIVE)));
+ }
+
+ private static void validateNameCollision(Map<String, CatalogRecord> records, String name, boolean ignoreCase) {
+ if (ignoreCase) {
+ checkArgument(records.keySet().stream().noneMatch(existing -> !existing.equals(name)
+ && existing.equalsIgnoreCase(name)), "Case-insensitive metadata name already exists: %s", name);
+ }
+ }
+
+ private static void requireNoNativeState(Snapshot snapshot, String name) {
+ if (snapshot.nativeTables().contains(name)) {
+ throw new UnsupportedOperationException(
+ "Table has native segment or operation state; reconciliation is required");
+ }
+ }
+
+ private CatalogImage image(FeedScope scope) throws IOException {
+ CatalogImage image = _access.image(scope, _maxRecords, _maxBytes, _timeoutMillis);
+ if (image == null || image.cursor() == null || !_history.equals(image.cursor().historyId())
+ || image.cursor().index() < 0 || !scope.scopeId().equals(image.cursor().scopeId())
+ || image.records() == null || image.records().size() > _maxRecords) {
+ throw new IOException("Invalid native controller metadata image");
+ }
+ Set<UUID> ids = new HashSet<>();
+ Set<String> names = new HashSet<>();
+ for (CatalogRecord record : image.records()) {
+ if (record == null || record.kind() == null || !scope.kinds().contains(record.kind())
+ || record.name() == null || record.name().isBlank() || record.objectId() == null || record.version() <= 0
+ || record.schemaVersion() != MetadataProtocol.FORMAT_VERSION || record.deleted() || record.value() == null
+ || !record.value().isObject() || record.lastModifiedIndex() <= 0
+ || record.lastModifiedIndex() > image.cursor().index() || !ids.add(record.objectId())
+ || !names.add(record.kind() + "/" + record.name())) {
+ throw new IOException("Invalid or duplicate native controller metadata record");
+ }
+ }
+ if (JsonUtils.objectToBytes(image).length > _maxBytes) {
+ throw new IOException("Native controller metadata image exceeds its reserved byte limit");
+ }
+ return image;
+ }
+
+ private static Map<String, String> clusterConfig(List<CatalogRecord> records) throws IOException {
+ Map<String, String> values = new HashMap<>();
+ boolean found = false;
+ for (CatalogRecord record : records) {
+ if (record.kind() == RecordKind.CLUSTER_CONFIG) {
+ if (found || !MetadataProtocol.CLUSTER_CONFIG_NAME.equals(record.name())) {
+ throw new IOException("Invalid native cluster config record");
+ }
+ found = true;
+ for (var entry : record.value().properties()) {
+ if (!entry.getValue().isTextual()) {
+ throw new IOException("Native cluster config values must be strings");
+ }
+ values.put(entry.getKey(), entry.getValue().textValue());
+ }
+ }
+ }
+ return Map.copyOf(values);
+ }
+
+ @Override
+ public Map<String, String> getClusterConfigs() {
+ checkState(_started && !_closed, "Native cluster config provider is not running");
+ return _clusterConfig.getClusterConfigs();
+ }
+
+ @Override
+ public boolean registerClusterConfigChangeListener(PinotClusterConfigChangeListener listener) {
+ checkState(_started && !_closed, "Native cluster config provider is not running");
+ try {
+ return _clusterConfig.registerClusterConfigChangeListener(listener);
+ } catch (RuntimeException e) {
+ _clusterConfigFailure = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
+ throw e;
+ }
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ ScheduledExecutorService refresher;
+ synchronized (this) {
+ if (_closed) {
+ return;
+ }
+ _closed = true;
+ refresher = _refresher;
+ if (refresher != null) {
+ refresher.shutdown();
+ }
+ }
+ if (refresher != null && !refresher.awaitTermination(_timeoutMillis + 1000, TimeUnit.MILLISECONDS)) {
+ throw new IllegalStateException("Native cluster config refresher did not stop within its read deadline");
+ }
+ }
+
+ /// All fields come from one image. Catalog records and membership records are detached immutable protocol values.
+ public record Snapshot(Revision revision, Map<String, String> clusterConfig, Map<String, CatalogRecord> schemas,
+ Map<String, CatalogRecord> tables, ControlState control,
+ Map<String, ParticipantState> participants, @Nullable CatalogRecord clusterRecord,
+ Set<String> nativeTables) {
+ public Snapshot {
+ clusterConfig = Map.copyOf(clusterConfig);
+ schemas = Map.copyOf(schemas);
+ tables = Map.copyOf(tables);
+ participants = Map.copyOf(participants);
+ nativeTables = Set.copyOf(nativeTables);
+ }
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerResource.java
new file mode 100644
index 00000000000..9b52b439b33
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerResource.java
@@ -0,0 +1,701 @@
+/**
+ * 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.StreamReadConstraints;
+import com.fasterxml.jackson.core.StreamReadFeature;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import javax.annotation.Nullable;
+import javax.inject.Inject;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.DefaultValue;
+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.AccessControl;
+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.controller.coordination.NativeControllerMetadata.Snapshot;
+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;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
+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.RecordKind;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status;
+import org.apache.pinot.core.auth.Actions;
+import org.apache.pinot.core.auth.Authorize;
+import org.apache.pinot.core.auth.ManualAuthorization;
+import org.apache.pinot.core.auth.TargetType;
+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.SchemaInfo;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+
+
+/// Native metadata routes with the normal controller JSON shapes and authorization actions. All successful
+/// responses identify one fresh committed image. Membership is projected explicitly, without Helix instance or
+/// ExternalView emulation; a live lease does not certify query serving. Conditional writes publish configuration
+/// only, and reject changes requiring native segment reconciliation. No query or table cleanup route is registered.
+@Path("/")
+@Produces(MediaType.APPLICATION_JSON)
+public final class NativeControllerResource {
+ public static final String REVISION_HEADER = "X-Pinot-Metadata-Revision";
+ private static final ObjectMapper REQUEST_JSON = JsonMapper.builder(JsonFactory.builder()
+ .enable(StreamReadFeature.STRICT_DUPLICATE_DETECTION)
+ .streamReadConstraints(StreamReadConstraints.builder().maxNestingDepth(64)
+ .maxStringLength(MetadataProtocol.MAX_COMMAND_BYTES).build()).build())
+ .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+ .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS).build();
+ private final NativeControllerMetadata _metadata;
+ private final AccessControlFactory _accessControlFactory;
+
+ @Inject
+ public NativeControllerResource(NativeControllerMetadata metadata, AccessControlFactory accessControlFactory) {
+ _metadata = metadata;
+ _accessControlFactory = accessControlFactory;
+ }
+
+ @GET
+ @Path("/cluster/info")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_CLUSTER_CONFIG)
+ public Response clusterInfo(@Context HttpHeaders headers) {
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_CLUSTER_CONFIG, "/cluster/info");
+ return response(snapshot, Map.of("clusterName", _metadata.clusterName()));
+ }
+
+ @GET
+ @Path("/cluster/configs")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_CLUSTER_CONFIG)
+ public Response clusterConfigs(@Context HttpHeaders headers) {
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_CLUSTER_CONFIG, "/cluster/configs");
+ return tagged(snapshot, snapshot.clusterConfig(), snapshot.clusterRecord());
+ }
+
+ @GET
+ @Path("/schemas")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_SCHEMA)
+ public Response schemas(@Context HttpHeaders headers) {
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_SCHEMA, "/schemas");
+ List<String> names = snapshot.schemas().keySet().stream().filter(NativeControllerResource::supportedRawName)
+ .filter(name -> inDatabase(name, headers))
+ .filter(name -> allowed(name, headers, Actions.Table.GET_SCHEMA, "/schemas")).sorted().toList();
+ return response(snapshot, names);
+ }
+
+ @GET
+ @Path("/schemas/info")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_SCHEMAS_INFO)
+ public Response schemasInfo(@Context HttpHeaders headers) {
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_SCHEMAS_INFO, "/schemas/info");
+ List<SchemaInfo> schemas = new ArrayList<>();
+ snapshot.schemas().keySet().stream().sorted().filter(NativeControllerResource::supportedRawName)
+ .filter(name -> inDatabase(name, headers))
+ .filter(name -> allowed(name, headers, Actions.Table.GET_SCHEMA, "/schemas/info"))
+ .forEach(name -> schemas.add(new SchemaInfo(schema(snapshot.schemas().get(name)))));
+ return response(snapshot, Map.of("schemasInfo", schemas));
+ }
+
+ @GET
+ @Path("/schemas/{schemaName}")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.TABLE, paramName = "schemaName", action = Actions.Table.GET_SCHEMA)
+ public Response schema(@PathParam("schemaName") String name, @Context HttpHeaders headers) {
+ name = translateSchemaName(name, headers);
+ Snapshot snapshot = read(headers, name, Actions.Table.GET_SCHEMA, "/schemas/" + name);
+ CatalogRecord record = snapshot.schemas().get(name);
+ if (record == null) {
+ throw new WebApplicationException("Schema not found", Response.Status.NOT_FOUND);
+ }
+ return tagged(snapshot, record.value(), record);
+ }
+
+ @GET
+ @Path("/tables")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_TABLE)
+ public Response tables(@QueryParam("type") String type, @QueryParam("taskType") String taskType,
+ @QueryParam("sortType") String sortType, @DefaultValue("true") @QueryParam("sortAsc") boolean sortAsc,
+ @Context HttpHeaders headers) {
+ if (taskType != null && !taskType.isBlank() || sortType != null && !"name".equalsIgnoreCase(sortType)) {
+ throw new WebApplicationException("Native table listing supports name sorting and no task filter",
+ Response.Status.NOT_IMPLEMENTED);
+ }
+ boolean dimension = "dimension".equalsIgnoreCase(type);
+ TableType tableType = dimension ? TableType.OFFLINE : tableType(type);
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_TABLE, "/tables");
+ List<String> names = snapshot.tables().values().stream().filter(record -> inDatabase(record.name(), headers))
+ .filter(record -> tableType == null || TableNameBuilder.getTableTypeFromTableName(record.name()) == tableType)
+ .filter(record -> !dimension || table(record).isDimTable())
+ .filter(record -> allowed(record.name(), headers, Actions.Table.GET_TABLE_CONFIG, "/tables"))
+ .map(record -> tableType == null ? TableNameBuilder.extractRawTableName(record.name()) : record.name())
+ .distinct().sorted(sortAsc ? Comparator.naturalOrder() : Comparator.reverseOrder()).toList();
+ return response(snapshot, Map.of("tables", names));
+ }
+
+ @GET
+ @Path("/tables/{tableName}")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_TABLE_CONFIG)
+ public Response table(@PathParam("tableName") String name, @QueryParam("type") String type,
+ @Context HttpHeaders headers) {
+ name = translateTableName(name, headers);
+ TableType requested = tableType(type);
+ TableType suffix = TableNameBuilder.getTableTypeFromTableName(name);
+ if (requested != null && suffix != null && requested != suffix) {
+ throw new WebApplicationException("Table name and type disagree", Response.Status.BAD_REQUEST);
+ }
+ requested = requested == null ? suffix : requested;
+ String raw = TableNameBuilder.extractRawTableName(name);
+ Snapshot snapshot = read(headers, raw, Actions.Table.GET_TABLE_CONFIG, "/tables/" + name);
+ ObjectNode configs = JsonUtils.newObjectNode();
+ CatalogRecord selected = null;
+ for (TableType candidate : TableType.values()) {
+ if (requested == null || requested == candidate) {
+ CatalogRecord record = snapshot.tables().get(TableNameBuilder.forType(candidate).tableNameWithType(raw));
+ if (record != null) {
+ configs.set(candidate.name(), record.value());
+ selected = record;
+ }
+ }
+ }
+ if (configs.isEmpty()) {
+ throw new WebApplicationException("Table not found", Response.Status.NOT_FOUND);
+ }
+ return configs.size() == 1 ? tagged(snapshot, configs, selected) : response(snapshot, configs);
+ }
+
+ @GET
+ @Path("/tables/{tableName}/schema")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.GET_SCHEMA)
+ public Response tableSchema(@PathParam("tableName") String name, @Context HttpHeaders headers) {
+ name = translateTableName(name, headers);
+ Snapshot snapshot = read(headers, name, Actions.Table.GET_SCHEMA, "/tables/" + name + "/schema");
+ CatalogRecord record = snapshot.schemas().get(TableNameBuilder.extractRawTableName(name));
+ if (record == null) {
+ throw new WebApplicationException("Schema not found for table", Response.Status.NOT_FOUND);
+ }
+ return tagged(snapshot, record.value(), record);
+ }
+
+ @GET
+ @Path("/metadata/participants")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_INSTANCE)
+ public Response participants(@Context HttpHeaders headers) {
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_INSTANCE, "/metadata/participants");
+ ObjectNode result = JsonUtils.newObjectNode();
+ result.set("revision", revision(snapshot.revision()));
+ result.putObject("control").put("round", Long.toString(snapshot.control().round()));
+ ArrayNode participants = result.putArray("participants");
+ snapshot.participants().entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(entry -> {
+ ParticipantState state = entry.getValue();
+ ParticipantToken token = state.participant();
+ ObjectNode participant = participants.addObject();
+ participant.putObject("participant").put("historyId", token.historyId().toString())
+ .put("participantId", token.participantId()).put("role", token.role().name())
+ .put("incarnation", token.incarnation().toString())
+ .put("registrationEpoch", Long.toString(token.registrationEpoch()));
+ participant.set("endpoints", JsonUtils.objectToJsonNode(state.endpoints()));
+ participant.set("capabilities", JsonUtils.objectToJsonNode(state.capabilities()));
+ participant.put("expiryRound", Long.toString(state.expiryRound())).put("ready", state.ready())
+ .put("draining", state.draining()).put("retired", state.retired())
+ .put("live", !state.retired() && state.expiryRound() > snapshot.control().round());
+ participant.set("appliedRevision", revision(state.appliedRevision()));
+ });
+ return response(snapshot, result);
+ }
+
+ @GET
+ @Path("/metadata/capabilities")
+ @Authenticate(AccessType.READ)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_APP_CONFIG)
+ public Response capabilities(@Context HttpHeaders headers) {
+ Snapshot snapshot = read(headers, null, Actions.Cluster.GET_APP_CONFIG, "/metadata/capabilities");
+ return response(snapshot, Map.of("schemaCreate", true, "schemaUpdate", true, "schemaDelete", true,
+ "tableCreate", true, "tableUpdate", true, "tableDelete", false, "clusterConfigUpdate", true,
+ "clusterConfigDelete", true));
+ }
+
+ @POST
+ @Path("/schemas")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @ManualAuthorization
+ public Response createSchema(String body, @DefaultValue("false") @QueryParam("override") boolean override,
+ @DefaultValue("false") @QueryParam("force") boolean force, @Context HttpHeaders headers) {
+ if (override || force) {
+ throw error(501, "Native schema creation does not override existing schemas or skip validation", false);
+ }
+ Schema schema = parse(body, Schema.class);
+ schema.setSchemaName(translateSchemaName(schema.getSchemaName(), headers));
+ String name = schema.getSchemaName();
+ Snapshot snapshot = writeSnapshot(headers, name, AccessType.CREATE, Actions.Table.CREATE_SCHEMA, "/schemas");
+ UUID id = createId(snapshot, RecordKind.SCHEMA, name, snapshot.schemas().get(name), headers);
+ validate(() -> _metadata.validateSchema(snapshot, schema, null));
+ return commit(snapshot, Mutation.createSchema(snapshot.revision().historyId(), id, schema), headers, name,
+ AccessType.CREATE, Actions.Table.CREATE_SCHEMA, "/schemas", "Schema configuration published");
+ }
+
+ @PUT
+ @Path("/schemas/{schemaName}")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Authenticate(AccessType.UPDATE)
+ @Authorize(targetType = TargetType.TABLE, paramName = "schemaName", action = Actions.Table.UPDATE_SCHEMA)
+ public Response updateSchema(@PathParam("schemaName") String name, String body,
+ @DefaultValue("false") @QueryParam("reload") boolean reload,
+ @DefaultValue("false") @QueryParam("force") boolean force, @Context HttpHeaders headers) {
+ if (reload || force) {
+ throw error(501, "Native schema reload and forced updates are not supported", false);
+ }
+ name = translateSchemaName(name, headers);
+ Schema schema = parse(body, Schema.class);
+ schema.setSchemaName(translateSchemaName(schema.getSchemaName(), headers));
+ if (!name.equals(schema.getSchemaName())) {
+ throw error(400, "Schema path and body names differ", false);
+ }
+ String path = "/schemas/" + name;
+ Snapshot snapshot = writeSnapshot(headers, name, AccessType.UPDATE, Actions.Table.UPDATE_SCHEMA, path);
+ CatalogRecord previous = snapshot.schemas().get(name);
+ ObjectVersion expected = match(snapshot, previous, headers);
+ validate(() -> _metadata.validateSchema(snapshot, schema, schema(previous)));
+ return commit(snapshot, Mutation.updateSchema(expected, schema), headers, name, AccessType.UPDATE,
+ Actions.Table.UPDATE_SCHEMA, path, "Schema configuration published");
+ }
+
+ @DELETE
+ @Path("/schemas/{schemaName}")
+ @Authenticate(AccessType.DELETE)
+ @Authorize(targetType = TargetType.TABLE, paramName = "schemaName", action = Actions.Table.DELETE_SCHEMA)
+ public Response deleteSchema(@PathParam("schemaName") String name, @Context HttpHeaders headers) {
+ name = translateSchemaName(name, headers);
+ String path = "/schemas/" + name;
+ Snapshot snapshot = writeSnapshot(headers, name, AccessType.DELETE, Actions.Table.DELETE_SCHEMA, path);
+ ObjectVersion expected = match(snapshot, snapshot.schemas().get(name), headers);
+ try {
+ _metadata.validateSchemaDeletion(snapshot, name);
+ } catch (UnsupportedOperationException e) {
+ throw error(501, e.getMessage(), false);
+ } catch (IllegalStateException e) {
+ throw error(409, e.getMessage(), false);
+ }
+ return commit(snapshot, Mutation.delete(RecordKind.SCHEMA, name, expected), headers, name, AccessType.DELETE,
+ Actions.Table.DELETE_SCHEMA, path, "Schema configuration deleted");
+ }
+
+ @POST
+ @Path("/tables")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @ManualAuthorization
+ public Response createTable(String body, @QueryParam("validationTypesToSkip") String typesToSkip,
+ @Context HttpHeaders headers) {
+ if (typesToSkip != null && !typesToSkip.isBlank()) {
+ throw error(501, "Native table creation does not skip validation", false);
+ }
+ TableConfig table = parse(body, TableConfig.class);
+ table.setTableName(translateTableName(table.getTableName(), headers));
+ String name = table.getTableName();
+ Snapshot snapshot = writeSnapshot(headers, name, AccessType.CREATE, Actions.Table.CREATE_TABLE, "/tables");
+ UUID id = createId(snapshot, RecordKind.TABLE_CONFIG, name, snapshot.tables().get(name), headers);
+ validate(() -> _metadata.validateTable(snapshot, table, null));
+ return commit(snapshot, Mutation.createTableConfig(snapshot.revision().historyId(), id, table), headers, name,
+ AccessType.CREATE, Actions.Table.CREATE_TABLE, "/tables",
+ "Table configuration published; no segments assigned");
+ }
+
+ @PUT
+ @Path("/tables/{tableName}")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Authenticate(AccessType.UPDATE)
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.UPDATE_TABLE_CONFIG)
+ public Response updateTable(@PathParam("tableName") String name, String body,
+ @QueryParam("validationTypesToSkip") String typesToSkip,
+ @DefaultValue("false") @QueryParam("force") boolean force,
+ @Context HttpHeaders headers) {
+ if (force || typesToSkip != null && !typesToSkip.isBlank()) {
+ throw error(501, "Native table updates do not skip validation", false);
+ }
+ TableConfig table = parse(body, TableConfig.class);
+ table.setTableName(translateTableName(table.getTableName(), headers));
+ if (table.getTableType() == null) {
+ throw error(400, "Table type is required", false);
+ }
+ name = translateTableName(TableNameBuilder.forType(table.getTableType()).tableNameWithType(name), headers);
+ if (!name.equals(table.getTableName())) {
+ throw error(400, "Table path and body names differ", false);
+ }
+ String path = "/tables/" + name;
+ Snapshot snapshot = writeSnapshot(headers, name, AccessType.UPDATE, Actions.Table.UPDATE_TABLE_CONFIG, path);
+ CatalogRecord previous = snapshot.tables().get(name);
+ ObjectVersion expected = match(snapshot, previous, headers);
+ validate(() -> _metadata.validateTable(snapshot, table, table(previous)));
+ return commit(snapshot, Mutation.updateTableConfig(expected, table), headers, name, AccessType.UPDATE,
+ Actions.Table.UPDATE_TABLE_CONFIG, path, "Table configuration published");
+ }
+
+ @POST
+ @Path("/cluster/configs")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Authenticate(AccessType.UPDATE)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.UPDATE_CLUSTER_CONFIG)
+ public Response updateClusterConfig(String body, @Context HttpHeaders headers) {
+ JsonNode changes = parseObject(body);
+ Snapshot snapshot = writeSnapshot(headers, null, AccessType.UPDATE, Actions.Cluster.UPDATE_CLUSTER_CONFIG,
+ "/cluster/configs");
+ Map<String, String> values = new HashMap<>(snapshot.clusterConfig());
+ for (var entry : changes.properties()) {
+ if (entry.getKey().isBlank() || entry.getKey().length() > 1024
+ || !entry.getValue().isTextual() && !entry.getValue().isNull()) {
+ throw error(400, "Cluster config requires nonblank keys and string or null values", false);
+ }
+ if (entry.getValue().isNull()) {
+ values.remove(entry.getKey());
+ } else {
+ values.put(entry.getKey(), entry.getValue().textValue());
+ }
+ }
+ Mutation mutation;
+ if (snapshot.clusterRecord() == null) {
+ UUID id = createId(snapshot, RecordKind.CLUSTER_CONFIG, MetadataProtocol.CLUSTER_CONFIG_NAME, null, headers);
+ mutation = Mutation.createClusterConfig(snapshot.revision().historyId(), id, values);
+ } else {
+ mutation = Mutation.updateClusterConfig(match(snapshot, snapshot.clusterRecord(), headers), values);
+ }
+ validate(() -> _metadata.validateClusterConfig(snapshot, values));
+ return commit(snapshot, mutation, headers, null, AccessType.UPDATE, Actions.Cluster.UPDATE_CLUSTER_CONFIG,
+ "/cluster/configs", "Cluster configuration published");
+ }
+
+ @DELETE
+ @Path("/cluster/configs/{configName}")
+ @Authenticate(AccessType.DELETE)
+ @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.DELETE_CLUSTER_CONFIG)
+ public Response deleteClusterConfig(@PathParam("configName") String name, @Context HttpHeaders headers) {
+ String path = "/cluster/configs/" + name;
+ Snapshot snapshot = writeSnapshot(headers, null, AccessType.DELETE, Actions.Cluster.DELETE_CLUSTER_CONFIG, path);
+ ObjectVersion expected = match(snapshot, snapshot.clusterRecord(), headers);
+ Map<String, String> values = new HashMap<>(snapshot.clusterConfig());
+ values.remove(name);
+ validate(() -> _metadata.validateClusterConfig(snapshot, values));
+ return commit(snapshot, Mutation.updateClusterConfig(expected, values), headers, null, AccessType.DELETE,
+ Actions.Cluster.DELETE_CLUSTER_CONFIG, path, "Cluster configuration key deleted");
+ }
+
+ private Snapshot writeSnapshot(HttpHeaders headers, @Nullable String name, AccessType type, String action,
+ String path) {
+ authorize(headers, name, type, action, path);
+ try {
+ Snapshot snapshot = _metadata.snapshot();
+ authorize(headers, name, type, action, path);
+ return snapshot;
+ } catch (IOException | IllegalStateException e) {
+ throw error(503, "Native metadata is unavailable before mutation submission", false);
+ }
+ }
+
+ private Response commit(Snapshot snapshot, Mutation mutation, HttpHeaders headers, @Nullable String name,
+ AccessType type, String action, String path, String message) {
+ authorize(headers, name, type, action, path);
+ Result result;
+ try {
+ result = _metadata.commit(snapshot, mutation);
+ } catch (IOException e) {
+ throw error(503, "Mutation outcome is unknown. Inspect fresh metadata; absence does not prove no pending commit. "
+ + "Any retry must retain the same body, precondition and creation key.", true);
+ }
+ try {
+ authorize(headers, name, type, action, path);
+ } catch (WebApplicationException e) {
+ throw error(e.getResponse().getStatus(), "Authorization changed while the mutation was in flight", true);
+ }
+ if (result.status() != Status.OK) {
+ int status = switch (result.status()) {
+ case CONFLICT, OUT_OF_ORDER, UNACKNOWLEDGED_RESULT -> 409;
+ case HISTORY_MISMATCH, FENCED -> 412;
+ case NOT_FOUND -> 404;
+ case INVALID_REQUEST -> 400;
+ case LIMIT_EXCEEDED, RESPONSE_TOO_LARGE -> 413;
+ default -> 503;
+ };
+ throw error(status, result.detail(), false);
+ }
+ ObjectVersion version = result.versions().get(0);
+ return Response.ok(Map.of("status", "OK", "message", message), MediaType.APPLICATION_JSON_TYPE)
+ .header(REVISION_HEADER, result.revision().historyId() + ":" + result.revision().index())
+ .header(HttpHeaders.ETAG, etag(version)).header(HttpHeaders.CACHE_CONTROL, "no-store").build();
+ }
+
+ private static UUID createId(Snapshot snapshot, RecordKind kind, String name, @Nullable CatalogRecord current,
+ HttpHeaders headers) {
+ if (!"*".equals(headers.getHeaderString(HttpHeaders.IF_NONE_MATCH))) {
+ throw error(428, "Creation requires If-None-Match: *", false);
+ }
+ if (headers.getHeaderString(HttpHeaders.IF_MATCH) != null) {
+ throw error(400, "Creation must not supply If-Match", false);
+ }
+ String key = headers.getHeaderString("Idempotency-Key");
+ UUID requestId;
+ try {
+ requestId = UUID.fromString(key);
+ if (!requestId.toString().equals(key)) {
+ throw new IllegalArgumentException("Noncanonical UUID");
+ }
+ } catch (IllegalArgumentException | NullPointerException e) {
+ throw error(428, "Creation requires a canonical UUID Idempotency-Key retained across retries", false);
+ }
+ if (current != null) {
+ throw error(409, "Metadata name already exists", false);
+ }
+ return UUID.nameUUIDFromBytes((snapshot.revision().historyId() + "/" + kind + "/" + name + "/" + requestId)
+ .getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static ObjectVersion match(Snapshot snapshot, @Nullable CatalogRecord record, HttpHeaders headers) {
+ String supplied = headers.getHeaderString(HttpHeaders.IF_MATCH);
+ if (supplied == null) {
+ throw error(428, "Mutation requires the exact If-Match ETag from a fresh metadata read", false);
+ }
+ if (headers.getHeaderString(HttpHeaders.IF_NONE_MATCH) != null) {
+ throw error(400, "An update must not supply If-None-Match", false);
+ }
+ if (record == null) {
+ throw error(404, "Metadata object not found", false);
+ }
+ ObjectVersion expected = new ObjectVersion(snapshot.revision().historyId(), record.objectId(), record.version());
+ if (!etag(expected).equals(supplied)) {
+ throw error(412, "Metadata incarnation or version changed; reload before preparing a new edit", false);
+ }
+ return expected;
+ }
+
+ private static String etag(ObjectVersion version) {
+ return "\"" + version.historyId() + ":" + version.objectId() + ":" + version.version() + "\"";
+ }
+
+ private static Response tagged(Snapshot snapshot, Object body, @Nullable CatalogRecord record) {
+ String tag = record == null ? "\"" + snapshot.revision().historyId() + ":absent\""
+ : etag(new ObjectVersion(snapshot.revision().historyId(), record.objectId(), record.version()));
+ return Response.fromResponse(response(snapshot, body)).header(HttpHeaders.ETAG, tag).build();
+ }
+
+ private static <T> T parse(String body, Class<T> type) {
+ try {
+ var parsed = JsonUtils.stringToObjectAndUnrecognizedProperties(parseObject(body).toString(), type);
+ if (!parsed.getRight().isEmpty()) {
+ throw error(400, "Unrecognized configuration properties: " + parsed.getRight().keySet(), false);
+ }
+ return parsed.getLeft();
+ } catch (IOException | IllegalArgumentException e) {
+ throw error(400, "Invalid configuration body: " + e.getMessage(), false);
+ }
+ }
+
+ private static JsonNode parseObject(String body) {
+ if (body == null || body.getBytes(StandardCharsets.UTF_8).length > MetadataProtocol.MAX_COMMAND_BYTES) {
+ throw error(413, "Configuration body exceeds the supported request size", false);
+ }
+ try {
+ JsonNode node = REQUEST_JSON.readTree(body);
+ if (node == null || !node.isObject()) {
+ throw error(400, "Configuration body must be a JSON object", false);
+ }
+ return node;
+ } catch (IOException e) {
+ throw error(400, "Malformed configuration JSON", false);
+ }
+ }
+
+ private static void validate(Validation validation) {
+ try {
+ validation.run();
+ } catch (UnsupportedOperationException e) {
+ throw error(501, e.getMessage(), false);
+ } catch (IOException | IllegalArgumentException | IllegalStateException e) {
+ throw error(400, e.getMessage(), false);
+ }
+ }
+
+ private static WebApplicationException error(int status, String message, boolean unknown) {
+ return new WebApplicationException(Response.status(status).type(MediaType.APPLICATION_JSON_TYPE)
+ .entity(Map.of("error", message == null ? "Native metadata request rejected" : message,
+ "outcomeUnknown", unknown)).header(HttpHeaders.CACHE_CONTROL, "no-store").build());
+ }
+
+ @FunctionalInterface
+ private interface Validation {
+ void run() throws IOException;
+ }
+
+ private Snapshot read(HttpHeaders headers, @Nullable String name, String action, String path) {
+ authorize(headers, name, action, path);
+ Snapshot snapshot;
+ try {
+ snapshot = _metadata.snapshot();
+ } catch (IOException | IllegalStateException e) {
+ throw new WebApplicationException("Native metadata is unavailable; no cached result was returned", e,
+ Response.Status.SERVICE_UNAVAILABLE);
+ }
+ // A quorum read may take time; consult current policy again before returning the captured data.
+ authorize(headers, name, action, path);
+ return snapshot;
+ }
+
+ private void authorize(HttpHeaders headers, @Nullable String name, String action, String path) {
+ authorize(headers, name, AccessType.READ, action, path);
+ }
+
+ private void authorize(HttpHeaders headers, @Nullable String name, AccessType type, String action, String path) {
+ String raw = name == null ? null : TableNameBuilder.extractRawTableName(name);
+ if (raw != null && !supportedRawName(raw)) {
+ throw error(400, "Native metadata requires an unambiguous raw table name", false);
+ }
+ AccessControl access = _accessControlFactory.create();
+ AccessControlUtils.validatePermission(raw, type, headers, path, access);
+ if (!access.hasAccess(headers, raw == null ? TargetType.CLUSTER : TargetType.TABLE, raw, action)) {
+ throw new WebApplicationException("Permission denied", Response.Status.FORBIDDEN);
+ }
+ }
+
+ private boolean allowed(String name, HttpHeaders headers, String action, String path) {
+ String raw = TableNameBuilder.extractRawTableName(name);
+ if (!supportedRawName(raw)) {
+ return false;
+ }
+ AccessControl access = _accessControlFactory.create();
+ return access.hasAccess(raw, AccessType.READ, headers, path)
+ && access.hasAccess(headers, TargetType.TABLE, raw, action);
+ }
+
+ private static boolean inDatabase(String name, HttpHeaders headers) {
+ return DatabaseUtils.isPartOfDatabase(name, headers.getHeaderString(CommonConstants.DATABASE));
+ }
+
+ // Authorization implementations normalize type suffixes too. Only one suffix may be consumed across all
+ // layers; schema identities are already raw and must never carry a suffix themselves.
+ private static boolean supportedRawName(@Nullable String name) {
+ return name != null && !name.isBlank() && TableNameBuilder.getTableTypeFromTableName(name) == null;
+ }
+
+ private static String translateSchemaName(String name, HttpHeaders headers) {
+ String translated = translate(name, headers);
+ if (!supportedRawName(translated)) {
+ throw error(400, "Native schema names must be raw names without a table type suffix", false);
+ }
+ return translated;
+ }
+
+ private static String translateTableName(String name, HttpHeaders headers) {
+ String translated = translate(name, headers);
+ if (!supportedRawName(TableNameBuilder.extractRawTableName(translated))) {
+ throw error(400, "Native table names must contain at most one table type suffix", false);
+ }
+ return translated;
+ }
+
+ private static String translate(String name, HttpHeaders headers) {
+ try {
+ return DatabaseUtils.translateTableName(name, headers);
+ } catch (RuntimeException e) {
+ throw new WebApplicationException("Invalid database-qualified name", e, Response.Status.BAD_REQUEST);
+ }
+ }
+
+ @Nullable
+ private static TableType tableType(@Nullable String value) {
+ if (value == null) {
+ return null;
+ }
+ if ("OFFLINE".equalsIgnoreCase(value)) {
+ return TableType.OFFLINE;
+ }
+ if ("REALTIME".equalsIgnoreCase(value)) {
+ return TableType.REALTIME;
+ }
+ throw new WebApplicationException("Unknown table type", Response.Status.BAD_REQUEST);
+ }
+
+ private static Schema schema(CatalogRecord record) {
+ try {
+ return JsonUtils.jsonNodeToObject(record.value(), Schema.class);
+ } catch (IOException e) {
+ throw new WebApplicationException("Invalid committed schema", e, Response.Status.SERVICE_UNAVAILABLE);
+ }
+ }
+
+ private static TableConfig table(CatalogRecord record) {
+ try {
+ return JsonUtils.jsonNodeToObject(record.value(), TableConfig.class);
+ } catch (IOException e) {
+ throw new WebApplicationException("Invalid committed table config", e, Response.Status.SERVICE_UNAVAILABLE);
+ }
+ }
+
+ private static ObjectNode revision(Revision revision) {
+ return JsonUtils.newObjectNode().put("historyId", revision.historyId().toString())
+ .put("index", Long.toString(revision.index()));
+ }
+
+ private static Response response(Snapshot snapshot, Object body) {
+ try {
+ return Response.ok(JsonUtils.objectToString(body), MediaType.APPLICATION_JSON_TYPE)
+ .header(REVISION_HEADER, snapshot.revision().historyId() + ":" + snapshot.revision().index())
+ .header(HttpHeaders.CACHE_CONTROL, "no-store").build();
+ } catch (IOException e) {
+ throw new WebApplicationException("Cannot encode native metadata response", e,
+ Response.Status.SERVICE_UNAVAILABLE);
+ }
+ }
+}
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 ba7cc2c69ef..19aa33f4d8d 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
@@ -21,12 +21,28 @@ package org.apache.pinot.controller.coordination;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.file.Path;
+import java.time.Instant;
+import java.util.List;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
+import org.apache.pinot.common.audit.AuditServiceBinder;
import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.utils.ServiceStatus;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.controller.BaseControllerStarter;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.api.ControllerAdminApiApplication;
+import org.apache.pinot.controller.api.access.AccessControlFactory;
+import org.apache.pinot.controller.api.resources.LandingPageHandler;
+import org.apache.pinot.controller.api.resources.PinotControllerAppConfigs;
+import org.apache.pinot.controller.api.resources.PinotControllerAuthResource;
+import org.apache.pinot.controller.api.resources.PinotControllerHealthCheck;
+import org.apache.pinot.controller.api.resources.PinotVersionRestletResource;
+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;
@@ -43,8 +59,13 @@ import org.apache.pinot.coordination.ratis.transport.MetadataGrpcServer;
import org.apache.pinot.coordination.ratis.transport.MetadataGrpcService.Limits;
import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy;
import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
-import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.core.transport.ListenerConfig;
+import org.apache.pinot.core.util.ListenerConfigUtil;
+import org.apache.pinot.spi.metrics.PinotMetricUtils;
+import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
import org.apache.pinot.spi.plugin.PluginManager;
+import org.apache.pinot.spi.services.ServiceRole;
+import org.glassfish.hk2.utilities.binding.AbstractBinder;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
@@ -52,10 +73,12 @@ import static java.util.Objects.requireNonNull;
/// Owns one embedded voter, its qualified membership clock and authenticated metadata endpoint. Lifecycle methods
-/// are serialized; health checks use a bounded consensus read and do not claim HTTP, scheduling or query readiness.
+/// are serialized; the normal HTTP/UI shell exposes committed metadata management, without scheduling or query
+/// readiness.
/// 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 PinotConfiguration _config;
+ private final ControllerConf _config;
+ private final List<ListenerConfig> _listeners;
private final UUID _history;
private final String _instanceId;
private final String _providerClass;
@@ -72,12 +95,16 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
private volatile RatisMetadataStore _store;
@Nullable
private volatile RatisMetadataAccess _access;
+ @Nullable
+ private ControllerAdminApiApplication _adminApp;
+ @Nullable
+ private NativeControllerMetadata _metadata;
private volatile boolean _started;
private volatile boolean _closed;
private volatile String _status = "Native controller has not started";
public RatisControllerRuntime(ControllerConf config) {
- _config = new PinotConfiguration(config.toMap());
+ _config = new ControllerConf(config.toMap());
_history = UUID.fromString(NativeMetadataConfig.required(config, "historyId"));
String voter = NativeMetadataConfig.required(config, "id");
_instanceId = config.getInstanceId() == null ? "Controller_ratis_" + voter : config.getInstanceId();
@@ -90,6 +117,7 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
_certificate = NativeMetadataConfig.readableFile(config, "tls.certChain");
_privateKey = NativeMetadataConfig.readableFile(config, "tls.privateKey");
_trust = NativeMetadataConfig.readableFile(config, "tls.trustCerts");
+ _listeners = ListenerConfigUtil.buildControllerConfigs(_config);
_startupMillis = config.getProperty(NativeMetadataConfig.PREFIX + "controller.startup.timeoutMs", 30_000L);
checkArgument(_startupMillis > 0 && _startupMillis <= 300_000,
"Native controller startup timeout must be between 1 and 300000 milliseconds");
@@ -102,6 +130,9 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
public synchronized void start() throws Exception {
checkState(!_started && !_closed && _provider == null, "Native controller is already started or closed");
try {
+ AccessControlFactory accessControlFactory =
+ PluginManager.get().createInstance(_config.getAccessControlFactoryClass());
+ accessControlFactory.initWithoutHelix(_config);
_provider = PluginManager.get().createInstance(_providerClass);
_provider.init(_config);
MembershipClockSettings clock = requireNonNull(_provider.membershipClockSettings(), "membershipClockSettings");
@@ -120,9 +151,10 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
ParticipantIdentity identity = participants.authorize(fingerprint);
return identity != null && identity.capabilities().isEmpty() ? identity : null;
}, audit, Limits.defaults());
+ startHttp(accessControlFactory);
checkClock();
_started = true;
- _status = "Native metadata endpoint and membership initialized; controller HTTP and query serving are disabled";
+ _status = "Native metadata and controller HTTP initialized; query serving is disabled";
} catch (Exception | Error e) {
_status = "Native controller startup failed: " + e.getMessage();
try {
@@ -137,6 +169,35 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
}
}
+ private void startHttp(AccessControlFactory accessControlFactory) throws IOException {
+ NativeControllerMetadata metadata = new NativeControllerMetadata(requireNonNull(_access), _history, _config);
+ _metadata = metadata;
+ metadata.start();
+ // The metrics factory owns its process-wide registry; a controller must not shut it down on restart.
+ PinotMetricsRegistry registry =
+ PinotMetricUtils.getPinotMetricsRegistry(_config.subset("pinot.controller.metrics"));
+ ControllerMetrics metrics = new ControllerMetrics(_config.getMetricsPrefix(), registry);
+ metrics.initializeGlobalMeters();
+ metrics.setValueOfGlobalGauge(ControllerGauge.VERSION, PinotVersion.VERSION_METRIC_NAME, 1);
+ _adminApp = new ControllerAdminApiApplication(_config, Set.of(NativeControllerResource.class,
+ LandingPageHandler.class, PinotControllerAuthResource.class, PinotVersionRestletResource.class,
+ PinotControllerAppConfigs.class, PinotControllerHealthCheck.class, WebApplicationExceptionMapper.class));
+ Instant startTime = Instant.now();
+ _adminApp.registerBinder(new AbstractBinder() {
+ @Override
+ protected void configure() {
+ bind(_config).to(ControllerConf.class);
+ bind(_instanceId).named(BaseControllerStarter.CONTROLLER_INSTANCE_ID);
+ bind(startTime).named(ControllerAdminApiApplication.START_TIME);
+ bind(metrics).to(ControllerMetrics.class);
+ bind(accessControlFactory).to(AccessControlFactory.class);
+ bind(metadata).to(NativeControllerMetadata.class);
+ }
+ });
+ _adminApp.registerBinder(new AuditServiceBinder(metadata, ServiceRole.CONTROLLER, metrics));
+ _adminApp.start(_listeners, metrics);
+ }
+
private static void validateBootstrapPolicy(Policy policy) {
long maxMillis = Long.MAX_VALUE / 1_000_000;
checkArgument(policy.version() == 1 && policy.minTickIntervalMillis() > 0
@@ -210,7 +271,14 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
if (response.status() != Status.OK || response.control() == null || response.control().policy() == null) {
throw new IOException("Committed membership policy is unavailable");
}
- _status = "Native metadata endpoint and membership initialized; controller HTTP and query serving are disabled";
+ String configFailure = requireNonNull(_metadata).clusterConfigFailure();
+ if (configFailure != null) {
+ throw new IOException("Native cluster configuration refresh failed: " + configFailure);
+ }
+ if (_adminApp == null || _adminApp.getHttpServer() == null || !_adminApp.getHttpServer().isStarted()) {
+ throw new IOException("Native controller HTTP is not running");
+ }
+ _status = "Native metadata and controller HTTP initialized; query serving is disabled";
return ServiceStatus.Status.GOOD;
} catch (Exception e) {
_status = "Native metadata is unavailable: " + e.getMessage();
@@ -231,7 +299,8 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu
_closed = true;
_started = false;
Exception failure = null;
- for (AutoCloseable resource : new AutoCloseable[]{_endpoint, _access, _store, _provider}) {
+ for (AutoCloseable resource : new AutoCloseable[]{_adminApp == null ? null : _adminApp::stop, _metadata,
+ _endpoint, _access, _store, _provider}) {
if (resource != null) {
try {
resource.close();
diff --git a/pinot-controller/src/main/resources/app/App.tsx b/pinot-controller/src/main/resources/app/App.tsx
index fc3436b580b..e177725ab4f 100644
--- a/pinot-controller/src/main/resources/app/App.tsx
+++ b/pinot-controller/src/main/resources/app/App.tsx
@@ -28,6 +28,7 @@ import { AppLoadingIndicator } from './components/AppLoadingIndicator';
import { AuthWorkflow } from 'Models';
import { TimezoneProvider } from './contexts/TimezoneContext';
import { runBootstrapRequest } from './utils/bootstrap';
+import NativeMetadataPage from './pages/NativeMetadataPage';
const INITIAL_BOOTSTRAP_TIMEOUT_MS = 3000;
@@ -38,7 +39,8 @@ export const App = () => {
const [loading, setLoading] = useState(true);
const [isAuthenticated, setIsAuthenticated] = useState(null);
const [role, setRole] = useState('');
- const { authUserName, authUserEmail, authenticated, authWorkflow } = useAuthProvider();
+ const { authUserName, authUserEmail, authenticated, authWorkflow, coordinationBackend } = useAuthProvider();
+ const nativeBackend = coordinationBackend === 'RATIS';
const history = useHistory();
useEffect(() => {
@@ -138,6 +140,9 @@ export const App = () => {
};
const getRouterData = () => {
+ if (nativeBackend) {
+ return [{ path: '/', Component: NativeMetadataPage }, ...RouterData.filter((route) => route.path === '/login')];
+ }
if (queryConsoleOnlyView) {
return RouterData.filter((routeObj) => {
return routeObj.path === '/query' || routeObj.path === '/query/timeseries';
@@ -151,6 +156,10 @@ export const App = () => {
useEffect(() => {
if (isAuthenticated) {
+ if (nativeBackend) {
+ setLoading(false);
+ return fetchClusterName();
+ }
const cancelClusterConfig = fetchClusterConfig();
const cancelClusterName = fetchClusterName();
const cancelUserRole = fetchUserRole();
@@ -161,7 +170,7 @@ export const App = () => {
cancelUserRole();
};
}
- }, [isAuthenticated]);
+ }, [isAuthenticated, nativeBackend]);
const loginRender = (Component, props) => {
if(isAuthenticated) {
@@ -179,7 +188,8 @@ export const App = () => {
const componentRender = (Component, props, role) => {
return (
<div className="p-8">
- <Layout clusterName={clusterName} {...props} role={role} authWorkflow={authWorkflow}>
+ <Layout clusterName={clusterName} {...props} role={role} authWorkflow={authWorkflow}
+ coordinationBackend={coordinationBackend}>
<Component {...props} />
</Layout>
</div>
@@ -212,7 +222,7 @@ export const App = () => {
))}
<Route path="*">
<Redirect
- to={PinotMethodUtils.getURLWithoutAccessToken(
+ to={nativeBackend ? '/' : PinotMethodUtils.getURLWithoutAccessToken(
queryConsoleOnlyView ? '/query' : '/'
)}
/>
diff --git a/pinot-controller/src/main/resources/app/components/Layout.tsx b/pinot-controller/src/main/resources/app/components/Layout.tsx
index 3c4c2c4d073..f146678065b 100644
--- a/pinot-controller/src/main/resources/app/components/Layout.tsx
+++ b/pinot-controller/src/main/resources/app/components/Layout.tsx
@@ -42,6 +42,9 @@ const BASE_NAVIGATION_ITEMS = [
const Layout = (props) => {
const role = props.role;
const navigationItems = React.useMemo(() => {
+ if (props.coordinationBackend === 'RATIS') {
+ return [{ id: 1, name: 'Native Metadata', link: '/', icon: <ClusterManagerIcon /> }];
+ }
// IMPORTANT: do not mutate a module-level array here. Layout can re-render many times,
// and mutation would duplicate entries in the sidebar.
if (props.authWorkflow === 'NONE') {
@@ -59,7 +62,7 @@ const Layout = (props) => {
];
}
return BASE_NAVIGATION_ITEMS;
- }, [props.authWorkflow, role]);
+ }, [props.authWorkflow, props.coordinationBackend, role]);
const hash = `/${window.location.hash.split('/')[1]}`;
const routeObj = navigationItems.find((obj) => obj.link === hash);
@@ -69,6 +72,9 @@ const Layout = (props) => {
const [openSidebar, setOpenSidebar] = React.useState(sidebarOpenState);
const appNavigationItems = React.useMemo(() => {
+ if (props.coordinationBackend === 'RATIS') {
+ return navigationItems;
+ }
if (app_state.queryConsoleOnlyView) {
return navigationItems.filter((navItem) => navItem.link === '/query');
}
@@ -77,7 +83,7 @@ const Layout = (props) => {
}
return navigationItems;
- }, [navigationItems, app_state.queryConsoleOnlyView, app_state.hideQueryConsoleTab]);
+ }, [navigationItems, props.coordinationBackend, app_state.queryConsoleOnlyView, app_state.hideQueryConsoleTab]);
const highlightSidebarLink = (id: number) => {
setSelectedId(id);
diff --git a/pinot-controller/src/main/resources/app/components/NativeMetadataEditor.tsx b/pinot-controller/src/main/resources/app/components/NativeMetadataEditor.tsx
new file mode 100644
index 00000000000..1aff2274f4a
--- /dev/null
+++ b/pinot-controller/src/main/resources/app/components/NativeMetadataEditor.tsx
@@ -0,0 +1,152 @@
+/**
+ * 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.
+ */
+
+import React, { useEffect, useRef, useState } from 'react';
+import { Box, Button, MenuItem, TextField, Typography } from '@material-ui/core';
+import Alert from '@material-ui/lab/Alert';
+import CustomDialog from './CustomDialog';
+import {
+ NativeWriteMarker, NativeWriteRequest, newNativeWriteId, rememberNativeWrite, validateNativeJson, writeNativeMetadata
+} from '../requests/nativeMetadata';
+
+export type NativeEditDraft = {
+ title: string;
+ method: 'POST' | 'PUT' | 'DELETE';
+ path: string;
+ json?: string;
+ etag?: string;
+ create: boolean;
+ clusterKeys?: string[];
+};
+
+type Props = {
+ draft: NativeEditDraft;
+ open: boolean;
+ onClose: () => void;
+ onInspect: () => void;
+ onCommitted: () => void;
+ onPending: (marker: NativeWriteMarker | null) => void;
+};
+
+// An uncertain request remains immutable. Retrying sends the same bytes, version precondition and create key.
+const NativeMetadataEditor = ({ draft, open, onClose, onInspect, onCommitted, onPending }: Props) => {
+ const [json, setJson] = useState(draft.json || '');
+ const [key, setKey] = useState(draft.clusterKeys?.[0] || '');
+ const [pending, setPending] = useState<NativeWriteRequest | null>(null);
+ const [sending, setSending] = useState(false);
+ const [unknown, setUnknown] = useState(false);
+ const [message, setMessage] = useState<string | null>(null);
+ const sendingRef = useRef(false);
+
+ useEffect(() => {
+ if (!pending) {
+ return;
+ }
+ const warn = (event: BeforeUnloadEvent) => {
+ event.preventDefault();
+ event.returnValue = '';
+ };
+ window.addEventListener('beforeunload', warn);
+ return () => window.removeEventListener('beforeunload', warn);
+ }, [pending]);
+
+ const submit = async () => {
+ if (sendingRef.current) {
+ return;
+ }
+ let request = pending;
+ try {
+ if (!request) {
+ if (draft.method !== 'DELETE') {
+ validateNativeJson(json);
+ if (new TextEncoder().encode(json).length > 2 * 1024 * 1024) {
+ throw new Error('Configuration exceeds the 2 MiB editor limit.');
+ }
+ }
+ if (draft.clusterKeys && !key) {
+ throw new Error('Select a configuration key.');
+ }
+ if (!draft.create && !draft.etag) {
+ throw new Error('Load the current metadata version before editing.');
+ }
+ request = Object.freeze({ id: newNativeWriteId(), method: draft.method,
+ path: draft.clusterKeys ? `${draft.path}/${encodeURIComponent(key)}` : draft.path,
+ body: draft.method === 'DELETE' ? undefined : json, etag: draft.etag, create: draft.create });
+ // Persist identity before sending. No payload or credentials are stored in the browser.
+ rememberNativeWrite(request);
+ setPending(request);
+ onPending(request);
+ }
+ } catch (failure) {
+ setMessage(failure instanceof Error ? failure.message : 'Unable to prepare the request.');
+ return;
+ }
+ sendingRef.current = true;
+ setSending(true);
+ setMessage(null);
+ const outcome = await writeNativeMetadata(request);
+ sendingRef.current = false;
+ setSending(false);
+ if (outcome.committed) {
+ rememberNativeWrite(null);
+ onPending(null);
+ onCommitted();
+ } else if (unknown || outcome.unknown) {
+ // A conflict on retry does not prove whether the original attempt committed.
+ setUnknown(true);
+ setMessage(`${outcome.message} The original write outcome remains unknown.`);
+ } else {
+ rememberNativeWrite(null);
+ setPending(null);
+ onPending(null);
+ setMessage(`Request rejected: ${outcome.message}`);
+ }
+ };
+
+ return (
+ <CustomDialog open={open} title={draft.title} size="md" handleClose={onClose} handleSave={submit}
+ btnOkText={sending ? 'Sending…' : pending ? 'Retry same request' : draft.method === 'DELETE' ? 'Delete' : 'Save'}
+ okBtnDisabled={sending} showCancelBtn={!pending} disableBackdropClick disableEscapeKeyDown
+ moreActions={unknown && !sending ? <Button onClick={onInspect}>Inspect current metadata</Button> : undefined}>
+ {unknown && <Box mb={2}><Alert severity="warning">
+ This write may have committed. Retrying preserves the original request. Reading current metadata does not
+ establish its outcome; there is no public write-outcome lookup.
+ </Alert></Box>}
+ {message && <Box mb={2}><Alert severity={unknown ? 'warning' : 'error'}>{message}</Alert></Box>}
+ {draft.clusterKeys && <TextField select fullWidth label="Configuration key" value={key}
+ disabled={!!pending} onChange={(event) => setKey(event.target.value)}>
+ {draft.clusterKeys.map((name) => <MenuItem key={name} value={name}>{name}</MenuItem>)}
+ </TextField>}
+ {draft.method === 'DELETE' ? <Typography>Delete this metadata only if its loaded version is still current.</Typography>
+ : <TextField label="JSON configuration" fullWidth multiline rows={16} variant="outlined" value={json}
+ disabled={!!pending} onChange={(event) => setJson(event.target.value)}
+ inputProps={{ style: { fontFamily: 'monospace' } }} />}
+ {draft.path === '/tables' && draft.create && <Box mt={1}><Typography variant="body2">
+ Create the schema first, using the table name without the _OFFLINE suffix.
+ </Typography></Box>}
+ {draft.path === '/cluster/configs' && draft.method === 'POST' && <Box mt={1}><Typography variant="body2">
+ Keys omitted from this JSON remain unchanged. Use Delete key to remove a key.
+ </Typography></Box>}
+ {draft.etag && <Box mt={1}><Typography variant="caption">Loaded version: {draft.etag}</Typography></Box>}
+ {pending && <Box mt={1}><Typography variant="caption">Request ID: {pending.id}</Typography></Box>}
+ </CustomDialog>
+ );
+};
+
+export default NativeMetadataEditor;
diff --git a/pinot-controller/src/main/resources/app/components/auth/AuthProvider.tsx b/pinot-controller/src/main/resources/app/components/auth/AuthProvider.tsx
index 8bdde8b150f..dbe9b6f5222 100644
--- a/pinot-controller/src/main/resources/app/components/auth/AuthProvider.tsx
+++ b/pinot-controller/src/main/resources/app/components/auth/AuthProvider.tsx
@@ -18,13 +18,18 @@
*/
import { AuthLocalStorageKeys, AuthWorkflow } from 'Models';
+import { Box, Button } from '@material-ui/core';
+import Alert from '@material-ui/lab/Alert';
import React, { createContext, useContext, useEffect, useState } from 'react'
import { useHistory, useLocation } from 'react-router-dom';
import { baseApi, getAxiosErrorInterceptor, getAxiosRequestInterceptor, getAxiosResponseInterceptor, transformApi } from '../../utils/axios-config';
import PinotMethodUtils from '../../utils/PinotMethodUtils';
import { AppLoadingIndicator } from '../AppLoadingIndicator';
+export type CoordinationBackend = 'HELIX' | 'RATIS';
+
interface AuthProviderContextProps {
+ coordinationBackend: CoordinationBackend;
accessToken: string;
authUserName: string;
authUserEmail: string;
@@ -34,6 +39,8 @@ interface AuthProviderContextProps {
export const AuthProvider = ({ children }) => {
const [loading, setLoading] = useState<boolean>(true);
+ const [discoveryError, setDiscoveryError] = useState<boolean>(false);
+ const [coordinationBackend, setCoordinationBackend] = useState<CoordinationBackend | null>(null);
const [redirectUri, setRedirectUri] = useState<string | null>(null);
const [clientId, setClientId] = useState<string | null>(null);
const [authWorkflow, setAuthWorkflow] = useState<AuthWorkflow | null>(null);
@@ -56,12 +63,12 @@ export const AuthProvider = ({ children }) => {
}, []);
useEffect(() => {
- if (loading || authenticated) {
+ if (loading || discoveryError || authenticated) {
return;
}
initOidcAuth();
- }, [loading, authenticated]);
+ }, [loading, discoveryError, authenticated]);
useEffect(() => {
if (!autoLogout) {
@@ -72,46 +79,33 @@ export const AuthProvider = ({ children }) => {
}, [autoLogout])
const initAuthDetails = async () => {
- // fetch auth info details
- const authInfoResponse = await PinotMethodUtils.getAuthInfo();
-
- const authWorkFlowInternal =
- authInfoResponse && authInfoResponse.workflow
- ? authInfoResponse.workflow
- : AuthWorkflow.NONE;
-
- // set auth workflow
- setAuthWorkflow(authWorkFlowInternal);
-
- if (authWorkFlowInternal === AuthWorkflow.NONE) {
- // No authentication required
- setAuthenticated(true);
- }
-
- if (authWorkFlowInternal === AuthWorkflow.BASIC) {
- // basic auth is handled by login page
+ setLoading(true);
+ setDiscoveryError(false);
+ try {
+ const authInfoResponse = await PinotMethodUtils.getAuthInfo();
+ const backend = authInfoResponse.coordinationBackend;
+ const workflow = authInfoResponse.workflow;
+ if ((backend !== 'HELIX' && backend !== 'RATIS')
+ || ![AuthWorkflow.NONE, AuthWorkflow.BASIC, AuthWorkflow.OIDC].includes(workflow)) {
+ throw new Error('Unsupported controller discovery response.');
+ }
+ setCoordinationBackend(backend);
+ setAuthWorkflow(workflow);
+ if (workflow === AuthWorkflow.NONE) {
+ setAuthenticated(true);
+ }
+ // BASIC authentication continues to use the normal login page.
+ if (workflow === AuthWorkflow.OIDC) {
+ setAuthorizationEndpoint(`${authInfoResponse.issuer || ''}/auth`);
+ setRedirectUri(authInfoResponse.redirectUri || '');
+ setClientId(authInfoResponse.clientId || '');
+ }
+ } catch (error) {
+ // Do not mount any routes until the server has identified its backend and auth workflow.
+ setDiscoveryError(true);
+ } finally {
+ setLoading(false);
}
-
- // set OIDC auth details
- if (authWorkFlowInternal === AuthWorkflow.OIDC) {
- const issuer =
- authInfoResponse && authInfoResponse.issuer ? authInfoResponse.issuer : '';
-
- setAuthorizationEndpoint(`${issuer}/auth`);
- setRedirectUri(
- authInfoResponse && authInfoResponse.redirectUri
- ? authInfoResponse.redirectUri
- : ''
- );
- setClientId(
- authInfoResponse && authInfoResponse.clientId
- ? authInfoResponse.clientId
- : ''
- );
- }
-
- // auth loading complete
- setLoading(false);
}
const initOidcAuth = () => {
@@ -219,6 +213,7 @@ export const AuthProvider = ({ children }) => {
}
const authProvider: AuthProviderContextProps = {
+ coordinationBackend: coordinationBackend,
authWorkflow: authWorkflow,
accessToken: accessToken,
authUserName: authUserName,
@@ -229,7 +224,14 @@ export const AuthProvider = ({ children }) => {
if (loading) {
return <AppLoadingIndicator />
}
-
+ if (discoveryError || !coordinationBackend) {
+ return (
+ <Box p={3}>
+ <Alert severity="error">Unable to load controller configuration. The console is unavailable.</Alert>
+ <Box mt={2}><Button onClick={initAuthDetails} variant="contained">Retry</Button></Box>
+ </Box>
+ );
+ }
return (
<AuthProviderContext.Provider value={authProvider}>
diff --git a/pinot-controller/src/main/resources/app/pages/NativeMetadataPage.tsx b/pinot-controller/src/main/resources/app/pages/NativeMetadataPage.tsx
new file mode 100644
index 00000000000..ea0e452d8c6
--- /dev/null
+++ b/pinot-controller/src/main/resources/app/pages/NativeMetadataPage.tsx
@@ -0,0 +1,289 @@
+/**
+ * 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.
+ */
+
+import React, { useEffect, useState } from 'react';
+import {
+ Box, Button, CircularProgress, Grid, Paper, Tab, Table, TableBody, TableCell, TableHead, TableRow,
+ Tabs, Typography, makeStyles
+} from '@material-ui/core';
+import Alert from '@material-ui/lab/Alert';
+import {
+ NativeMetadataResponse, NativeParticipants, NativeWriteCapabilities, NativeWriteMarker,
+ nativeEditorJson, pendingNativeWrite, readNativeMetadata, readNativeWriteCapabilities
+} from '../requests/nativeMetadata';
+import NativeMetadataEditor, { NativeEditDraft } from '../components/NativeMetadataEditor';
+
+const useStyles = makeStyles({
+ page: { flex: 1, minWidth: 0, padding: 24, maxHeight: 'calc(100vh - 70px)', overflowY: 'auto' },
+ json: { whiteSpace: 'pre-wrap', overflowWrap: 'anywhere', margin: 0, fontSize: 13 },
+ revision: { overflowWrap: 'anywhere', marginTop: 12 },
+});
+
+const SECTIONS = ['Tables', 'Schemas', 'Cluster configuration', 'Membership'];
+
+// This page deliberately mounts no Helix table/instance widgets. Published config and lease membership are
+// metadata facts; neither establishes segment ONLINE state, serving authority, or query readiness.
+const NativeMetadataPage = () => {
+ const classes = useStyles();
+ const [section, setSection] = useState(0);
+ const [refresh, setRefresh] = useState(0);
+ const [selected, setSelected] = useState<string | null>(null);
+ const [loading, setLoading] = useState(true);
+ const [error, setError] = useState<string | null>(null);
+ const [result, setResult] = useState<NativeMetadataResponse<any> | null>(null);
+ const [capabilities, setCapabilities] = useState<NativeWriteCapabilities | null>(null);
+ const [capabilityError, setCapabilityError] = useState<string | null>(null);
+ const [draft, setDraft] = useState<NativeEditDraft | null>(null);
+ const [editorOpen, setEditorOpen] = useState(false);
+ const [pending, setPending] = useState<NativeWriteMarker | null>(() => pendingNativeWrite());
+ const [inspectedPending, setInspectedPending] = useState(false);
+ const [acknowledgedPending, setAcknowledgedPending] = useState(false);
+ const [writeStatus, setWriteStatus] = useState<string | null>(null);
+ const writesBlocked = loading || !!error || !!draft || (!!pending && !acknowledgedPending);
+
+ useEffect(() => {
+ let cancelled = false;
+ setCapabilities(null);
+ setCapabilityError(null);
+ readNativeWriteCapabilities().then((value) => {
+ if (!cancelled) {
+ setCapabilities(value);
+ }
+ }).catch((failure) => {
+ if (!cancelled) {
+ setCapabilityError(failure instanceof Error ? failure.message : 'Metadata editing is unavailable.');
+ }
+ });
+ return () => { cancelled = true; };
+ }, [refresh]);
+
+ useEffect(() => {
+ let cancelled = false;
+ setLoading(true);
+ setError(null);
+ setResult(null);
+ const path = section === 0
+ ? (selected ? `/tables/${encodeURIComponent(selected)}` : '/tables?type=OFFLINE')
+ : section === 1
+ ? (selected ? `/schemas/${encodeURIComponent(selected)}` : '/schemas')
+ : section === 2 ? '/cluster/configs' : '/metadata/participants';
+ readNativeMetadata(path).then((response) => {
+ if (!cancelled) {
+ setResult(response);
+ if (pending && !editorOpen) {
+ setInspectedPending(true);
+ }
+ }
+ }).catch((failure) => {
+ if (!cancelled) {
+ setError(failure instanceof Error ? failure.message : 'Unable to read committed metadata.');
+ }
+ }).finally(() => {
+ if (!cancelled) {
+ setLoading(false);
+ }
+ });
+ return () => { cancelled = true; };
+ }, [section, selected, refresh]);
+
+ const changeSection = (_event: React.ChangeEvent<{}>, next: number) => {
+ setSelected(null);
+ setResult(null);
+ setSection(next);
+ };
+
+ const openDraft = (operation: NativeEditDraft) => {
+ setWriteStatus(null);
+ setDraft(operation);
+ setEditorOpen(true);
+ };
+
+ const editLoaded = () => {
+ if (!result?.etag) {
+ return;
+ }
+ try {
+ const cluster = section === 2;
+ openDraft({ title: cluster ? 'Merge cluster configuration' : `Edit ${selected}`,
+ method: cluster ? 'POST' : 'PUT',
+ path: cluster ? '/cluster/configs' : `/${section === 0 ? 'tables' : 'schemas'}/${encodeURIComponent(selected)}`,
+ etag: result.etag, create: cluster && result.etag.endsWith(':absent"'),
+ json: nativeEditorJson(result, section === 0) });
+ } catch (_) {
+ setWriteStatus('This JSON cannot be edited without changing its numeric values. Use an API client that preserves them.');
+ }
+ };
+
+ const inspectPending = () => {
+ setEditorOpen(false);
+ setInspectedPending(false);
+ setRefresh((value) => value + 1);
+ };
+
+ const renderActions = () => {
+ if (!capabilities || !result) {
+ return null;
+ }
+ return <Box display="flex" mb={2}>
+ {section < 2 && !selected && (section === 0 ? capabilities.tableCreate : capabilities.schemaCreate) === true
+ && <Button color="primary" variant="outlined" disabled={writesBlocked} onClick={() => openDraft({
+ title: section === 0 ? 'Create OFFLINE table' : 'Create schema', method: 'POST',
+ path: section === 0 ? '/tables' : '/schemas?override=false', create: true,
+ json: section === 0 ? '{\n "tableName": "",\n "tableType": "OFFLINE",\n "segmentsConfig": {"replication": "1"}\n}'
+ : '{\n "schemaName": "",\n "dimensionFieldSpecs": [],\n "metricFieldSpecs": [],\n "dateTimeFieldSpecs": []\n}',
+ })}>{section === 0 ? 'Create OFFLINE table' : 'Create schema'}</Button>}
+ {result.etag && ((section === 0 && selected && capabilities.tableUpdate === true)
+ || (section === 1 && selected && capabilities.schemaUpdate === true)
+ || (section === 2 && capabilities.clusterConfigUpdate === true))
+ && <Button color="primary" variant="outlined" disabled={writesBlocked} onClick={editLoaded}>
+ {section === 2 ? 'Merge configuration' : 'Edit JSON'}
+ </Button>}
+ {section === 1 && selected && result.etag && capabilities.schemaDelete === true
+ && <Button disabled={writesBlocked} onClick={() => openDraft({ title: `Delete schema ${selected}`,
+ method: 'DELETE', path: `/schemas/${encodeURIComponent(selected)}`, create: false, etag: result.etag })}>
+ Delete schema
+ </Button>}
+ {section === 2 && result.etag && capabilities.clusterConfigDelete === true && Object.keys(result.value).length > 0
+ && <Button disabled={writesBlocked} onClick={() => openDraft({ title: 'Delete cluster configuration key',
+ method: 'DELETE', path: '/cluster/configs', create: false, etag: result.etag,
+ clusterKeys: Object.keys(result.value) })}>Delete key</Button>}
+ </Box>;
+ };
+
+ const renderMembership = (membership: NativeParticipants) => (
+ <>
+ <Typography variant="body2">
+ Lease state at committed round {membership.control?.round ?? 'uninitialized'}. Refresh to read a new snapshot.
+ A live lease or a ready flag does not establish query readiness.
+ </Typography>
+ <Table size="small" aria-label="Native membership">
+ <TableHead><TableRow>
+ <TableCell>Participant</TableCell><TableCell>Role</TableCell><TableCell>Lease at snapshot</TableCell>
+ <TableCell>Expiry round</TableCell><TableCell>Ready</TableCell><TableCell>Draining</TableCell>
+ <TableCell>Registration epoch</TableCell>
+ </TableRow></TableHead>
+ <TableBody>{membership.participants.map((member) => (
+ <TableRow key={member.participant.participantId}>
+ <TableCell>{member.participant.participantId}
+ <Typography variant="caption" component="div">Incarnation: {member.participant.incarnation}</Typography>
+ </TableCell>
+ <TableCell>{member.participant.role}</TableCell>
+ <TableCell>{member.retired ? 'Retired' : member.live ? 'Live' : 'Expired'}</TableCell>
+ <TableCell>{member.expiryRound}</TableCell><TableCell>{member.ready ? 'Yes' : 'No'}</TableCell>
+ <TableCell>{member.draining ? 'Yes' : 'No'}</TableCell>
+ <TableCell>{member.participant.registrationEpoch}</TableCell>
+ </TableRow>
+ ))}</TableBody>
+ </Table>
+ {membership.participants.length === 0 && <Typography>No registered participants.</Typography>}
+ </>
+ );
+
+ const renderResult = () => {
+ if (!result) {
+ return null;
+ }
+ if (section === 3) {
+ return renderMembership(result.value as NativeParticipants);
+ }
+ if (section < 2 && !selected) {
+ const names: string[] = section === 0 ? result.value.tables : result.value;
+ return names.length === 0 ? <Typography>No {SECTIONS[section].toLowerCase()} published.</Typography> : (
+ <Table size="small" aria-label={SECTIONS[section]}>
+ <TableHead><TableRow><TableCell>Name</TableCell></TableRow></TableHead>
+ <TableBody>{names.map((name) => (
+ <TableRow key={name}><TableCell>
+ <Button color="primary" onClick={() => { setResult(null); setSelected(name); }}>{name}</Button>
+ </TableCell></TableRow>
+ ))}</TableBody>
+ </Table>
+ );
+ }
+ return <pre className={classes.json}>{result.json}</pre>;
+ };
+
+ return (
+ <Grid item className={classes.page}>
+ <Box display="flex" alignItems="center" justifyContent="space-between" mb={2}>
+ <Typography variant="h5">Ratis metadata</Typography>
+ <Button variant="outlined" disabled={loading} onClick={() => setRefresh((value) => value + 1)}>Refresh</Button>
+ </Box>
+ <Alert severity="info">
+ Browse and update validated metadata. Query serving and segment lifecycle actions are disabled.
+ Updates may be rejected while native segment content or operations depend on the configuration.
+ </Alert>
+ {capabilityError && <Box mt={2}><Alert severity="warning">{capabilityError}</Alert></Box>}
+ {writeStatus && <Box mt={2}><Alert severity="info">{writeStatus}</Alert></Box>}
+ {pending && <Box mt={2}><Alert severity="warning">
+ Outcome unknown for {pending.method} {pending.path} (request {pending.id}). The write may have committed.
+ Reading current metadata does not establish its outcome. There is no public write-outcome lookup.
+ {acknowledgedPending && ' You acknowledged this uncertainty; any further write is a new operation.'}
+ <Box mt={1}>
+ <Button disabled={loading} onClick={inspectPending}>Inspect current metadata</Button>
+ {draft && !acknowledgedPending && <Button onClick={() => setEditorOpen(true)}>Review pending request</Button>}
+ {!acknowledgedPending && <Button disabled={!inspectedPending || loading} onClick={() => {
+ setAcknowledgedPending(true);
+ setDraft(null);
+ setEditorOpen(false);
+ }}>Acknowledge uncertainty</Button>}
+ </Box>
+ </Alert></Box>}
+ <Box mt={2}>
+ <Paper variant="outlined">
+ <Tabs value={section} onChange={changeSection} indicatorColor="primary" variant="scrollable">
+ {SECTIONS.map((label) => <Tab key={label} label={label} />)}
+ </Tabs>
+ <Box p={2}>
+ {selected && <Box mb={2}>
+ <Button onClick={() => { setResult(null); setSelected(null); }}>Back to {SECTIONS[section].toLowerCase()}</Button>
+ <Typography variant="h6">{selected}</Typography>
+ </Box>}
+ {loading && <CircularProgress size={24} aria-label="Loading committed metadata" />}
+ {error && <Alert severity="error">{error}</Alert>}
+ {!loading && !error && renderActions()}
+ {!loading && !error && renderResult()}
+ {result && <Typography variant="caption" component="p" className={classes.revision}>
+ Committed revision: {result.revision}
+ </Typography>}
+ </Box>
+ </Paper>
+ </Box>
+ {draft && <NativeMetadataEditor draft={draft} open={editorOpen} onClose={() => {
+ setDraft(null);
+ setEditorOpen(false);
+ }} onInspect={inspectPending} onPending={(marker) => {
+ setPending(marker);
+ setInspectedPending(false);
+ setAcknowledgedPending(false);
+ }} onCommitted={() => {
+ // React 16 does not batch these asynchronous updates. Clear the detail image before selecting its list.
+ setResult(null);
+ if (draft.method === 'DELETE' && section === 1) {
+ setSelected(null);
+ }
+ setDraft(null);
+ setEditorOpen(false);
+ setWriteStatus('Metadata change committed.');
+ setRefresh((value) => value + 1);
+ }} />}
+ </Grid>
+ );
+};
+
+export default NativeMetadataPage;
diff --git a/pinot-controller/src/main/resources/app/requests/index.ts b/pinot-controller/src/main/resources/app/requests/index.ts
index 9dd1dfaa736..bd432c3cd2d 100644
--- a/pinot-controller/src/main/resources/app/requests/index.ts
+++ b/pinot-controller/src/main/resources/app/requests/index.ts
@@ -391,7 +391,7 @@ export const getState = (tableName: string, tableType: string): Promise<AxiosRes
baseApi.get(`/tables/${tableName}/state?type=${tableType}`);
export const getInfo = (): Promise<AxiosResponse<OperationResponse>> =>
- baseApi.get(`/auth/info`);
+ baseApi.get(`/auth/info`, { timeout: 5000 });
export const authenticateUser = (authToken): Promise<AxiosResponse<OperationResponse>> =>
baseApi.get(`/auth/verify/v2`, {headers:{"Authorization": authToken}});
diff --git a/pinot-controller/src/main/resources/app/requests/nativeMetadata.ts b/pinot-controller/src/main/resources/app/requests/nativeMetadata.ts
new file mode 100644
index 00000000000..675696787b8
--- /dev/null
+++ b/pinot-controller/src/main/resources/app/requests/nativeMetadata.ts
@@ -0,0 +1,138 @@
+/**
+ * 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.
+ */
+
+import JSONbigBase from 'json-bigint';
+import { baseApi } from '../utils/axios-config';
+
+const JSONbig = JSONbigBase({ strict: true });
+const PENDING_WRITE_KEY = 'pinot_ui:nativePendingWrite';
+
+export type NativeMetadataResponse<T> = { value: T; revision: string; json: string; etag?: string };
+export type NativeWriteCapabilities = {
+ schemaCreate: boolean; schemaUpdate: boolean; schemaDelete: boolean;
+ tableCreate: boolean; tableUpdate: boolean; tableDelete: boolean;
+ clusterConfigUpdate: boolean; clusterConfigDelete: boolean;
+};
+export type NativeWriteMarker = { id: string; method: string; path: string };
+export type NativeWriteRequest = NativeWriteMarker & { method: 'POST' | 'PUT' | 'DELETE'; body?: string; etag?: string; create: boolean };
+export type NativeWriteOutcome = { committed: boolean; unknown: boolean; message: string };
+
+// Keep only the request identity across navigation. Config JSON and credentials must never be saved here.
+export const rememberNativeWrite = (request: NativeWriteMarker | null) => {
+ if (request) {
+ sessionStorage.setItem(PENDING_WRITE_KEY, JSON.stringify({ id: request.id, method: request.method, path: request.path }));
+ } else {
+ try { sessionStorage.removeItem(PENDING_WRITE_KEY); } catch (_) { /* A stale warning is safe. */ }
+ }
+};
+
+export const pendingNativeWrite = (): NativeWriteMarker | null => {
+ try {
+ const stored = sessionStorage.getItem(PENDING_WRITE_KEY);
+ if (!stored) {
+ return null;
+ }
+ const marker = JSON.parse(stored);
+ if (typeof marker.id === 'string' && typeof marker.method === 'string' && typeof marker.path === 'string') {
+ return marker;
+ }
+ } catch (_) { /* An unreadable marker still represents an unresolved write. */ }
+ return { id: 'unavailable', method: 'Unknown request', path: '' };
+};
+
+export const newNativeWriteId = (): string => {
+ const bytes = window.crypto.getRandomValues(new Uint8Array(16));
+ bytes[6] = (bytes[6] & 15) | 64;
+ bytes[8] = (bytes[8] & 63) | 128;
+ const hex = Array.from(bytes, (byte) => byte.toString(16).padStart(2, '0')).join('');
+ return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`;
+};
+
+// Unwrap the existing table response without rounding long/default numeric values through JavaScript Number.
+export const nativeEditorJson = (response: NativeMetadataResponse<any>, table = false): string => {
+ const value = JSONbig.parse(response.json);
+ return JSONbig.stringify(table ? value.OFFLINE : value, null, 2);
+};
+
+export const validateNativeJson = (json: string) => {
+ const value = JSONbig.parse(json);
+ if (!value || typeof value !== 'object' || Array.isArray(value)) {
+ throw new Error('Enter a JSON object.');
+ }
+};
+
+export const readNativeWriteCapabilities = async (): Promise<NativeWriteCapabilities> => {
+ const response = await baseApi.get('/metadata/capabilities', { timeout: 10000 });
+ if (response.status !== 200 || !response.data || typeof response.data !== 'object') {
+ throw new Error('Metadata editing is unavailable because supported operations could not be loaded.');
+ }
+ return response.data;
+};
+
+export const writeNativeMetadata = async (request: NativeWriteRequest): Promise<NativeWriteOutcome> => {
+ try {
+ const response = await baseApi.request({
+ method: request.method, url: request.path, data: request.body, timeout: 10000,
+ headers: { 'Content-Type': 'application/json', 'Idempotency-Key': request.id,
+ ...(request.create ? { 'If-None-Match': '*' } : { 'If-Match': request.etag }) },
+ });
+ const revision = response.headers?.['x-pinot-metadata-revision'];
+ if (response.status === 200 && response.data?.status === 'OK' && typeof response.data?.message === 'string'
+ && response.data?.error === undefined && response.data?.outcomeUnknown !== true && typeof revision === 'string'
+ && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}:[0-9]+$/i.test(revision)) {
+ return { committed: true, unknown: false, message: 'Metadata change committed.' };
+ }
+ const knownRejection = [400, 401, 403, 404, 409, 412, 413, 428, 501].includes(response.status)
+ && response.data?.outcomeUnknown === false && typeof response.data?.error === 'string';
+ return { committed: false, unknown: !knownRejection,
+ message: typeof response.data?.error === 'string' ? response.data.error : 'No confirmed write result was received.' };
+ } catch (_) {
+ return { committed: false, unknown: true, message: 'No confirmed write result was received.' };
+ }
+};
+
+export type NativeParticipants = {
+ revision: { historyId: string; index: string };
+ control: { round: string } | null;
+ participants: Array<{
+ participant: { participantId: string; role: string; incarnation: string; registrationEpoch: string };
+ expiryRound: string;
+ ready: boolean;
+ draining: boolean;
+ retired: boolean;
+ live: boolean;
+ }>;
+};
+
+// The shared axios client attaches the normal BASIC/OIDC credentials but resolves HTTP errors. Check status
+// explicitly so a denied or unavailable read can never be rendered as an empty successful metadata image.
+export const readNativeMetadata = async <T,>(path: string): Promise<NativeMetadataResponse<T>> => {
+ const response = await baseApi.get<string>(path, { timeout: 10000, transformResponse: [(data) => data] });
+ if (response.status !== 200 || response.data == null) {
+ throw new Error(response.status === 401 || response.status === 403
+ ? 'You are not authorized to browse this metadata.'
+ : response.status === 404 ? 'This metadata no longer exists. Reload its list.'
+ : 'Committed metadata is unavailable. Retry after the controller can reach a quorum.');
+ }
+ const revision = response.headers?.['x-pinot-metadata-revision'];
+ if (typeof revision !== 'string' || !revision) {
+ throw new Error('The controller did not provide a metadata revision.');
+ }
+ return { value: JSON.parse(response.data), json: response.data, revision, etag: response.headers?.etag };
+};
diff --git a/pinot-controller/src/main/resources/app/utils/PinotMethodUtils.ts b/pinot-controller/src/main/resources/app/utils/PinotMethodUtils.ts
index 6910045feb5..2acc8338b31 100644
--- a/pinot-controller/src/main/resources/app/utils/PinotMethodUtils.ts
+++ b/pinot-controller/src/main/resources/app/utils/PinotMethodUtils.ts
@@ -1237,7 +1237,10 @@ const getTableState = (tableName, tableType) => {
};
const getAuthInfo = () => {
- return getInfo().then((response)=>{
+ return getInfo().then((response) => {
+ if (response.status !== 200 || !response.data) {
+ throw new Error('Unable to discover controller authentication and coordination backend.');
+ }
return response.data;
});
};
diff --git a/pinot-controller/src/main/resources/app/utils/axios-config.ts b/pinot-controller/src/main/resources/app/utils/axios-config.ts
index 0a8ff97505b..8533408faf3 100644
--- a/pinot-controller/src/main/resources/app/utils/axios-config.ts
+++ b/pinot-controller/src/main/resources/app/utils/axios-config.ts
@@ -35,6 +35,7 @@ export const getAxiosRequestInterceptor = (
// basic auth
if (app_state.authWorkflow === AuthWorkflow.BASIC && app_state.authToken) {
requestConfig.headers = {
+ ...requestConfig.headers,
Authorization: app_state.authToken,
};
}
@@ -42,6 +43,7 @@ export const getAxiosRequestInterceptor = (
// OIDC auth
if (accessToken) {
requestConfig.headers = {
+ ...requestConfig.headers,
Authorization: accessToken,
};
}
diff --git a/pinot-controller/src/main/resources/vite.config.ts b/pinot-controller/src/main/resources/vite.config.ts
index 32ffa82bb73..c184e7ce882 100644
--- a/pinot-controller/src/main/resources/vite.config.ts
+++ b/pinot-controller/src/main/resources/vite.config.ts
@@ -28,7 +28,7 @@ const BACKEND_API_PREFIXES = [
'tenants', 'tables', 'schemas', 'segments', 'instances',
'cluster', 'periodictask', 'tasks', 'zk', 'sql', 'timeseries',
'brokers', 'table', 'auth', 'users', 'health', 'appconfigs',
- 'query', 'version', 'debug', 'loggers',
+ 'query', 'version', 'debug', 'loggers', 'metadata',
];
// Match the prefix then require either '/', '?', or end-of-path.
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactoryTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactoryTest.java
index a93dcd4732c..029de7a7d50 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactoryTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/access/BasicAuthAccessControlFactoryTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.api.access;
+import java.io.IOException;
import java.util.List;
import java.util.Map;
import javax.ws.rs.NotAuthorizedException;
@@ -149,6 +150,32 @@ public class BasicAuthAccessControlFactoryTest {
assertEquals(accessControl.getAuthWorkflowInfo().getWorkflow(), AccessControl.WORKFLOW_BASIC);
}
+ @Test
+ public void testInitializationWithoutHelixRequiresExplicitSupport() throws Exception {
+ PinotConfiguration configuration = new PinotConfiguration(Map.of(
+ "controller.admin.access.control.principals", FULL_USER,
+ "controller.admin.access.control.principals." + FULL_USER + ".password", FULL_PASSWORD));
+ expectThrows(IOException.class, () -> new ZkBasicAuthAccessControlFactory().initWithoutHelix(configuration));
+ BasicAuthAccessControlFactory custom = new BasicAuthAccessControlFactory() {
+ @Override
+ public void init(PinotConfiguration config, PinotHelixResourceManager manager) {
+ throw new IllegalStateException("Custom policy requires Helix metadata");
+ }
+ };
+ expectThrows(IOException.class, () -> custom.initWithoutHelix(configuration));
+ expectThrows(IOException.class, () -> new AllowAllAccessFactory() { }.initWithoutHelix(configuration));
+ AllowAllAccessFactory allowAll = new AllowAllAccessFactory();
+ allowAll.initWithoutHelix(configuration);
+ assertEquals(allowAll.create().getAuthWorkflowInfo().getWorkflow(), AccessControl.WORKFLOW_NONE);
+ BasicAuthAccessControlFactory basicAuth = new BasicAuthAccessControlFactory();
+ basicAuth.initWithoutHelix(configuration);
+ AccessControlUtils.validatePermission(null, AccessType.READ, headers(FULL_USER, FULL_PASSWORD), ENDPOINT_URL,
+ basicAuth.create());
+ expectThrows(NotAuthorizedException.class,
+ () -> AccessControlUtils.validatePermission(null, AccessType.READ, headers(FULL_USER, "wrong"), ENDPOINT_URL,
+ basicAuth.create()));
+ }
+
@Test(dataProvider = "accessControls")
public void testTablePermissionsAndTypedNameCompatibility(AccessControl accessControl) {
HttpHeaders restrictedHeaders = headers(RESTRICTED_USER, RESTRICTED_PASSWORD);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResourceTest.java
index f0fbd283968..8cf93170ff2 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResourceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotControllerAuthResourceTest.java
@@ -18,8 +18,14 @@
*/
package org.apache.pinot.controller.api.resources;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.Set;
import javax.ws.rs.NotAuthorizedException;
+import javax.ws.rs.core.Application;
import javax.ws.rs.core.HttpHeaders;
+import org.apache.pinot.common.coordination.CoordinationBackend;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.api.ControllerAdminApiApplication;
import org.apache.pinot.controller.api.access.AccessControl;
import org.apache.pinot.controller.api.access.AccessControlFactory;
import org.apache.pinot.controller.api.access.AccessType;
@@ -31,6 +37,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -45,6 +52,8 @@ public class PinotControllerAuthResourceTest {
private AccessControl _accessControl;
@Mock
private HttpHeaders _httpHeaders;
+ @Mock
+ private Application _application;
@InjectMocks
private PinotControllerAuthResource _resource;
@@ -69,4 +78,49 @@ public class PinotControllerAuthResourceTest {
assertFalse(_resource.verify(TABLE_NAME, AccessType.READ, ENDPOINT_URL));
}
+
+ @Test
+ public void testAuthInfoPreservesProviderFieldsAndUsesStartupBackend() {
+ for (CoordinationBackend backend : CoordinationBackend.values()) {
+ ControllerConf configuration = new ControllerConf();
+ if (backend == CoordinationBackend.RATIS) {
+ configuration.setProperty(CoordinationBackend.CONFIG_KEY, "RATIS");
+ }
+ ControllerAdminApiApplication application = backend == CoordinationBackend.RATIS
+ ? new ControllerAdminApiApplication(configuration, Set.of(PinotControllerAuthResource.class))
+ : new ControllerAdminApiApplication(configuration);
+ try {
+ configuration.setProperty(CoordinationBackend.CONFIG_KEY,
+ backend == CoordinationBackend.RATIS ? "HELIX" : "RATIS");
+ when(_application.getProperties()).thenReturn(application.getProperties());
+ when(_accessControlFactory.create()).thenReturn(_accessControl);
+ when(_accessControl.getAuthWorkflowInfo()).thenReturn(new ExtendedWorkflow());
+ ObjectNode info = _resource.info();
+ assertEquals(info.get("workflow").asText(), AccessControl.WORKFLOW_BASIC);
+ assertEquals(info.get("loginHint").asText(), "Sign in");
+ assertEquals(info.get("coordinationBackend").asText(), backend.name());
+ if (backend == CoordinationBackend.RATIS) {
+ assertFalse(application.isRegistered(PinotSchemaRestletResource.class));
+ }
+ } finally {
+ // Constructed applications also own the async executor, even when no HTTP listener was started.
+ application.stop();
+ application.stop();
+ }
+ }
+ }
+
+ private static final class ExtendedWorkflow extends AccessControl.AuthWorkflowInfo {
+ private ExtendedWorkflow() {
+ super(AccessControl.WORKFLOW_BASIC);
+ }
+
+ public String getLoginHint() {
+ return "Sign in";
+ }
+
+ public String getCoordinationBackend() {
+ return "provider-value-must-not-control-backend";
+ }
+ }
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java
new file mode 100644
index 00000000000..fa339575d67
--- /dev/null
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerResourceTest.java
@@ -0,0 +1,291 @@
+/**
+ * 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.databind.JsonNode;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.api.access.AccessControl;
+import org.apache.pinot.controller.api.access.AccessControlFactory;
+import org.apache.pinot.controller.api.access.AccessType;
+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.ControlToken;
+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.Policy;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Role;
+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.FeedCursor;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
+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.transport.MetadataAccess;
+import org.apache.pinot.core.auth.TargetType;
+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.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+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;
+import static org.testng.Assert.expectThrows;
+
+
+/// Checks native REST shapes, request authorization and the separate recoverable audit-config consumer.
+public class NativeControllerResourceTest {
+ @Test
+ public void testNormalReadShapesDatabaseAndAccessFiltering() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ // Unsupported old names must not become visible after repeated auth-layer suffix normalization.
+ fixture.record(RecordKind.SCHEMA, "orders_OFFLINE",
+ new Schema.SchemaBuilder().setSchemaName("orders_OFFLINE").build());
+ fixture.record(RecordKind.TABLE_CONFIG, "orders_OFFLINE_OFFLINE",
+ new TableConfigBuilder(TableType.OFFLINE).setTableName("orders_OFFLINE_OFFLINE").build());
+ HttpHeaders headers = headers(null);
+ NativeControllerResource resource = fixture.resource();
+ Response response = resource.tables(null, null, null, true, headers);
+ assertEquals(json(response).get("tables"), JsonUtils.objectToJsonNode(List.of("orders")));
+ assertEquals(response.getHeaderString(NativeControllerResource.REVISION_HEADER), fixture._history + ":10");
+ assertEquals(response.getHeaderString(HttpHeaders.CACHE_CONTROL), "no-store");
+ assertEquals(json(resource.tables("offline", null, "name", true, headers)).get("tables"),
+ JsonUtils.objectToJsonNode(List.of("orders_OFFLINE")));
+ assertEquals(json(resource.schemas(headers)), JsonUtils.objectToJsonNode(List.of("orders")));
+ assertEquals(json(resource.schema("orders", headers)).path("schemaName").asText(), "orders");
+ assertEquals(json(resource.table("orders", null, headers)).path("OFFLINE").path("tableName").asText(),
+ "orders_OFFLINE");
+ assertEquals(json(resource.tableSchema("orders_OFFLINE", headers)).path("schemaName").asText(), "orders");
+ assertEquals(json(resource.schemasInfo(headers)).path("schemasInfo").size(), 1);
+ assertEquals(json(resource.tables(null, null, null, true, headers("sales"))).path("tables"),
+ JsonUtils.objectToJsonNode(List.of("sales.orders")));
+ assertEquals(json(resource.schema("orders", headers("sales"))).path("schemaName").asText(), "sales.orders");
+ assertStatus(403, () -> resource.schema("hidden", headers));
+ assertStatus(404, () -> resource.table("absent", null, headers));
+ assertStatus(400, () -> resource.schema("sales.orders", headers("other")));
+ assertStatus(501, () -> resource.tables(null, null, "creationTime", true, headers));
+ assertEquals(json(resource.clusterInfo(headers)).path("clusterName").asText(), "native-test");
+ assertEquals(json(resource.clusterConfigs(headers)), JsonUtils.objectToJsonNode(Map.of("audit", "old")));
+ }
+ }
+
+ @Test
+ public void testWritesRejectNativeStateRequiringReconciliation() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ NativeControllerResource resource = fixture.resource();
+ var snapshot = fixture._metadata.snapshot();
+ CatalogRecord table = snapshot.tables().get("orders_OFFLINE");
+ fixture.record(RecordKind.OFFLINE_SEGMENT_CONTENT, "segment",
+ new Content("orders_OFFLINE", table.objectId(), "segment", UUID.randomUUID(), "a".repeat(64),
+ "file:///segment.tar.gz", new LoaderDescriptor(1, 0, false, null, null, 1, 1, Map.of())));
+ String tableTag = resource.table("orders_OFFLINE", null, headers(null)).getHeaderString(HttpHeaders.ETAG);
+ HttpHeaders tableHeaders = headers(null);
+ when(tableHeaders.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn(tableTag);
+ assertStatus(501, () -> resource.updateTable("orders_OFFLINE", table.value().toString(), null, false,
+ tableHeaders));
+ String schemaTag = resource.schema("orders", headers(null)).getHeaderString(HttpHeaders.ETAG);
+ HttpHeaders schemaHeaders = headers(null);
+ when(schemaHeaders.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn(schemaTag);
+ assertStatus(501, () -> resource.updateSchema("orders", snapshot.schemas().get("orders").value().toString(),
+ false, false, schemaHeaders));
+ String clusterTag = resource.clusterConfigs(headers(null)).getHeaderString(HttpHeaders.ETAG);
+ HttpHeaders clusterHeaders = headers(null);
+ when(clusterHeaders.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn(clusterTag);
+ assertStatus(501, () -> resource.updateClusterConfig("{}", clusterHeaders));
+ }
+ }
+
+ @Test
+ public void testMembershipAndUnavailableReadsNeverUseCache() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ NativeControllerResource resource = fixture.resource();
+ JsonNode membership = json(resource.participants(headers(null)));
+ assertEquals(membership.path("revision").path("index").asText(), "10");
+ assertEquals(membership.path("control").path("round").asText(), "5");
+ JsonNode broker = membership.path("participants").get(0);
+ assertEquals(broker.path("participant").path("role").asText(), "BROKER");
+ assertTrue(broker.path("live").asBoolean());
+ assertFalse(broker.path("ready").asBoolean());
+ JsonNode server = membership.path("participants").get(1);
+ assertTrue(server.path("ready").asBoolean());
+ assertFalse(server.path("live").asBoolean());
+ assertEquals(server.path("expiryRound").asText(), "5");
+ fixture._unavailable.set(true);
+ assertStatus(503, () -> resource.clusterConfigs(headers(null)));
+ assertEquals(fixture._metadata.getClusterConfigs(), Map.of("audit", "old"));
+ fixture._unavailable.set(false);
+ fixture._responseHistory.set(UUID.randomUUID());
+ assertStatus(503, () -> resource.schemas(headers(null)));
+ }
+ }
+
+ @Test
+ public void testCurrentAuthorizationIsRecheckedAfterRead() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ fixture._afterRead = () -> fixture._authorized.set(false);
+ assertStatus(403, () -> fixture.resource().clusterConfigs(headers(null)));
+ }
+ }
+
+ @Test
+ public void testAuditConfigInitialChangesUnsetAndCallbackRecovery() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ List<Set<String>> changed = new ArrayList<>();
+ AtomicBoolean rejectNext = new AtomicBoolean();
+ fixture._metadata.registerClusterConfigChangeListener((keys, values) -> {
+ changed.add(Set.copyOf(keys));
+ if (rejectNext.getAndSet(false)) {
+ throw new IllegalStateException("listener failed");
+ }
+ });
+ assertEquals(changed, List.of(Set.of("audit")));
+ fixture.config(Map.of("new", "enabled"), 11);
+ rejectNext.set(true);
+ assertThrows(IllegalStateException.class, fixture._metadata::refreshClusterConfig);
+ assertEquals(fixture._metadata.clusterConfigFailure(), "listener failed");
+ fixture._metadata.refreshClusterConfig();
+ assertEquals(changed, List.of(Set.of("audit"), Set.of("audit", "new"), Set.of("audit", "new")));
+ assertEquals(fixture._metadata.getClusterConfigs(), Map.of("new", "enabled"));
+ assertEquals(fixture._metadata.clusterConfigFailure(), null);
+ fixture._metadata.refreshClusterConfig();
+ assertEquals(changed.size(), 3);
+ }
+ }
+
+ private static JsonNode json(Response response) throws IOException {
+ assertEquals(response.getStatus(), 200);
+ return JsonUtils.stringToJsonNode((String) response.getEntity());
+ }
+
+ private static HttpHeaders headers(String database) {
+ HttpHeaders headers = mock(HttpHeaders.class);
+ when(headers.getHeaderString(CommonConstants.DATABASE)).thenReturn(database);
+ return headers;
+ }
+
+ private static void assertStatus(int status, Runnable request) {
+ WebApplicationException failure = expectThrows(WebApplicationException.class, request::run);
+ assertEquals(failure.getResponse().getStatus(), status);
+ }
+
+ private static final class Fixture implements AutoCloseable {
+ private final UUID _history = UUID.randomUUID();
+ private final AtomicReference<UUID> _responseHistory = new AtomicReference<>(_history);
+ private final AtomicBoolean _unavailable = new AtomicBoolean();
+ private final AtomicBoolean _authorized = new AtomicBoolean(true);
+ private final List<CatalogRecord> _records = new ArrayList<>();
+ private final NativeControllerMetadata _metadata;
+ private final AccessControlFactory _factory;
+ private long _index = 10;
+ private Runnable _afterRead = () -> { };
+
+ private Fixture() throws Exception {
+ for (String name : List.of("orders", "hidden", "sales.orders")) {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(name).addSingleValueDimension("name", DataType.STRING)
+ .build();
+ record(RecordKind.SCHEMA, name, schema);
+ record(RecordKind.TABLE_CONFIG, name + "_OFFLINE",
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(name).build());
+ }
+ record(RecordKind.MEMBERSHIP_CONTROL, "control", new ControlState(
+ new ControlToken(_history, 1, 1, "controller", UUID.randomUUID()), 5,
+ new Policy(1, 1000, 4, 1000, 500, 100), 8, 1000, 1000));
+ participant("Broker_live", Role.BROKER, 6, false);
+ participant("Server_expired", Role.SERVER, 5, true);
+ config(Map.of("audit", "old"), 10);
+ MetadataAccess access = mock(MetadataAccess.class);
+ when(access.image(any(), anyInt(), anyInt(), anyLong())).thenAnswer(invocation -> {
+ if (_unavailable.get()) {
+ throw new IOException("quorum unavailable");
+ }
+ FeedScope scope = invocation.getArgument(0);
+ CatalogImage image = new CatalogImage(new FeedCursor(_responseHistory.get(), _index, scope.scopeId()),
+ _records.stream().filter(record -> scope.kinds().contains(record.kind())).toList());
+ _afterRead.run();
+ return image;
+ });
+ ControllerConf conf = new ControllerConf();
+ conf.setProperty(NativeMetadataConfig.PREFIX + "id", "test-controller");
+ conf.setProperty(ControllerConf.HELIX_CLUSTER_NAME, "native-test");
+ conf.setProperty(NativeMetadataConfig.PREFIX + "controller.http.configRefreshMs", 60_000L);
+ _metadata = new NativeControllerMetadata(access, _history, conf);
+ _metadata.start();
+ _factory = () -> new AccessControl() {
+ @Override
+ public boolean hasAccess(String tableName, AccessType accessType, HttpHeaders headers, String endpoint) {
+ return _authorized.get() && !"hidden".equals(tableName);
+ }
+
+ @Override
+ public boolean hasAccess(HttpHeaders headers, TargetType targetType, String targetId, String action) {
+ return _authorized.get() && !"hidden".equals(targetId);
+ }
+ };
+ }
+
+ private void record(RecordKind kind, String name, Object value) {
+ _records.add(new CatalogRecord(kind, name, UUID.randomUUID(), 1, 1, 1, false,
+ JsonUtils.objectToJsonNode(value)));
+ }
+
+ private void participant(String name, Role role, long expiryRound, boolean ready) {
+ record(RecordKind.PARTICIPANT, name,
+ new ParticipantState(new ParticipantToken(_history, name, role, UUID.randomUUID(), 1), Map.of(),
+ Set.<Capability>of(), expiryRound, ready, false, false, new Revision(_history, 1), 1, 1));
+ }
+
+ private void config(Map<String, String> values, long index) {
+ _records.removeIf(record -> record.kind() == RecordKind.CLUSTER_CONFIG);
+ record(RecordKind.CLUSTER_CONFIG, "cluster", values);
+ _index = index;
+ }
+
+ private NativeControllerResource resource() {
+ return new NativeControllerResource(_metadata, _factory);
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ _metadata.close();
+ }
+ }
+}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerWriteTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerWriteTest.java
new file mode 100644
index 00000000000..3036dc48891
--- /dev/null
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerWriteTest.java
@@ -0,0 +1,334 @@
+/**
+ * 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.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.Response;
+import org.apache.pinot.common.auth.BasicAuthTokenUtils;
+import org.apache.pinot.common.coordination.NativeMetadataConfig;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.api.access.AccessControl;
+import org.apache.pinot.controller.api.access.BasicAuthAccessControlFactory;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ActivateControlEpoch;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ConfigurePolicy;
+import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
+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.ConfigTransaction;
+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.GetRevision;
+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.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.RegisterClient;
+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.MetadataProtocol.Transaction;
+import org.apache.pinot.coordination.ratis.transport.MetadataAccess;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+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.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Exercises HTTP preconditions and lost replies against the actual durable catalog and serialized controller stream.
+public class NativeControllerWriteTest {
+ @Test
+ public void testValidatedSchemaTableAndClusterWrites() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ NativeControllerResource resource = fixture._resource;
+ Schema schema = schema("orders");
+ assertOk(resource.createSchema(schema.toSingleLineJsonString(), false, false, createHeaders()));
+ String firstSchemaTag = resource.schema("orders", headers()).getHeaderString(HttpHeaders.ETAG);
+ String table = JsonUtils.objectToString(new TableConfigBuilder(TableType.OFFLINE).setTableName("orders")
+ .setTimeColumnName("ts").setNumReplicas(1).build());
+ assertOk(resource.createTable(table, null, createHeaders()));
+ assertError(409, () -> resource.deleteSchema("orders", matchHeaders(firstSchemaTag)));
+ Schema changed = schema("orders");
+ changed.addField(new DimensionFieldSpec("added", DataType.STRING, true));
+ assertOk(resource.updateSchema("orders", changed.toSingleLineJsonString(), false, false,
+ matchHeaders(firstSchemaTag)));
+ assertError(412, () -> resource.updateSchema("orders", changed.toSingleLineJsonString(), false, false,
+ matchHeaders(firstSchemaTag)));
+ String tableTag = resource.table("orders_OFFLINE", null, headers()).getHeaderString(HttpHeaders.ETAG);
+ JsonNode updated = JsonUtils.stringToJsonNode(table);
+ ((ObjectNode) updated).put("description", "metadata-only edit");
+ Response tableUpdated = resource.updateTable("orders_OFFLINE", updated.toString(), null, false,
+ matchHeaders(tableTag));
+ assertOk(tableUpdated);
+ ((ObjectNode) updated).put("description", "${HOST_SPECIFIC_SETTING:default}");
+ assertError(501, () -> resource.updateTable("orders_OFFLINE", updated.toString(), null, false,
+ matchHeaders(tableUpdated.getHeaderString(HttpHeaders.ETAG))));
+ assertOk(resource.updateClusterConfig("{\"one\":\"1\",\"two\":\"2\"}", createHeaders()));
+ String clusterTag = resource.clusterConfigs(headers()).getHeaderString(HttpHeaders.ETAG);
+ assertOk(resource.updateClusterConfig("{\"one\":null,\"two\":\"changed\"}", matchHeaders(clusterTag)));
+ assertEquals(JsonUtils.stringToJsonNode((String) resource.clusterConfigs(headers()).getEntity()),
+ JsonUtils.objectToJsonNode(Map.of("two", "changed")));
+ assertError(412, () -> resource.deleteClusterConfig("two", matchHeaders(clusterTag)));
+ assertError(428, () -> resource.updateClusterConfig("{}", headers()));
+ String currentClusterTag = resource.clusterConfigs(headers()).getHeaderString(HttpHeaders.ETAG);
+ assertError(501, () -> resource.updateClusterConfig("{\"key\":\"${HOST_SPECIFIC_SETTING}\"}",
+ matchHeaders(currentClusterTag)));
+ assertError(400, () -> resource.createSchema("{\"schemaName\":\"a\",\"schemaName\":\"b\"}",
+ false, false, createHeaders()));
+ assertError(400, () -> resource.createSchema("{\"schemaName\":\"a\",\"unknownProperty\":true}",
+ false, false, createHeaders()));
+ ObjectNode misspelledTable = (ObjectNode) JsonUtils.stringToJsonNode(table);
+ misspelledTable.withObject("/segmentsConfig").put("replicaton", "3");
+ assertError(400, () -> resource.createTable(misspelledTable.toString(), null, createHeaders()));
+ assertError(400, () -> resource.createSchema("{\"schemaName\":\"script\",\"dimensionFieldSpecs\":["
+ + "{\"name\":\"id\",\"dataType\":\"STRING\",\"transformFunction\":\"Groovy({1}, x)\"}]}",
+ false, false, createHeaders()));
+ String realtime = JsonUtils.objectToString(new TableConfigBuilder(TableType.REALTIME).setTableName("orders")
+ .build());
+ assertError(501, () -> resource.createTable(realtime, null, createHeaders()));
+ assertEquals(fixture._catalog.captureSnapshot().path("clients").size(), 1,
+ "HTTP writes must reuse one bounded controller stream rather than allocating a client per request");
+ }
+ }
+
+ @Test
+ public void testAmbiguousNamesCannotEscapeBasicAuthTableScope() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ BasicAuthAccessControlFactory factory = new BasicAuthAccessControlFactory();
+ factory.initWithoutHelix(new PinotConfiguration(Map.of(
+ "controller.admin.access.control.principals", "scoped",
+ "controller.admin.access.control.principals.scoped.password", "password",
+ "controller.admin.access.control.principals.scoped.tables", "orders",
+ "controller.admin.access.control.principals.scoped.permissions", "create,read,update,delete")));
+ NativeControllerResource resource = new NativeControllerResource(fixture._metadata, factory);
+ HttpHeaders authenticated = createHeaders();
+ when(authenticated.getRequestHeader(HttpHeaders.AUTHORIZATION))
+ .thenReturn(List.of(BasicAuthTokenUtils.toBasicAuthToken("scoped", "password")));
+ assertOk(resource.createSchema(schema("orders").toSingleLineJsonString(), false, false, authenticated));
+ for (String name : List.of("orders_OFFLINE", "orders_OFFLINE_OFFLINE", "orders_REALTIME")) {
+ assertError(400, () -> resource.createSchema(schema(name).toSingleLineJsonString(), false, false,
+ authenticated));
+ assertError(400, () -> resource.schema(name, authenticated));
+ }
+ String ambiguousTable = JsonUtils.objectToString(new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName("orders_OFFLINE_OFFLINE_OFFLINE").setTimeColumnName("ts").setNumReplicas(1).build());
+ assertError(400, () -> resource.createTable(ambiguousTable, null, authenticated));
+ assertError(400, () -> resource.table("orders_OFFLINE_OFFLINE", null, authenticated));
+ assertError(400, () -> resource.tableSchema("orders_OFFLINE_OFFLINE", authenticated));
+ assertEquals(fixture._metadata.snapshot().schemas().keySet(), Set.of("orders"));
+ assertTrue(fixture._metadata.snapshot().tables().isEmpty());
+ }
+ }
+
+ @Test
+ public void testLostReplyRestartAndCreateTombstoneFence() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ Schema schema = schema("orders");
+ HttpHeaders create = createHeaders();
+ fixture._loseReply = true;
+ WebApplicationException lost = assertError(503,
+ () -> fixture._resource.createSchema(schema.toSingleLineJsonString(), false, false, create));
+ assertTrue(JsonUtils.objectToJsonNode(lost.getResponse().getEntity()).path("outcomeUnknown").asBoolean());
+ fixture.restart();
+ assertError(409, () -> fixture._resource.createSchema(schema.toSingleLineJsonString(), false, false, create));
+ String tag = fixture._resource.schema("orders", headers()).getHeaderString(HttpHeaders.ETAG);
+ assertOk(fixture._resource.deleteSchema("orders", matchHeaders(tag)));
+ assertError(409, () -> fixture._resource.createSchema(schema.toSingleLineJsonString(), false, false, create));
+ assertError(404, () -> fixture._resource.schema("orders", headers()));
+ assertOk(fixture._resource.createSchema(schema.toSingleLineJsonString(), false, false, createHeaders()));
+ assertEquals(fixture._catalog.captureSnapshot().path("clients").size(), 1);
+ StreamTipResult tip = fixture.query(new GetStreamTip(fixture._history, "controller-http/test-controller", 1,
+ "config"), 1000);
+ assertTrue(tip.sequence() > 1);
+ assertEquals(tip.floor(), tip.sequence() - 1);
+ }
+ }
+
+ @Test
+ public void testDependencyMutationAfterValidationCannotCommit() throws Exception {
+ try (Fixture fixture = new Fixture()) {
+ assertOk(fixture._resource.createSchema(schema("orders").toSingleLineJsonString(), false, false,
+ createHeaders()));
+ var record = fixture._metadata.snapshot().schemas().get("orders");
+ fixture._beforeWrite = () -> {
+ assertEquals(fixture.execute(new RegisterClient(fixture._history, "other-controller", UUID.randomUUID(), 0),
+ 1000).status(), Status.OK);
+ assertEquals(fixture.execute(new Transaction(fixture._history, "other-controller", 1, "config", 1,
+ List.of(Mutation.delete(RecordKind.SCHEMA, "orders",
+ new ObjectVersion(fixture._history, record.objectId(), record.version())))), 1000).status(), Status.OK);
+ };
+ String table = JsonUtils.objectToString(new TableConfigBuilder(TableType.OFFLINE).setTableName("orders")
+ .setTimeColumnName("ts").setNumReplicas(1).build());
+ WebApplicationException conflict = assertError(409,
+ () -> fixture._resource.createTable(table, null, createHeaders()));
+ assertFalse(JsonUtils.objectToJsonNode(conflict.getResponse().getEntity()).path("outcomeUnknown").asBoolean());
+ assertTrue(fixture._metadata.snapshot().tables().isEmpty());
+ assertTrue(fixture._metadata.snapshot().schemas().isEmpty());
+ }
+ }
+
+ private static Schema schema(String name) {
+ return new Schema.SchemaBuilder().setSchemaName(name).addSingleValueDimension("id", DataType.STRING)
+ .addDateTime("ts", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();
+ }
+
+ private static HttpHeaders headers() {
+ return mock(HttpHeaders.class);
+ }
+
+ private static HttpHeaders createHeaders() {
+ HttpHeaders headers = headers();
+ when(headers.getHeaderString(HttpHeaders.IF_NONE_MATCH)).thenReturn("*");
+ when(headers.getHeaderString("Idempotency-Key")).thenReturn(UUID.randomUUID().toString());
+ return headers;
+ }
+
+ private static HttpHeaders matchHeaders(String tag) {
+ HttpHeaders headers = headers();
+ when(headers.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn(tag);
+ return headers;
+ }
+
+ private static void assertOk(Response response) {
+ assertEquals(response.getStatus(), 200);
+ assertEquals(JsonUtils.objectToJsonNode(response.getEntity()).path("status").asText(), "OK");
+ assertTrue(response.getHeaderString(NativeControllerResource.REVISION_HEADER).contains(":"));
+ assertTrue(response.getHeaderString(HttpHeaders.ETAG).startsWith("\""));
+ }
+
+ private static WebApplicationException assertError(int status, ThrowingRequest request) {
+ WebApplicationException failure = expectThrows(WebApplicationException.class, request::run);
+ assertEquals(failure.getResponse().getStatus(), status);
+ return failure;
+ }
+
+ @FunctionalInterface
+ private interface ThrowingRequest {
+ void run() throws Exception;
+ }
+
+ private static final class Fixture implements MetadataAccess, AutoCloseable {
+ private final UUID _history = UUID.randomUUID();
+ private final MetadataCatalog _catalog = new MetadataCatalog(UUID.randomUUID(), _history);
+ private final ControllerConf _config = new ControllerConf();
+ private NativeControllerMetadata _metadata;
+ private NativeControllerResource _resource;
+ private long _index;
+ private boolean _loseReply;
+ private ThrowingRequest _beforeWrite;
+
+ private Fixture() throws Exception {
+ _config.setProperty(NativeMetadataConfig.PREFIX + "id", "test-controller");
+ _config.setProperty(ControllerConf.HELIX_CLUSTER_NAME, "native-writes");
+ _config.setProperty(NativeMetadataConfig.PREFIX + "controller.http.configRefreshMs", 60_000L);
+ var activated = MembershipProtocol.decodeResponse(apply(new MembershipCommand(_history, "bootstrap",
+ new ActivateControlEpoch(1, "leader", UUID.randomUUID()))));
+ assertEquals(activated.status(), Status.OK);
+ var configured = MembershipProtocol.decodeResponse(apply(new MembershipCommand(_history, "bootstrap",
+ new ConfigurePolicy(activated.control().controlEpoch(), 0, new Policy(1, 1000, 4, 1000, 500, 100)))));
+ assertEquals(configured.status(), Status.OK);
+ restart();
+ }
+
+ private void restart() throws Exception {
+ if (_metadata != null) {
+ _metadata.close();
+ }
+ _metadata = new NativeControllerMetadata(this, _history, _config);
+ _metadata.start();
+ _resource = new NativeControllerResource(_metadata, () -> new AccessControl() { });
+ }
+
+ private byte[] apply(Command command) throws IOException {
+ return _catalog.execute(MetadataProtocol.encode(command), 1, ++_index);
+ }
+
+ @Override
+ public synchronized Result execute(Command command, long timeoutMillis) throws IOException {
+ if (command instanceof ConfigTransaction && _beforeWrite != null) {
+ ThrowingRequest before = _beforeWrite;
+ _beforeWrite = null;
+ try {
+ before.run();
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+ Result result = MetadataProtocol.decodeResult(apply(command));
+ if (command instanceof ConfigTransaction && _loseReply) {
+ _loseReply = false;
+ throw new IOException("reply lost after commit");
+ }
+ return result;
+ }
+
+ @Override
+ public synchronized Result query(Query query, long timeoutMillis) throws IOException {
+ return MetadataProtocol.decodeResult(apply(new ReadBarrier(query)));
+ }
+
+ @Override
+ public synchronized StreamTipResult query(GetStreamTip query, long timeoutMillis) throws IOException {
+ return MetadataProtocol.decodeStreamTipResult(apply(new ReadBarrier(query)));
+ }
+
+ @Override
+ public synchronized CatalogImage image(FeedScope scope, int maxRecords, int maxBytes, long timeoutMillis)
+ throws IOException {
+ query(new GetRevision(_history), timeoutMillis);
+ 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 InterruptedException {
+ _metadata.close();
+ }
+ }
+}
diff --git a/pinot-coordination-ratis/README.md b/pinot-coordination-ratis/README.md
index 3c821f56878..1f14cef394a 100644
--- a/pinot-coordination-ratis/README.md
+++ b/pinot-coordination-ratis/README.md
@@ -324,7 +324,8 @@ selection, process identity, peers, TLS and local storage are bootstrap settings
- Controller starts one embedded voter, its explicitly qualified membership clock and the mutually authenticated
metadata/membership/installation endpoint. It initializes an absent membership policy with expected version zero;
- restart or a competing controller cannot overwrite an existing committed policy.
+ restart or a competing controller cannot overwrite an existing committed policy. The normal controller HTTP/UI shell
+ serves authenticated metadata reads and conditional configuration writes through an explicit native resource set.
- Server starts the existing native offline participant with independent membership/work connections, a locked durable
data directory, metadata hydration and installation reconciliation. It requests no query capabilities and advertises
no query endpoints. The existing server query-start gate stays closed.
@@ -333,9 +334,9 @@ selection, process identity, peers, TLS and local storage are bootstrap settings
membership/control projection, not a routing image.
No broker HTTP/gRPC/MSE or server Netty/TLS/gRPC/MSE query listeners are allocated, even if legacy query-port settings
-are present. Helix-dependent controller HTTP/UI resources and server admin resources are not started on the native
-path. Minions and multi-cluster brokers are unsupported. These entry points exercise coordination and offline
-installation; they do not yet provide a queryable replacement cluster or the normal controller UI.
+are present. Helix-dependent REST resources and server admin resources are not started on the native path. Minions
+and multi-cluster brokers are unsupported. These entry points exercise coordination and offline installation; they
+do not yet provide a queryable replacement cluster.
All native components need `pinot.coordination.ratis.historyId`. Controllers retain the voter settings in the next
section and additionally require:
@@ -344,6 +345,7 @@ section and additionally require:
pinot.coordination.backend=RATIS
pinot.coordination.ratis.metadata.host=127.0.0.1
pinot.coordination.ratis.metadata.port=24001
+controller.port=9000
pinot.coordination.ratis.tls.certChain=/absolute/path/controller.pem
pinot.coordination.ratis.tls.privateKey=/absolute/path/controller-key.pem
pinot.coordination.ratis.tls.trustCerts=/absolute/path/trust.pem
@@ -391,6 +393,82 @@ bin/pinot-admin.sh StartServer -configFileName server.conf
Startup failure closes owned channels, voters and directory ownership. Shutdown retires only the process's confirmed
registration; unknown remote outcomes rely on lease expiry. The existing Helix/ZK quickstarts are unchanged.
+## Controller REST and UI
+
+The native controller reuses `ControllerAdminApiApplication`, the normal static UI, login, HTTP/HTTPS listener
+configuration, authentication filter, and audit services. Configure `controller.port` or `controller.access.protocols`
+and the normal protocol listener settings; each controller process needs its own ports. HTTP TLS continues to use
+`controller.tls.*`, separately from the metadata endpoint's mandatory mutual TLS configuration.
+
+The default access control remains allow-all, as on Helix. Configuration-based `BasicAuthAccessControlFactory` is
+supported. Other factories must explicitly implement `AccessControlFactory.initWithoutHelix`; ZK Basic Auth and
+factories that do not opt in fail startup. No null Helix manager or substitute Helix state is supplied.
+
+Supported read endpoints are `/cluster/info`, `/cluster/configs`, `/schemas`, `/schemas/info`,
+`/schemas/{schemaName}`, `/tables`, `/tables/{tableName}`, `/tables/{tableName}/schema`, and
+`/metadata/participants`, alongside normal health, version, auth, uptime and app-config reads. Catalog responses use
+bounded fresh quorum reads; unavailable quorum returns 503, never a successful stale cached image. Native responses
+include `X-Pinot-Metadata-Revision`. Schema/table reads retain database scoping and permission filtering.
+Membership views distinguish active, expired, retired and draining registrations; a member's readiness does not
+claim query-serving authority. Embedded controller voters are not fabricated as participant registrations.
+
+The following conditional JSON mutations are available; `/metadata/capabilities` identifies this supported surface:
+
+| Method and path | Behavior |
+| --- | --- |
+| `POST /schemas` | Create a validated schema |
+| `PUT /schemas/{schemaName}` | Publish a compatible schema update |
+| `DELETE /schemas/{schemaName}` | Delete an unreferenced schema |
+| `POST /tables` | Create an ordinary OFFLINE table configuration |
+| `PUT /tables/{tableName}` | Publish a compatible OFFLINE table configuration update |
+| `POST /cluster/configs` | Merge string values; null values remove keys |
+| `DELETE /cluster/configs/{configName}` | Remove a cluster configuration key |
+
+Creation requires `If-None-Match: *` and a canonical UUID `Idempotency-Key`. Object IDs are derived from the history,
+record kind, name and key; retained tombstones prevent a retry from recreating an object that has since been deleted.
+Updates and deletes require the exact quoted `If-Match` ETag returned by the object's GET. ETags contain history,
+object ID and version; neither an old version nor a recreated name can be overwritten. An absent cluster config
+returns an `:absent` ETag and must be created with creation headers. A create retry can return conflict after an
+original success; conflict alone does not establish the outcome of an earlier uncertain attempt.
+
+Validation uses one fresh catalog image. The sequenced `ConfigTransaction` checks an atomic catalog mutation
+watermark before publishing: intervening configuration, ownership, operation or offline lifecycle changes invalidate
+validation, including writes from legacy internal transactions. Membership heartbeats, read barriers and request
+acknowledgments do not invalidate it. Retained records and tombstones reconstruct this watermark on snapshot restore.
+Each controller uses one bounded durable request stream and recovers its sequence and retained result after restart.
+No public session or acknowledgment API is exposed, and there is no automatic mutation retry or precondition rebase.
+
+A successful write returns `status: "OK"`, its committed revision and resulting ETag. Missing preconditions return
+428, stale ETags 412, catalog conflicts 409, invalid configurations 400, and unsupported lifecycle operations 501.
+A transport failure during submission returns 503 with `outcomeUnknown: true`. Keep the original body, precondition
+and creation key for any retry. A subsequent absent read does not prove that an earlier pending write failed.
+Authorization is checked before submission and again before response; losing permission after submission cannot
+undo an already committed mutation.
+
+Schema and table validation reuse Pinot validators with explicit controller-scoped Groovy and assignment policies.
+This surface rejects unknown properties, forced updates, skipped validation and host-local environment/system-property
+templates.
+Schema names must be raw names without a table type suffix; a table name may carry one suffix, preventing repeated
+normalization from changing its authorization identity. Table deletion, REALTIME/hybrid tables, tasks, tiers,
+upsert/dedup, tuning and Helix assignment settings remain unsupported. Changes to a table or schema with native
+segment, assignment or operation state are rejected because those records need lifecycle reconciliation; cluster
+config changes are also rejected while such native state exists. Publishing a table config does not assign segments
+or enable querying.
+
+`/auth/info` identifies the startup-selected backend. The same UI provides native metadata browsing and JSON editing,
+without issuing ZK, task, tenant, segment-status or query requests. Discovery failures block the console rather than
+enabling query routes. Pending edits keep their original body and conditions in memory; only a nonsecret request marker
+survives page reload. An uncertain result requires an explicit retry or fresh inspection and acknowledgment before
+preparing a distinct write. Uploads, rebalance, query execution, IdealState/ExternalView, ZK browsing and minion APIs
+are absent from the native resource set. Swagger is also unavailable on this path because its current shared global
+scanner cannot safely describe the isolated native resource set. Native query listeners stay disabled.
+
+Audit configuration is initialized from committed cluster configuration and refreshed with the existing recoverable
+listener cache. A refresh failure retains the last listener value and makes health unhealthy until recovery; this
+cache is never used to answer REST reads. Under `pinot.coordination.ratis.controller.http.`, `timeoutMs` defaults to
+5000, `configRefreshMs` to 1000 (both bounded to 1–60000), `maxRecords` to 4096, and `maxBytes` to 2097152. Image limits
+bound full reads; oversized images fail rather than returning partial lists. These are component bootstrap limits.
+
## Voter configuration and validation
The voter configuration uses the `pinot.coordination.ratis.` prefix with `clusterId`, `historyId`, `id`, `voters`,
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 d35d1ac3830..7d256786410 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
@@ -31,11 +31,13 @@ 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.GetInstallationWork;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetRevision;
+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.Query;
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.OfflineInstallationProtocol;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
import org.apache.pinot.coordination.ratis.transport.InstallationAccess;
@@ -74,12 +76,18 @@ public final class RatisMetadataAccess implements MetadataAccess, MembershipAcce
@Override
public Result query(Query query, long timeoutMillis)
throws IOException {
- if (query instanceof MembershipQuery || query instanceof GetInstallationWork) {
- throw new IOException("Membership and installation work queries require their typed access method");
+ if (query instanceof MembershipQuery || query instanceof GetInstallationWork || query instanceof GetStreamTip) {
+ throw new IOException("Membership, installation work and stream queries require their typed access method");
}
return MetadataProtocol.decodeResult(_client.query(MetadataProtocol.encode(query), timeoutMillis));
}
+ @Override
+ public StreamTipResult query(GetStreamTip query, long timeoutMillis)
+ throws IOException {
+ return MetadataProtocol.decodeStreamTipResult(_client.query(MetadataProtocol.encode(query), timeoutMillis));
+ }
+
@Override
public MembershipProtocol.Response execute(MembershipCommand command, long timeoutMillis)
throws IOException {
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 6ad366715a2..223f564fbf5 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
@@ -41,12 +41,14 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogReco
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ChangePage;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ChangeTransaction;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Command;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ConfigTransaction;
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.GetInstallationWork;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetOutcome;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetRecord;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetRevision;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetStreamTip;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.InstallationSessionCommand;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.InstallationSessionQuery;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipCommand;
@@ -68,6 +70,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RetireClien
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.SequencedCommand;
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.Transaction;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.InstallationSnapshot;
import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol.WorkResult;
@@ -99,6 +102,8 @@ public final class MetadataCatalog {
private final UUID _clusterId;
private final UUID _historyId;
private long _lastAppliedIndex;
+ // Retained records, including tombstones, reconstruct this fence on restore. Membership-only progress is excluded.
+ private long _catalogMutationIndex;
private Map<UUID, CatalogRecord> _records = new HashMap<>();
private Map<UUID, Integer> _recordBytes = new HashMap<>();
private Map<RecordKey, UUID> _names = new HashMap<>();
@@ -219,6 +224,9 @@ public final class MetadataCatalog {
if (query instanceof GetInstallationWork work) {
return installationWork(work, readIndex);
}
+ if (query instanceof GetStreamTip tip) {
+ return streamTip(tip, readIndex);
+ }
try {
require(_historyId.equals(query.historyId()), Status.HISTORY_MISMATCH, "Metadata history differs");
return encode(switch (query) {
@@ -233,6 +241,7 @@ public final class MetadataCatalog {
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");
});
} catch (Rejection e) {
@@ -244,6 +253,27 @@ public final class MetadataCatalog {
}
}
+ private byte[] streamTip(GetStreamTip query, long index)
+ throws IOException {
+ StreamTipResult result;
+ try {
+ require(_historyId.equals(query.historyId()), Status.HISTORY_MISMATCH, "Metadata history differs");
+ ClientState client = currentClient(query.clientId(), query.clientEpoch());
+ identifier(query.streamId());
+ StreamState stream = client.streams().get(query.streamId());
+ Outcome outcome = stream == null ? null : stream.outcome();
+ result = new StreamTipResult(Status.OK, revision(index), "", client.epoch(),
+ stream == null ? 0 : stream.sequence(), stream == null ? 0 : stream.floor(),
+ outcome == null ? null : outcome.digest(), outcome == null ? null : outcome.result());
+ } catch (Rejection e) {
+ result = new StreamTipResult(e._status, revision(index), e.getMessage(), 0, 0, 0, null, null);
+ }
+ byte[] encoded = MetadataProtocol.encodeValue(result);
+ return encoded.length <= MAX_RESULT_BYTES ? encoded : MetadataProtocol.encodeValue(
+ new StreamTipResult(Status.RESPONSE_TOO_LARGE, revision(index), "Stream tip exceeds the supported size",
+ 0, 0, 0, null, null));
+ }
+
private byte[] installationWork(GetInstallationWork query, long index)
throws IOException {
WorkResult result;
@@ -699,6 +729,7 @@ public final class MetadataCatalog {
require(restored.lastAppliedIndex() >= 0 && restored.records() != null && restored.clients() != null,
Status.INVALID_REQUEST, "Invalid snapshot header");
Map<UUID, CatalogRecord> records = new HashMap<>();
+ long catalogMutationIndex = 0;
Map<UUID, Integer> recordBytes = new HashMap<>();
Map<RecordKey, UUID> names = new HashMap<>();
for (CatalogRecord record : restored.records()) {
@@ -706,6 +737,9 @@ public final class MetadataCatalog {
&& detached.path("formatVersion").asInt() >= minimumSnapshotFormat(record),
Status.INVALID_REQUEST, "Snapshot contains a record family newer than its format");
validateRecord(record, restored.lastAppliedIndex());
+ if (isCatalogMutation(record)) {
+ catalogMutationIndex = Math.max(catalogMutationIndex, record.lastModifiedIndex());
+ }
require(records.putIfAbsent(record.objectId(), record) == null, Status.INVALID_REQUEST,
"Duplicate object ID in snapshot");
int encodedBytes = MetadataProtocol.encodeValue(record).length;
@@ -755,6 +789,7 @@ public final class MetadataCatalog {
_names = names;
_clients = clients;
_lastAppliedIndex = restored.lastAppliedIndex();
+ _catalogMutationIndex = catalogMutationIndex;
_journal = journal;
_journalFloor = restored.journalFloor();
_journalBytes = journalBytes;
@@ -857,13 +892,16 @@ public final class MetadataCatalog {
Result outcome;
try {
switch (command) {
- case Transaction transaction -> {
- require(transaction.mutations() != null && !transaction.mutations().isEmpty()
- && transaction.mutations().size() <= MetadataProtocol.MAX_MUTATIONS, Status.INVALID_REQUEST,
- "Transaction mutation count is outside the supported bounds");
- for (Mutation mutation : transaction.mutations()) {
- staged.add(stage(mutation, index));
- }
+ case Transaction transaction -> stageMutations(transaction.mutations(), index, staged);
+ case ConfigTransaction transaction -> {
+ Revision validated = transaction.validatedSnapshot();
+ require(validated != null && validated.index() >= 0 && validated.index() <= _lastAppliedIndex,
+ Status.INVALID_REQUEST, "Validation snapshot must be an applied catalog revision");
+ require(_historyId.equals(validated.historyId()), Status.HISTORY_MISMATCH,
+ "Validation snapshot history differs");
+ require(_catalogMutationIndex <= validated.index(), Status.CONFLICT,
+ "Catalog changed after validation; read and validate a fresh image");
+ stageMutations(transaction.mutations(), index, staged);
}
case OwnershipCommand ownership -> staged.addAll(
OwnershipCatalog.stage(ownership.request(), coordinationView(), actualTerm, index));
@@ -914,6 +952,9 @@ public final class MetadataCatalog {
int recordIndex = 0;
for (CatalogRecord record : staged) {
_records.put(record.objectId(), record);
+ if (isCatalogMutation(record)) {
+ _catalogMutationIndex = index;
+ }
_recordBytes.put(record.objectId(), encodedSizes.get(recordIndex++));
RecordKey key = new RecordKey(record.kind(), record.name());
if (record.deleted()) {
@@ -927,6 +968,18 @@ public final class MetadataCatalog {
return outcome;
}
+ private static boolean isCatalogMutation(CatalogRecord record) {
+ return record.kind() != RecordKind.MEMBERSHIP_CONTROL && record.kind() != RecordKind.PARTICIPANT;
+ }
+
+ private void stageMutations(List<Mutation> mutations, long index, List<CatalogRecord> staged) {
+ require(mutations != null && !mutations.isEmpty() && mutations.size() <= MetadataProtocol.MAX_MUTATIONS,
+ Status.INVALID_REQUEST, "Transaction mutation count is outside the supported bounds");
+ for (Mutation mutation : mutations) {
+ staged.add(stage(mutation, index));
+ }
+ }
+
private CatalogRecord stage(Mutation mutation, long index) {
require(mutation != null && mutation.action() != null && mutation.kind() != null && mutation.expected() != null,
Status.INVALID_REQUEST, "Mutation type and expected version are required");
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 14fe0675b3e..8d549c96ecc 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
@@ -232,6 +232,7 @@ public final class MetadataProtocol {
@JsonSubTypes({@JsonSubTypes.Type(value = RegisterClient.class, name = "REGISTER_CLIENT"),
@JsonSubTypes.Type(value = RetireClient.class, name = "RETIRE_CLIENT"),
@JsonSubTypes.Type(value = Transaction.class, name = "TRANSACTION"),
+ @JsonSubTypes.Type(value = ConfigTransaction.class, name = "CONFIG_TRANSACTION"),
@JsonSubTypes.Type(value = Acknowledge.class, name = "ACKNOWLEDGE"),
@JsonSubTypes.Type(value = MembershipCommand.class, name = "MEMBERSHIP"),
@JsonSubTypes.Type(value = OwnershipCommand.class, name = "OWNERSHIP"),
@@ -249,8 +250,8 @@ public final class MetadataProtocol {
}
/// All effect-producing controller commands share the same durable request stream and retained outcome contract.
- public sealed interface SequencedCommand extends Command permits Transaction, OwnershipCommand, OperationCommand,
- OfflineSegmentCommand, NativeOperationCommand, NativeInstallationCommand {
+ public sealed interface SequencedCommand extends Command permits Transaction, ConfigTransaction, OwnershipCommand,
+ OperationCommand, OfflineSegmentCommand, NativeOperationCommand, NativeInstallationCommand {
long clientEpoch();
String streamId();
@@ -333,6 +334,16 @@ public final class MetadataProtocol {
}
}
+ /// Config changes validated against a coherent image must fail if any non-membership catalog record changed
+ /// afterward. The distinct command tag preserves legacy transaction replay; older decoders reject this command.
+ /// Validation and authorization remain the caller's responsibility; the catalog atomically fences their read image.
+ public record ConfigTransaction(UUID historyId, String clientId, long clientEpoch, String streamId, long sequence,
+ Revision validatedSnapshot, List<Mutation> mutations) implements SequencedCommand {
+ public ConfigTransaction {
+ mutations = mutations == null ? null : List.copyOf(mutations);
+ }
+ }
+
public record Acknowledge(UUID historyId, String clientId, long clientEpoch, String streamId, long throughSequence)
implements Command {
}
@@ -340,12 +351,13 @@ public final class MetadataProtocol {
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "operation")
@JsonSubTypes({@JsonSubTypes.Type(value = GetRecord.class, name = "GET_RECORD"),
@JsonSubTypes.Type(value = GetOutcome.class, name = "GET_OUTCOME"),
+ @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 = GetInstallationWork.class, name = "GET_INSTALLATION_WORK"),
@JsonSubTypes.Type(value = MembershipQuery.class, name = "MEMBERSHIP")})
- public sealed interface Query permits GetRecord, GetOutcome, GetRevision, MembershipQuery, InstallationSessionQuery,
- GetInstallationWork {
+ public sealed interface Query permits GetRecord, GetOutcome, GetStreamTip, GetRevision, MembershipQuery,
+ InstallationSessionQuery, GetInstallationWork {
UUID historyId();
}
@@ -364,6 +376,16 @@ public final class MetadataProtocol {
implements Query {
}
+ /// Internal authenticated controller recovery; this does not register a stream or acknowledge an outcome.
+ public record GetStreamTip(UUID historyId, String clientId, long clientEpoch, String streamId) implements Query {
+ }
+
+ /// A missing stream has sequence and floor zero. A pending result retains its original revision and digest;
+ /// acknowledged results have no digest/outcome. This response is separate from catalog values.
+ public record StreamTipResult(Status status, Revision revision, String detail, long clientEpoch, long sequence,
+ long floor, String digest, Result outcome) {
+ }
+
public record InstallationSessionQuery(MembershipProtocol.ParticipantToken server, GetOutcome request)
implements Query {
@Override
@@ -416,6 +438,11 @@ public final class MetadataProtocol {
return MAPPER.readValue(bytes, Result.class);
}
+ public static StreamTipResult decodeStreamTipResult(byte[] bytes)
+ throws IOException {
+ return decodeValue(bytes, StreamTipResult.class);
+ }
+
static <T> T decodeValue(byte[] bytes, Class<T> type)
throws IOException {
if (bytes == null || bytes.length == 0 || bytes.length > MAX_CHANGE_BYTES) {
diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataAccess.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataAccess.java
index bd768282af5..324be39bda7 100644
--- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataAccess.java
+++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/transport/MetadataAccess.java
@@ -24,8 +24,10 @@ 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.Query;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.StreamTipResult;
/// Typed authoritative metadata boundary. Implementations provide committed writes and linearizable read barriers;
@@ -35,6 +37,11 @@ public interface MetadataAccess {
Result query(Query query, long timeoutMillis) throws IOException;
+ /// Internal controller stream recovery; transports must authorize the requested client identity.
+ default StreamTipResult query(GetStreamTip query, long timeoutMillis) throws IOException {
+ throw new IOException("Stream recovery is not supported by this metadata access implementation");
+ }
+
CatalogImage image(FeedScope scope, int maxRecords, int maxBytes, long timeoutMillis) throws IOException;
ChangePage changes(FeedScope scope, FeedCursor cursor, int maxTransactions, int maxRecords, int maxBytes,
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/RatisMetadataStoreTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/RatisMetadataStoreTest.java
index acf27d4d555..ca78b2a5d9d 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/RatisMetadataStoreTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/RatisMetadataStoreTest.java
@@ -56,6 +56,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogImag
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ChangePage;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetRecord;
+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;
@@ -183,6 +184,12 @@ public class RatisMetadataStoreTest {
assertEquals(current.records(RecordKind.CLUSTER_CONFIG).values().iterator().next().value()
.path("value").asText(), "second");
assertEquals(current.cursor(), page.next());
+ var tip = access.query(new GetStreamTip(cluster._historyId, "test-client", 1, "main"), 10_000);
+ assertEquals(tip.status(), Status.OK);
+ assertEquals(tip.sequence(), 2L);
+ assertEquals(tip.floor(), 1L);
+ assertEquals(tip.outcome().revision().index(), cluster._lastCommittedIndex);
+ assertTrue(tip.revision().index() > page.next().index());
for (int i = 0; i < 3; i++) {
if (i != follower) {
cluster.closeVoter(i);
diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/ControllerCoordinationTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/ControllerCoordinationTest.java
index 308b1225ca8..e410c8f657b 100644
--- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/ControllerCoordinationTest.java
+++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/ControllerCoordinationTest.java
@@ -48,6 +48,7 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Acknowledge
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.CatalogRecord;
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.ConfigTransaction;
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.GetOutcome;
@@ -179,7 +180,11 @@ public class ControllerCoordinationTest {
public void testAttemptFenceAndAtomicTerminalPublication()
throws Exception {
Fixture fixture = new Fixture();
+ Revision beforeOwnership = fixture.revision();
OwnerToken owner = fixture.assign(fixture._first, 0);
+ Mutation cluster = Mutation.createClusterConfig(fixture._history, UUID.randomUUID(), Map.of());
+ assertEquals(fixture.apply(new ConfigTransaction(fixture._history, "controller", 1, "config-owner", 1,
+ beforeOwnership, List.of(cluster))).status(), Status.CONFLICT);
UUID id = UUID.randomUUID();
assertEquals(fixture.operation(new Submit(id, fixture.input())).status(), Status.OK);
assertEquals(fixture.claim(id, owner).status(), Status.OK);
@@ -210,6 +215,7 @@ public class ControllerCoordinationTest {
OperationCommand complete = fixture.command(new Complete(id, beforeOperation.version(), owner, attempt,
State.SUCCEEDED, new Progress(1, 0, "done"), List.of(effect)));
FeedCursor cursor = fixture._catalog.captureImage(ALL).cursor();
+ Revision beforeEffects = fixture.revision();
Result outcome = fixture.apply(complete);
assertEquals(outcome.status(), Status.OK);
assertEquals(fixture.operationState(id).status(), State.SUCCEEDED);
@@ -221,6 +227,8 @@ public class ControllerCoordinationTest {
assertEquals(publication.transactions().getFirst().records().stream().map(CatalogRecord::kind)
.collect(Collectors.toSet()), Set.of(RecordKind.TABLE_CONFIG, RecordKind.CONTROLLER_OPERATION));
assertEquals(publication.transactions().getFirst().records().size(), 2);
+ assertEquals(fixture.apply(new ConfigTransaction(fixture._history, "controller", 1, "config-effects", 1,
+ beforeEffects, List.of(cluster))).status(), Status.CONFLICT);
assertEquals(fixture.apply(complete), outcome);
assertEquals(fixture.record(RecordKind.TABLE_CONFIG, TABLE).version(), beforeTable.version() + 1);
assertEquals(fixture.operation(new Submit(id, fixture.input())).status(), Status.CONFLICT);
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 21740404019..09e002a7c13 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
@@ -34,10 +34,12 @@ import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.ActivateC
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Acknowledge;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Action;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Command;
+import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ConfigTransaction;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetOutcome;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetRecord;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.GetRevision;
+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;
@@ -47,7 +49,9 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RegisterClient;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RetireClient;
+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.MetadataProtocol.StreamTipResult;
import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Transaction;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -60,6 +64,7 @@ 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.assertNull;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.expectThrows;
@@ -67,6 +72,112 @@ import static org.testng.Assert.expectThrows;
/// Exercises catalog invariants across conditional writes, lost responses, client retirement and snapshot replay.
public class MetadataCatalogTest {
+ @Test
+ public void testValidatedConfigFencesSchemaDeletionAndConcurrentTableCreationAcrossRestore()
+ throws Exception {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("orders")
+ .addSingleValueDimension("id", DataType.STRING).build();
+ TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").build();
+ for (boolean deleteFirst : List.of(false, true)) {
+ Fixture fixture = new Fixture();
+ UUID schemaId = UUID.randomUUID();
+ assertStatus(fixture.transact(Mutation.createSchema(fixture.history(), schemaId, schema)), "OK");
+ Revision validated = fixture.catalog().currentRevision();
+ ConfigTransaction delete = new ConfigTransaction(fixture.history(), "client", 1, "delete", 1, validated,
+ List.of(Mutation.delete(RecordKind.SCHEMA, "orders", new ObjectVersion(fixture.history(), schemaId, 1))));
+ Mutation create = Mutation.createTableConfig(fixture.history(), UUID.randomUUID(), table);
+ ConfigTransaction createTable = new ConfigTransaction(fixture.history(), "client", 1, "create", 1,
+ validated, List.of(create));
+ // Exercise both serializations of the race; a legacy writer must invalidate validation too.
+ JsonNode first = fixture.execute(deleteFirst ? delete : fixture.transaction("legacy", 1, create));
+ assertStatus(first, "OK");
+ ConfigTransaction losing = deleteFirst ? createTable : delete;
+ JsonNode rejected = fixture.execute(losing);
+ assertStatus(rejected, "CONFLICT");
+ assertEquals(fixture.execute(losing), rejected, "Lost failure replies must replay the original result");
+ assertStatus(fixture.read(RecordKind.SCHEMA, "orders"), deleteFirst ? "NOT_FOUND" : "OK");
+ assertStatus(fixture.read(RecordKind.TABLE_CONFIG, "orders_OFFLINE"), deleteFirst ? "NOT_FOUND" : "OK");
+
+ MetadataCatalog restored = new MetadataCatalog(fixture.cluster(), fixture.history());
+ restored.restoreSnapshot(JsonUtils.bytesToJsonNode(JsonUtils.objectToBytes(fixture.catalog().captureSnapshot())));
+ long index = fixture.index();
+ assertEquals(JsonUtils.bytesToJsonNode(restored.execute(MetadataProtocol.encode(losing), ++index)), rejected);
+ ConfigTransaction afterRestart = new ConfigTransaction(fixture.history(), "client", 1, "restart", 1,
+ validated, losing.mutations());
+ assertEquals(MetadataProtocol.decodeResult(restored.execute(MetadataProtocol.encode(afterRestart), ++index))
+ .status(), Status.CONFLICT, "Retained tombstones must reconstruct the validation fence");
+ GetStreamTip query = new GetStreamTip(fixture.history(), "client", 1, losing.streamId());
+ StreamTipResult tip = MetadataProtocol.decodeStreamTipResult(restored.execute(
+ MetadataProtocol.encode(new ReadBarrier(query)), ++index));
+ assertEquals(tip.status(), Status.OK);
+ assertEquals(tip.sequence(), 1L);
+ assertEquals(tip.floor(), 0L);
+ assertEquals(tip.digest(), MetadataProtocol.digest(losing));
+ assertEquals(tip.outcome(), MetadataProtocol.fromTree(rejected, Result.class));
+ assertEquals(tip.revision().index(), index);
+ restored.execute(MetadataProtocol.encode(new Acknowledge(fixture.history(), "client", 1,
+ losing.streamId(), 1)), ++index);
+ StreamTipResult acknowledged = MetadataProtocol.decodeStreamTipResult(
+ restored.query(MetadataProtocol.encode(query)));
+ assertEquals(acknowledged.floor(), acknowledged.sequence());
+ assertNull(acknowledged.digest());
+ assertNull(acknowledged.outcome());
+ assertEquals(MetadataProtocol.decodeResult(restored.execute(MetadataProtocol.encode(losing), ++index))
+ .status(), Status.RESULT_EXPIRED);
+ }
+ }
+
+ @Test
+ public void testValidatedConfigIgnoresHeartbeatAndRequestProgressAndRetainsSuccess()
+ throws Exception {
+ Fixture fixture = new Fixture();
+ UUID id = UUID.randomUUID();
+ assertStatus(fixture.execute(fixture.transaction("main", 1,
+ Mutation.createClusterConfig(fixture.history(), id, Map.of("value", "before")))), "OK");
+ Revision validated = fixture.catalog().currentRevision();
+ assertStatus(fixture.execute(new Acknowledge(fixture.history(), "client", 1, "main", 1)), "OK");
+ assertStatus(fixture.execute(new ReadBarrier(new GetRevision(fixture.history()))), "OK");
+ assertEquals(fixture.member(new ActivateControlEpoch(1, "controller", UUID.randomUUID())).status(), Status.OK);
+ var control = fixture.catalog().membershipControl().controlEpoch();
+ assertEquals(fixture.member(new MembershipProtocol.ConfigurePolicy(control, 0,
+ new MembershipProtocol.Policy(1, 100, 3, 200, 300, 1000))).status(), Status.OK);
+ var participant = fixture.member(new MembershipProtocol.Register("broker", MembershipProtocol.Role.BROKER,
+ UUID.randomUUID(), 0, Map.of(), Set.of())).participant().participant();
+ assertEquals(fixture.member(new MembershipProtocol.Tick(control, 0, 1)).status(), Status.OK);
+ assertEquals(fixture.member(new MembershipProtocol.RenewBatch(List.of(new MembershipProtocol.Renewal(participant,
+ 1, fixture.catalog().currentRevision(), Set.of())))).renewals().getFirst().status(), Status.OK);
+ Mutation update = Mutation.updateClusterConfig(new ObjectVersion(fixture.history(), id, 1),
+ Map.of("value", "after"));
+ ConfigTransaction command = new ConfigTransaction(fixture.history(), "client", 1, "main", 2, validated,
+ List.of(update));
+ assertEquals(MetadataProtocol.decodeCommand(MetadataProtocol.encode(command)), command);
+ JsonNode result = fixture.execute(command);
+ assertStatus(result, "OK");
+ assertEquals(fixture.execute(command), result, "The command's own write must not invalidate its retry");
+ ConfigTransaction changedRequest = new ConfigTransaction(fixture.history(), "client", 1, "main", 2,
+ fixture.catalog().currentRevision(), List.of(update));
+ assertStatus(fixture.execute(changedRequest), "CONFLICT");
+ StreamTipResult tip = MetadataProtocol.decodeStreamTipResult(fixture.catalog().query(MetadataProtocol.encode(
+ new GetStreamTip(fixture.history(), "client", 1, "main"))));
+ assertEquals(tip.digest(), MetadataProtocol.digest(command));
+ assertEquals(tip.outcome(), MetadataProtocol.fromTree(result, Result.class));
+ StreamTipResult absent = MetadataProtocol.decodeStreamTipResult(fixture.catalog().query(MetadataProtocol.encode(
+ new GetStreamTip(fixture.history(), "client", 1, "absent"))));
+ assertEquals(absent.status(), Status.OK);
+ assertEquals(absent.sequence(), 0L);
+ assertEquals(absent.floor(), 0L);
+ assertNull(absent.outcome());
+ assertEquals(MetadataProtocol.decodeStreamTipResult(fixture.catalog().query(MetadataProtocol.encode(
+ new GetStreamTip(fixture.history(), "client", 2, "main")))).status(), Status.FENCED);
+ for (Revision invalid : List.of(new Revision(UUID.randomUUID(), validated.index()),
+ new Revision(fixture.history(), -1), new Revision(fixture.history(), Long.MAX_VALUE))) {
+ String stream = "invalid-" + invalid.index();
+ JsonNode rejected = fixture.execute(new ConfigTransaction(fixture.history(), "client", 1, stream, 1,
+ invalid, List.of(update)));
+ assertStatus(rejected, invalid.historyId().equals(fixture.history()) ? "INVALID_REQUEST" : "HISTORY_MISMATCH");
+ }
+ }
+
@Test
public void testCommittedReadBarrierPreservesOutcomesAndAdvancesCompleteFeed()
throws Exception {
@@ -481,6 +592,12 @@ public class MetadataCatalogTest {
return JsonUtils.bytesToJsonNode(_catalog.execute(MetadataProtocol.encode(command), ++_index));
}
+ private MembershipProtocol.Response member(MembershipProtocol.Operation operation)
+ throws IOException {
+ return MembershipProtocol.decodeResponse(_catalog.execute(MetadataProtocol.encode(
+ new MembershipCommand(_history, "membership", operation)), 1, ++_index));
+ }
+
private JsonNode transact(Mutation... mutations)
throws IOException {
return execute(transaction("s" + _stream++, 1, mutations));
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 4e76120982b..858ef2f933b 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -177,15 +177,23 @@ public final class TableConfigUtils {
public static void validate(TableConfig tableConfig, Schema schema, @Nullable String typesToSkip,
@Nullable TableConfig existingTableConfig) {
+ validate(tableConfig, schema, typesToSkip, existingTableConfig, _disableGroovy, _enforcePoolBasedAssignment);
+ }
+
+ /// Validates with controller-scoped policy without changing the legacy process-wide defaults.
+ public static void validate(TableConfig tableConfig, Schema schema, @Nullable String typesToSkip,
+ @Nullable TableConfig existingTableConfig, boolean disableGroovy, boolean enforcePoolBasedAssignment) {
Preconditions.checkArgument(schema != null, "Schema should not be null for table: %s", tableConfig.getTableName());
Set<ValidationType> skipTypes = parseTypesToSkipString(typesToSkip);
- validateEffectiveTableConfig(tableConfig, schema, skipTypes, existingTableConfig);
+ validateEffectiveTableConfig(tableConfig, schema, skipTypes, existingTableConfig, disableGroovy,
+ enforcePoolBasedAssignment);
if (skipTypes.contains(ValidationType.ALL) || !hasConsumingSegmentTierOverwriteForRealtimeTable(tableConfig)) {
return;
}
try {
TableConfig consumingTableConfig = overwriteTableConfigForConsumingSegmentTier(tableConfig);
- validateEffectiveTableConfig(consumingTableConfig, schema, skipTypes, existingTableConfig);
+ validateEffectiveTableConfig(consumingTableConfig, schema, skipTypes, existingTableConfig, disableGroovy,
+ enforcePoolBasedAssignment);
} catch (RuntimeException e) {
throw new IllegalStateException(
"tierOverwrites.consuming produces an invalid table config: " + e.getMessage(), e);
@@ -193,7 +201,8 @@ public final class TableConfigUtils {
}
private static void validateEffectiveTableConfig(TableConfig tableConfig, Schema schema,
- Set<ValidationType> skipTypes, @Nullable TableConfig existingTableConfig) {
+ Set<ValidationType> skipTypes, @Nullable TableConfig existingTableConfig, boolean disableGroovy,
+ boolean enforcePoolBasedAssignment) {
// Sanitize the table config before validation
sanitize(tableConfig);
@@ -212,7 +221,7 @@ public final class TableConfigUtils {
validateValidationConfig(tableConfig, schema);
validateSegmentAssignmentConfig(tableConfig);
- validateIngestionConfig(tableConfig, schema, existingTableConfig);
+ validateIngestionConfig(tableConfig, schema, existingTableConfig, disableGroovy);
if (tableConfig.getTableType() == TableType.REALTIME) {
validateStreamConfigMaps(tableConfig);
}
@@ -223,13 +232,13 @@ public final class TableConfigUtils {
validateInstanceAssignmentConfigs(tableConfig);
if (!skipTypes.contains(ValidationType.UPSERT)) {
- validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig);
+ validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig, disableGroovy);
}
validateTaskConfig(tableConfig);
validateMaterializedViewInvariants(tableConfig);
- if (_enforcePoolBasedAssignment) {
+ if (enforcePoolBasedAssignment) {
validateInstancePoolsAndReplicaGroups(tableConfig);
}
}
@@ -483,6 +492,11 @@ public final class TableConfigUtils {
@VisibleForTesting
static void validateIngestionConfig(TableConfig tableConfig, Schema schema,
@Nullable TableConfig existingTableConfig) {
+ validateIngestionConfig(tableConfig, schema, existingTableConfig, _disableGroovy);
+ }
+
+ private static void validateIngestionConfig(TableConfig tableConfig, Schema schema,
+ @Nullable TableConfig existingTableConfig, boolean disableGroovy) {
// All metrics-aggregation validation lives here; it returns the columns referenced as aggregation sources, which
// a transform config is allowed to target as its destination (see the transform validation below).
Set<String> aggregationSourceColumns = validateMetricsAggregation(tableConfig, schema);
@@ -548,7 +562,7 @@ public final class TableConfigUtils {
if (filterConfig != null) {
String filterFunction = filterConfig.getFilterFunction();
if (filterFunction != null) {
- if (_disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(filterFunction)) {
+ if (disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(filterFunction)) {
throw new IllegalStateException(
"Groovy filter functions are disabled for table config. Found '" + filterFunction + "'");
}
@@ -583,7 +597,7 @@ public final class TableConfigUtils {
if (enrichmentConfigs != null) {
for (EnrichmentConfig enrichmentConfig : enrichmentConfigs) {
RecordEnricherRegistry.validateEnrichmentConfig(enrichmentConfig,
- new RecordEnricherValidationConfig(_disableGroovy));
+ new RecordEnricherValidationConfig(disableGroovy));
}
}
@@ -604,7 +618,7 @@ public final class TableConfigUtils {
// Skip Groovy expressions when Groovy is disabled: do not compile them just to collect arguments (the main
// loop below rejects Groovy without compiling). Such a config is rejected anyway, so these columns are not
// needed as valid intermediate targets.
- if (transformFunction != null && !(_disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(
+ if (transformFunction != null && !(disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(
transformFunction))) {
try {
transformInputColumns.addAll(
@@ -630,7 +644,7 @@ public final class TableConfigUtils {
+ "' of the transform function must be present in the schema, be consumed as the input of another "
+ "transform function, or be a source column for aggregations");
FunctionEvaluator expressionEvaluator;
- if (_disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
+ if (disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
throw new IllegalStateException(
"Groovy transform functions are disabled for table config. Found '" + transformFunction + "' for column '"
+ columnName + "'");
@@ -1016,6 +1030,11 @@ public final class TableConfigUtils {
static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema,
@Nullable TableConfig existingTableConfig) {
+ validateUpsertAndDedupConfig(tableConfig, schema, existingTableConfig, _disableGroovy);
+ }
+
+ private static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema,
+ @Nullable TableConfig existingTableConfig, boolean disableGroovy) {
boolean upsertEnabled = tableConfig.isUpsertEnabled();
boolean dedupEnabled = tableConfig.isDedupEnabled();
if (!upsertEnabled && !dedupEnabled) {
@@ -1179,7 +1198,7 @@ public final class TableConfigUtils {
Preconditions.checkState(schema.hasColumn(columnName),
"The destination column '%s' of the post-partial-upsert transform function must be present in the "
+ "schema", columnName);
- if (_disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
+ if (disableGroovy && FunctionEvaluatorFactory.isGroovyExpression(transformFunction)) {
throw new IllegalStateException(
"Groovy transform functions are disabled. Found '" + transformFunction + "' for column '"
+ columnName + "' in postPartialUpsertTransformConfigs");
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 66eef1d9794..7d21a836753 100644
--- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -107,6 +107,34 @@ public class TableConfigUtilsTest {
private static final RoutingConfig STRICT_REPLICA_ROUTING_CONFIG =
new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false);
+ @Test
+ public void validateControllerScopedPolicy() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("value", DataType.LONG).build();
+ IngestionConfig ingestion = new IngestionConfig();
+ ingestion.setTransformConfigs(List.of(new TransformConfig("value", "Groovy({x + 1}, x)")));
+ TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setIngestionConfig(ingestion).build();
+ try {
+ TableConfigUtils.setDisableGroovy(false);
+ expectThrows(IllegalStateException.class,
+ () -> TableConfigUtils.validate(table, schema, null, null, true, false));
+ TableConfigUtils.validate(table, schema);
+
+ TableConfigUtils.setDisableGroovy(true);
+ TableConfigUtils.validate(table, schema, null, null, false, false);
+ expectThrows(IllegalStateException.class, () -> TableConfigUtils.validate(table, schema));
+
+ ingestion.setTransformConfigs(null);
+ TableConfigUtils.setEnforcePoolBasedAssignment(true);
+ TableConfigUtils.validate(table, schema, null, null, true, false);
+ expectThrows(IllegalStateException.class, () -> TableConfigUtils.validate(table, schema));
+ } finally {
+ TableConfigUtils.setDisableGroovy(false);
+ TableConfigUtils.setEnforcePoolBasedAssignment(false);
+ }
+ }
+
@Test
public void validateTimeColumnValidationConfig() {
// REALTIME table
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 09bbd583919..00bcd6ca7ea 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
@@ -20,11 +20,16 @@ package org.apache.pinot.tools.service;
import java.io.IOException;
import java.net.ServerSocket;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.KeyStore;
import java.security.cert.X509Certificate;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Base64;
import java.util.HashMap;
@@ -36,12 +41,16 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.coordination.MetadataRpcClient;
import org.apache.pinot.common.coordination.NativeMetadataConfig;
import org.apache.pinot.common.coordination.ServingPermitGuard.ClockReading;
import org.apache.pinot.common.utils.ServiceStatus;
import org.apache.pinot.controller.ControllerStarter;
+import org.apache.pinot.controller.api.access.BasicAuthAccessControlFactory;
+import org.apache.pinot.controller.api.access.ZkBasicAuthAccessControlFactory;
import org.apache.pinot.controller.coordination.RatisControllerProvider;
import org.apache.pinot.coordination.ratis.RatisMetadataStore.MembershipClockSettings;
import org.apache.pinot.coordination.ratis.metadata.MembershipProtocol.Policy;
@@ -52,9 +61,16 @@ import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy;
import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy.ControllerIdentity;
import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy;
import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.DimensionFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.services.ServiceRole;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
@@ -66,7 +82,7 @@ import static org.testng.Assert.expectThrows;
/// Exercises normal service-manager dispatch with real voters and mTLS. No ZooKeeper is started; configured query
-/// ports remain available because these native starters expose metadata and installation membership only.
+/// ports remain available while the controller serves its normal HTTP/UI shell and conditional metadata writes.
public class RatisStartersTest {
private static final String PREFIX = NativeMetadataConfig.PREFIX;
private static final String BROKER = "Broker_native";
@@ -86,7 +102,7 @@ public class RatisStartersTest {
new ServiceStatus.LifecycleServiceStatusCallback(() -> false, () -> false));
try {
CertificateFiles certificate = certificate(directory, "controller");
- int[] ports = ports(4);
+ int[] ports = ports(5);
Map<String, Object> config = tls(UUID.randomUUID().toString(), certificate, certificate.certificate());
config.put(PREFIX + "clusterId", UUID.randomUUID().toString());
config.put(PREFIX + "id", "failure-0");
@@ -96,8 +112,18 @@ public class RatisStartersTest {
config.put(PREFIX + "bootstrap", true);
config.put(PREFIX + "metadata.host", "127.0.0.1");
config.put(PREFIX + "metadata.port", ports[3]);
+ config.put("controller.port", ports[4]);
config.put(PREFIX + "controller.provider.class", TestProvider.class.getName());
config.put(PREFIX + "controller.startup.timeoutMs", 250L);
+ config.put("controller.admin.access.control.factory.class", ZkBasicAuthAccessControlFactory.class.getName());
+ ControllerStarter unsupportedAuth = new ControllerStarter();
+ try {
+ unsupportedAuth.init(new PinotConfiguration(config));
+ expectThrows(IOException.class, unsupportedAuth::start);
+ } finally {
+ unsupportedAuth.stop();
+ }
+ config.remove("controller.admin.access.control.factory.class");
for (int attempt = 0; attempt < 2; attempt++) {
ControllerStarter starter = new ControllerStarter();
try {
@@ -132,7 +158,7 @@ public class RatisStartersTest {
Path trust = directory.resolve("clients.pem");
Files.writeString(trust, Files.readString(controller.certificate()) + Files.readString(broker.certificate())
+ Files.readString(server.certificate()));
- int[] ports = ports(9);
+ int[] ports = ports(12);
String history = UUID.randomUUID().toString();
String cluster = UUID.randomUUID().toString();
String voters = "voter-0@127.0.0.1:" + ports[0] + ",voter-1@127.0.0.1:" + ports[1]
@@ -149,6 +175,23 @@ public class RatisStartersTest {
config.put(PREFIX + "bootstrap", true);
config.put(PREFIX + "metadata.host", "127.0.0.1");
config.put(PREFIX + "metadata.port", ports[3 + i]);
+ config.put(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, "native-test");
+ config.put("pinot.controller.http.server.thread.pool.corePoolSize", 2);
+ config.put("pinot.controller.http.server.thread.pool.maxPoolSize", 4);
+ if (i == 1) {
+ config.put("controller.access.protocols", "https");
+ config.put("controller.access.protocols.https.port", ports[9 + i]);
+ config.put("controller.tls.keystore.type", "PKCS12");
+ config.put("controller.tls.keystore.path", directory.resolve("controller.p12").toString());
+ config.put("controller.tls.keystore.password", "test-password");
+ } else {
+ config.put("controller.port", ports[9 + i]);
+ }
+ if (i < 2) {
+ config.put("controller.admin.access.control.factory.class", BasicAuthAccessControlFactory.class.getName());
+ config.put("controller.admin.access.control.principals", "admin");
+ config.put("controller.admin.access.control.principals.admin.password", "test-password");
+ }
config.put(PREFIX + "controller.provider.class", TestProvider.class.getName());
config.put("test.controllerFingerprint", controller.fingerprint());
config.put("test.brokerFingerprint", broker.fingerprint());
@@ -162,6 +205,78 @@ public class RatisStartersTest {
assertTrue(ServiceStatus.getStatusDescription("Controller_ratis_voter-" + i).contains("HTTP"));
}
+ HttpClient http = httpClient(directory.resolve("controller.p12"));
+ for (int i = 0; i < 3; i++) {
+ String base = (i == 1 ? "https" : "http") + "://127.0.0.1:" + ports[9 + i];
+ assertEquals(get(http, base + "/", false).statusCode(), 200);
+ assertTrue(get(http, base + "/", false).body().contains("<html"));
+ HttpResponse<String> auth = get(http, base + "/auth/info", false);
+ assertEquals(auth.statusCode(), 200);
+ assertEquals(JsonUtils.stringToJsonNode(auth.body()).path("coordinationBackend").asText(), "RATIS");
+ assertEquals(get(http, base + "/schemas", false).statusCode(), i < 2 ? 401 : 200);
+ assertEquals(get(http, base + "/schemas", true).body(), "[]");
+ assertEquals(get(http, base + "/health", false).statusCode(), 200);
+ assertEquals(get(http, base + "/zk/ls", true).statusCode(), 404);
+ assertEquals(get(http, base + "/query/sql", true).statusCode(), 404);
+ assertEquals(get(http, base + "/tables/example/idealstate", true).statusCode(), 404);
+ }
+
+ String controllerUrl = "http://127.0.0.1:" + ports[9];
+ String secondControllerUrl = "https://127.0.0.1:" + ports[10];
+ Schema schema = new Schema.SchemaBuilder().setSchemaName("orders")
+ .addSingleValueDimension("id", DataType.LONG).build();
+ String schemaJson = JsonUtils.objectToString(schema);
+ Map<String, String> createHeaders = Map.of("If-None-Match", "*", "Idempotency-Key", UUID.randomUUID().toString());
+ HttpResponse<String> rejected = write(http, controllerUrl + "/schemas", "POST", schemaJson, Map.of());
+ assertEquals(rejected.statusCode(), 428);
+ assertTrue(JsonUtils.stringToJsonNode(rejected.body()).path("outcomeUnknown").isBoolean(), rejected.body());
+ assertFalse(JsonUtils.stringToJsonNode(rejected.body()).path("outcomeUnknown").asBoolean());
+ HttpResponse<String> created = write(http, controllerUrl + "/schemas", "POST", schemaJson, createHeaders);
+ assertEquals(created.statusCode(), 200, created.body());
+ assertEquals(JsonUtils.stringToJsonNode(created.body()).path("status").asText(), "OK");
+ assertEquals(write(http, secondControllerUrl + "/schemas", "POST", schemaJson, createHeaders).statusCode(), 409,
+ "Repeating a create must not publish another object");
+ String schemaTag = get(http, secondControllerUrl + "/schemas/orders", true).headers()
+ .firstValue("ETag").orElseThrow();
+ schema.addField(new DimensionFieldSpec("label", DataType.STRING, true));
+ HttpResponse<String> updated = write(http, secondControllerUrl + "/schemas/orders", "PUT",
+ JsonUtils.objectToString(schema), Map.of("If-Match", schemaTag));
+ assertEquals(updated.statusCode(), 200, updated.body());
+ assertEquals(write(http, controllerUrl + "/schemas/orders", "PUT", schemaJson,
+ Map.of("If-Match", schemaTag)).statusCode(), 412, "A stale editor must not overwrite newer metadata");
+
+ TableConfig table = new TableConfigBuilder(TableType.OFFLINE).setTableName("orders").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());
+ String tableTag = get(http, secondControllerUrl + "/tables/orders_OFFLINE", true).headers()
+ .firstValue("ETag").orElseThrow();
+ table.getValidationConfig().setReplication("2");
+ updated = write(http, controllerUrl + "/tables/orders_OFFLINE", "PUT", JsonUtils.objectToString(table),
+ Map.of("If-Match", tableTag));
+ assertEquals(updated.statusCode(), 200, updated.body());
+ schemaTag = get(http, controllerUrl + "/schemas/orders", true).headers().firstValue("ETag").orElseThrow();
+ assertEquals(write(http, controllerUrl + "/schemas/orders", "DELETE", "",
+ Map.of("If-Match", schemaTag)).statusCode(), 409, "A referenced schema cannot be deleted");
+ assertEquals(write(http, controllerUrl + "/tables/orders_OFFLINE", "DELETE", "",
+ Map.of("If-Match", tableTag)).statusCode(), 405, "Table cleanup remains unsupported");
+
+ Schema disposable = new Schema.SchemaBuilder().setSchemaName("disposable")
+ .addSingleValueDimension("id", DataType.LONG).build();
+ String disposableJson = JsonUtils.objectToString(disposable);
+ created = write(http, controllerUrl + "/schemas", "POST", disposableJson, createHeaders);
+ assertEquals(created.statusCode(), 200, created.body());
+ String deletedTag = created.headers().firstValue("ETag").orElseThrow();
+ assertEquals(write(http, controllerUrl + "/schemas/disposable", "DELETE", "",
+ Map.of("If-Match", deletedTag)).statusCode(), 200);
+ assertEquals(write(http, secondControllerUrl + "/schemas", "POST", disposableJson, createHeaders).statusCode(),
+ 409, "A lost create reply must not resurrect the same object after deletion");
+ created = write(http, secondControllerUrl + "/schemas", "POST", disposableJson,
+ Map.of("If-None-Match", "*", "Idempotency-Key", UUID.randomUUID().toString()));
+ assertEquals(created.statusCode(), 200, created.body());
+ assertEquals(write(http, controllerUrl + "/schemas/disposable", "DELETE", "",
+ Map.of("If-Match", deletedTag)).statusCode(), 412, "Old ETags must not affect a recreated name");
+
String endpoints = "127.0.0.1:" + ports[3] + ",127.0.0.1:" + ports[4] + ",127.0.0.1:" + ports[5];
Map<String, Object> brokerConfig = tls(history, broker, controller.certificate());
brokerConfig.put(PREFIX + "participantId", BROKER);
@@ -183,6 +298,35 @@ public class RatisStartersTest {
try (MetadataRpcClient metadata = client(ports[4], controller, controller.certificate());
MetadataRpcClient brokerClient = client(ports[4], broker, controller.certificate());
MetadataRpcClient serverClient = client(ports[4], server, controller.certificate())) {
+ assertEquals(metadata.execute(MetadataRpc.CommandRequest.newBuilder().setHistoryId(history)
+ .setRegister(MetadataRpc.RegisterClient.newBuilder().setIncarnation(UUID.randomUUID().toString()))
+ .build()).getStatus(), ResultStatus.OK);
+ assertEquals(metadata.execute(MetadataRpc.CommandRequest.newBuilder().setHistoryId(history)
+ .setTransaction(MetadataRpc.Transaction.newBuilder().setClientEpoch(1).setStreamId("http-test")
+ .setSequence(1).addMutations(MetadataRpc.Mutation.newBuilder()
+ .setAction(MetadataRpc.Mutation.Action.CREATE).setKind(MetadataRpc.RecordKind.CLUSTER_CONFIG)
+ .setName("cluster").setExpected(MetadataRpc.ObjectVersion.newBuilder().setHistoryId(history)
+ .setObjectId(UUID.randomUUID().toString()))
+ .setValue(MetadataRpc.CatalogValue.newBuilder().setClusterConfig(MetadataRpc.ClusterConfig
+ .newBuilder().putProperties("example", "committed")))))
+ .build()).getStatus(), ResultStatus.OK);
+ for (int i = 0; i < 3; i++) {
+ String base = (i == 1 ? "https" : "http") + "://127.0.0.1:" + ports[9 + i];
+ HttpResponse<String> configs = get(http, base + "/cluster/configs", true);
+ assertEquals(configs.statusCode(), 200);
+ assertEquals(JsonUtils.stringToJsonNode(configs.body()).path("example").asText(), "committed");
+ }
+ String clusterTag = get(http, controllerUrl + "/cluster/configs", true).headers()
+ .firstValue("ETag").orElseThrow();
+ updated = write(http, controllerUrl + "/cluster/configs", "POST", "{\"http\":\"committed\"}",
+ Map.of("If-Match", clusterTag));
+ assertEquals(updated.statusCode(), 200, updated.body());
+ HttpResponse<String> clusterConfig = get(http, secondControllerUrl + "/cluster/configs", true);
+ assertEquals(JsonUtils.stringToJsonNode(clusterConfig.body()).path("example").asText(), "committed");
+ assertEquals(JsonUtils.stringToJsonNode(clusterConfig.body()).path("http").asText(), "committed");
+ updated = write(http, secondControllerUrl + "/cluster/configs/http", "DELETE", "",
+ Map.of("If-Match", clusterConfig.headers().firstValue("ETag").orElseThrow()));
+ assertEquals(updated.statusCode(), 200, updated.body());
assertEquals(metadata.read(MetadataRpc.ReadRequest.newBuilder().setHistoryId(history)
.setRevision(MetadataRpc.Empty.getDefaultInstance()).build()).getStatus(), ResultStatus.OK);
MetadataRpc.MembershipReadRequest read = MetadataRpc.MembershipReadRequest.newBuilder()
@@ -197,7 +341,7 @@ public class RatisStartersTest {
assertEquals(serverState.getEndpointsCount(), 0);
assertEquals(ServiceStatus.getServiceStatus(BROKER), ServiceStatus.Status.BAD);
assertEquals(ServiceStatus.getServiceStatus(SERVER), ServiceStatus.Status.GOOD);
- for (int i = 6; i < ports.length; i++) {
+ for (int i = 6; i < 9; i++) {
try (ServerSocket unused = new ServerSocket(ports[i])) {
assertTrue(unused.isBound(), "Native query/admin listener must remain closed");
}
@@ -213,9 +357,33 @@ public class RatisStartersTest {
assertTrue(manager.stopPinotInstanceById("Controller_ratis_voter-0"));
controllers.getFirst().put(PREFIX + "bootstrap", false);
controllers.getFirst().put("test.bootstrapPermitMillis", 2000L);
+ try (ServerSocket occupied = new ServerSocket(ports[9])) {
+ assertTrue(occupied.isBound());
+ expectThrows(RuntimeException.class,
+ () -> manager.startRole(ServiceRole.CONTROLLER, controllers.getFirst()));
+ }
assertEquals(manager.startRole(ServiceRole.CONTROLLER, controllers.getFirst()), "Controller_ratis_voter-0");
assertEquals(metadata.membershipRead(read).getControl().getPolicy().getPermitLifetimeMillis(), 1000L,
"Restart must retain committed policy instead of resetting it from bootstrap configuration");
+ clusterTag = get(http, controllerUrl + "/cluster/configs", true).headers().firstValue("ETag").orElseThrow();
+ updated = write(http, controllerUrl + "/cluster/configs", "POST", "{\"restart\":\"committed\"}",
+ Map.of("If-Match", clusterTag));
+ assertEquals(updated.statusCode(), 200, updated.body());
+ assertEquals(JsonUtils.stringToJsonNode(get(http, secondControllerUrl + "/cluster/configs", true).body())
+ .path("restart").asText(), "committed", "Restart must recover the bounded HTTP writer stream");
+ // Stop participants before losing quorum so the test does not wait for failed retirements.
+ manager.stopPinotInstanceById(BROKER);
+ manager.stopPinotInstanceById(SERVER);
+ assertTrue(manager.stopPinotInstanceById("Controller_ratis_voter-1"));
+ assertTrue(manager.stopPinotInstanceById("Controller_ratis_voter-2"));
+ assertEquals(get(http, "http://127.0.0.1:" + ports[9] + "/cluster/configs", true).statusCode(), 503,
+ "Quorum loss must not return a successful cached catalog response");
+ assertTrue(manager.stopPinotInstanceById("Controller_ratis_voter-0"));
+ for (int i = 9; i < 12; i++) {
+ try (ServerSocket released = new ServerSocket(ports[i])) {
+ assertTrue(released.isBound(), "Controller shutdown must release its HTTP listener");
+ }
+ }
}
} finally {
starters.shutdownNow();
@@ -228,6 +396,40 @@ public class RatisStartersTest {
}
}
+ private static HttpClient httpClient(Path trustFile) throws Exception {
+ KeyStore trust = KeyStore.getInstance("PKCS12");
+ try (var input = Files.newInputStream(trustFile)) {
+ trust.load(input, "test-password".toCharArray());
+ }
+ TrustManagerFactory manager = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ manager.init(trust);
+ SSLContext context = SSLContext.getInstance("TLS");
+ context.init(null, manager.getTrustManagers(), null);
+ return HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).sslContext(context)
+ .connectTimeout(Duration.ofSeconds(5)).build();
+ }
+
+ private static HttpResponse<String> get(HttpClient client, String url, boolean authenticated) throws Exception {
+ HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(10));
+ if (authenticated) {
+ request.header("Authorization", authorization());
+ }
+ return client.send(request.build(), HttpResponse.BodyHandlers.ofString());
+ }
+
+ private static HttpResponse<String> write(HttpClient client, String url, String method, String body,
+ Map<String, String> headers) throws Exception {
+ HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(10))
+ .header("Authorization", authorization()).header("Content-Type", "application/json")
+ .method(method, HttpRequest.BodyPublishers.ofString(body));
+ headers.forEach(request::header);
+ return client.send(request.build(), HttpResponse.BodyHandlers.ofString());
+ }
+
+ private static String authorization() {
+ return "Basic " + Base64.getEncoder().encodeToString("admin:test-password".getBytes(StandardCharsets.UTF_8));
+ }
+
private static Map<String, Object> tls(String history, CertificateFiles certificate, Path trust) {
Map<String, Object> config = new HashMap<>();
config.put("pinot.coordination.backend", "RATIS");
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org