You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/03/09 07:09:39 UTC
(paimon) branch master updated: [rest] Introduce reset consumer api in rest (#7372)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new dd27fcde82 [rest] Introduce reset consumer api in rest (#7372)
dd27fcde82 is described below
commit dd27fcde826143c443ae6d92f9dd84f656a4beb5
Author: xuzifu666 <12...@qq.com>
AuthorDate: Mon Mar 9 15:09:34 2026 +0800
[rest] Introduce reset consumer api in rest (#7372)
---
docs/static/rest-catalog-open-api.yaml | 112 +++++++++++++++++++++
.../main/java/org/apache/paimon/rest/RESTApi.java | 21 ++++
.../java/org/apache/paimon/rest/ResourcePaths.java | 12 +++
.../paimon/rest/requests/ResetConsumerRequest.java | 62 ++++++++++++
.../java/org/apache/paimon/catalog/Catalog.java | 16 +++
.../java/org/apache/paimon/rest/RESTCatalog.java | 13 +++
.../org/apache/paimon/rest/RESTCatalogServer.java | 24 +++++
.../org/apache/paimon/rest/RESTCatalogTest.java | 65 ++++++++++++
8 files changed, 325 insertions(+)
diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml
index 0fc27922df..237d9a987e 100644
--- a/docs/static/rest-catalog-open-api.yaml
+++ b/docs/static/rest-catalog-open-api.yaml
@@ -1363,6 +1363,86 @@ paths:
$ref: '#/components/responses/TagNotExistErrorResponse'
"500":
$ref: '#/components/responses/ServerErrorResponse'
+ /v1/{prefix}/databases/{database}/tables/{table}/consumers:
+ get:
+ tags:
+ - table
+ summary: List consumers
+ operationId: listConsumers
+ parameters:
+ - name: prefix
+ in: path
+ required: true
+ schema:
+ type: string
+ - name: database
+ in: path
+ required: true
+ schema:
+ type: string
+ - name: table
+ in: path
+ required: true
+ schema:
+ type: string
+ - name: maxResults
+ in: query
+ schema:
+ type: integer
+ format: int32
+ - name: pageToken
+ in: query
+ schema:
+ type: string
+ responses:
+ "200":
+ description: OK
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ListConsumersResponse'
+ "401":
+ $ref: '#/components/responses/UnauthorizedErrorResponse'
+ "404":
+ $ref: '#/components/responses/TableNotExistErrorResponse'
+ "500":
+ $ref: '#/components/responses/ServerErrorResponse'
+ /v1/{prefix}/databases/{database}/tables/{table}/consumers/reset:
+ post:
+ tags:
+ - table
+ summary: Reset consumer
+ operationId: resetConsumer
+ parameters:
+ - name: prefix
+ in: path
+ required: true
+ schema:
+ type: string
+ - name: database
+ in: path
+ required: true
+ schema:
+ type: string
+ - name: table
+ in: path
+ required: true
+ schema:
+ type: string
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ResetConsumerRequest'
+ responses:
+ "200":
+ description: Success, no content
+ "401":
+ $ref: '#/components/responses/UnauthorizedErrorResponse'
+ "404":
+ $ref: '#/components/responses/TableNotExistErrorResponse'
+ "500":
+ $ref: '#/components/responses/ServerErrorResponse'
/v1/{prefix}/databases/{database}/views:
get:
tags:
@@ -3305,6 +3385,38 @@ components:
format: int64
updatedBy:
type: string
+ ConsumerInfo:
+ type: object
+ properties:
+ consumerId:
+ type: string
+ description: The consumer identifier
+ nextSnapshot:
+ type: integer
+ format: int64
+ description: The next snapshot id of the consumer
+ ListConsumersResponse:
+ type: object
+ properties:
+ consumers:
+ type: array
+ items:
+ $ref: '#/components/schemas/ConsumerInfo'
+ nextPageToken:
+ type: string
+ description: Token for the next page of results
+ ResetConsumerRequest:
+ type: object
+ properties:
+ consumerId:
+ type: string
+ description: The consumer identifier to be reset or deleted
+ nextSnapshotId:
+ type: integer
+ format: int64
+ description: The new next snapshot id of the consumer. If not provided, the consumer will be deleted
+ required:
+ - consumerId
ViewSchema:
type: object
properties:
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index eefc2e3de0..1e549a116c 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -50,6 +50,7 @@ import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest;
import org.apache.paimon.rest.requests.MarkDonePartitionsRequest;
import org.apache.paimon.rest.requests.RegisterTableRequest;
import org.apache.paimon.rest.requests.RenameTableRequest;
+import org.apache.paimon.rest.requests.ResetConsumerRequest;
import org.apache.paimon.rest.requests.RollbackTableRequest;
import org.apache.paimon.rest.responses.AlterDatabaseResponse;
import org.apache.paimon.rest.responses.AuthTableQueryResponse;
@@ -627,6 +628,26 @@ public class RESTApi {
return new PagedList<>(consumers, response.getNextPageToken());
}
+ /**
+ * Reset consumer for table.
+ *
+ * @param identifier database name and table name.
+ * @param consumerId consumer id
+ * @param nextSnapshotId next snapshot id. If null, the consumer will be deleted.
+ * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table not exists
+ * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for
+ * this table
+ */
+ public void resetConsumer(
+ Identifier identifier, String consumerId, @Nullable Long nextSnapshotId) {
+ ResetConsumerRequest request = new ResetConsumerRequest(consumerId, nextSnapshotId);
+ client.post(
+ resourcePaths.resetConsumer(
+ identifier.getDatabaseName(), identifier.getObjectName()),
+ request,
+ restAuthFunction);
+ }
+
/**
* Commit snapshot for table.
*
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
index 2387b36fdf..baf0b8a4af 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java
@@ -273,6 +273,18 @@ public class ResourcePaths {
CONSUMERS);
}
+ public String resetConsumer(String databaseName, String objectName) {
+ return SLASH.join(
+ V1,
+ prefix,
+ DATABASES,
+ encodeString(databaseName),
+ TABLES,
+ encodeString(objectName),
+ CONSUMERS,
+ "reset");
+ }
+
public String tag(String databaseName, String objectName, String tagName) {
return SLASH.join(
V1,
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/ResetConsumerRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/ResetConsumerRequest.java
new file mode 100644
index 0000000000..15bc8d375e
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/requests/ResetConsumerRequest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.paimon.rest.requests;
+
+import org.apache.paimon.rest.RESTRequest;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/** Request for resetting consumer. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ResetConsumerRequest implements RESTRequest {
+
+ private static final String FIELD_CONSUMER_ID = "consumerId";
+ private static final String FIELD_NEXT_SNAPSHOT_ID = "nextSnapshotId";
+
+ @JsonProperty(FIELD_CONSUMER_ID)
+ private final String consumerId;
+
+ @Nullable
+ @JsonProperty(FIELD_NEXT_SNAPSHOT_ID)
+ private final Long nextSnapshotId;
+
+ @JsonCreator
+ public ResetConsumerRequest(
+ @JsonProperty(FIELD_CONSUMER_ID) String consumerId,
+ @Nullable @JsonProperty(FIELD_NEXT_SNAPSHOT_ID) Long nextSnapshotId) {
+ this.consumerId = consumerId;
+ this.nextSnapshotId = nextSnapshotId;
+ }
+
+ @JsonGetter(FIELD_CONSUMER_ID)
+ public String consumerId() {
+ return consumerId;
+ }
+
+ @Nullable
+ @JsonGetter(FIELD_NEXT_SNAPSHOT_ID)
+ public Long nextSnapshotId() {
+ return nextSnapshotId;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index ef4a943e69..3947a950e5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -765,6 +765,22 @@ public interface Catalog extends AutoCloseable {
throw new UnsupportedOperationException("This catalog does not support list consumers");
}
+ /**
+ * Reset consumer for table.
+ *
+ * @param identifier path of the table
+ * @param consumerId consumer id
+ * @param nextSnapshotId next snapshot id. If null, the consumer will be deleted.
+ * @throws TableNotExistException if the table does not exist
+ * @throws UnsupportedOperationException if the catalog does not {@link
+ * #supportsVersionManagement()}
+ */
+ default void resetConsumer(
+ Identifier identifier, String consumerId, @Nullable Long nextSnapshotId)
+ throws TableNotExistException {
+ throw new UnsupportedOperationException("This catalog does not support reset consumer");
+ }
+
/**
* rollback table by the given {@link Identifier} and instant.
*
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 4accdcb02d..7b053d120c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -372,6 +372,19 @@ public class RESTCatalog implements Catalog {
}
}
+ @Override
+ public void resetConsumer(
+ Identifier identifier, String consumerId, @Nullable Long nextSnapshotId)
+ throws TableNotExistException {
+ try {
+ api.resetConsumer(identifier, consumerId, nextSnapshotId);
+ } catch (NoSuchResourceException e) {
+ throw new TableNotExistException(identifier);
+ } catch (ForbiddenException e) {
+ throw new TableNoPermissionException(identifier, e);
+ }
+ }
+
@Override
public boolean supportsListObjectsPaged() {
return true;
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index 1e0a5125be..81cb097f5a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -62,6 +62,7 @@ import org.apache.paimon.rest.requests.CreateViewRequest;
import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest;
import org.apache.paimon.rest.requests.MarkDonePartitionsRequest;
import org.apache.paimon.rest.requests.RenameTableRequest;
+import org.apache.paimon.rest.requests.ResetConsumerRequest;
import org.apache.paimon.rest.requests.RollbackTableRequest;
import org.apache.paimon.rest.responses.AlterDatabaseResponse;
import org.apache.paimon.rest.responses.AuthTableQueryResponse;
@@ -395,6 +396,11 @@ public class RESTCatalogServer {
resources.length == 4
&& ResourcePaths.TABLES.equals(resources[1])
&& ResourcePaths.CONSUMERS.equals(resources[3]);
+ boolean isResetConsumer =
+ resources.length == 5
+ && ResourcePaths.TABLES.equals(resources[1])
+ && ResourcePaths.CONSUMERS.equals(resources[3])
+ && "reset".equals(resources[4]);
boolean isLoadSnapshot =
resources.length == 5
&& ResourcePaths.TABLES.equals(resources[1])
@@ -503,6 +509,8 @@ public class RESTCatalogServer {
return listSnapshots(identifier);
} else if (isListConsumers) {
return listConsumers(identifier);
+ } else if (isResetConsumer) {
+ return resetConsumer(identifier, restAuthParameter.data());
} else if (isLoadSnapshot) {
return loadSnapshot(identifier, resources[4]);
} else if (isTableAuth) {
@@ -807,6 +815,22 @@ public class RESTCatalogServer {
return new MockResponse().setResponseCode(200).setBody(RESTApi.toJson(response));
}
+ private MockResponse resetConsumer(Identifier identifier, String data) throws Exception {
+ ResetConsumerRequest request = RESTApi.fromJson(data, ResetConsumerRequest.class);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ ConsumerManager consumerManager =
+ new ConsumerManager(table.fileIO(), table.location(), "main");
+ if (request.nextSnapshotId() != null) {
+ table.snapshotManager().snapshot(request.nextSnapshotId());
+ consumerManager.resetConsumer(
+ request.consumerId(),
+ new org.apache.paimon.consumer.Consumer(request.nextSnapshotId()));
+ } else {
+ consumerManager.deleteConsumer(request.consumerId());
+ }
+ return new MockResponse().setResponseCode(200);
+ }
+
private MockResponse loadSnapshot(Identifier identifier, String version) throws Exception {
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 8eb3c56aea..d2586b92e1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -2630,6 +2630,71 @@ public abstract class RESTCatalogTest extends CatalogTestBase {
.containsExactlyInAnyOrder(1L, 2L);
}
+ @Test
+ void testResetConsumer() throws Exception {
+ Identifier identifier = Identifier.create("test_table_db", "reset_consumer_table");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+ catalog.createTable(
+ identifier,
+ new Schema(
+ Lists.newArrayList(new DataField(0, "col", DataTypes.INT())),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ emptyMap(),
+ ""),
+ true);
+ FileStoreTable fileStoreTable = (FileStoreTable) catalog.getTable(identifier);
+
+ // Create some snapshots
+ batchWrite(fileStoreTable, singletonList(1));
+ batchWrite(fileStoreTable, singletonList(1));
+ batchWrite(fileStoreTable, singletonList(1));
+
+ // Create consumers
+ ConsumerManager consumerManager =
+ new ConsumerManager(fileStoreTable.fileIO(), fileStoreTable.location());
+ consumerManager.resetConsumer("consumer1", new org.apache.paimon.consumer.Consumer(1));
+ consumerManager.resetConsumer("consumer2", new org.apache.paimon.consumer.Consumer(2));
+
+ // Verify initial state
+ List<ConsumerInfo> consumers =
+ PagedList.listAllFromPagedApi(
+ token ->
+ ((RESTCatalog) catalog)
+ .api()
+ .listConsumersPaged(identifier, null, token));
+ assertThat(consumers).hasSize(2);
+
+ // Test reset consumer with new snapshot id
+ catalog.resetConsumer(identifier, "consumer1", 3L);
+
+ // Verify consumer1 has been reset
+ consumers =
+ PagedList.listAllFromPagedApi(
+ token ->
+ ((RESTCatalog) catalog)
+ .api()
+ .listConsumersPaged(identifier, null, token));
+ assertThat(consumers).hasSize(2);
+ assertThat(consumers)
+ .filteredOn(c -> c.getConsumerId().equals("consumer1"))
+ .extracting(ConsumerInfo::getNextSnapshot)
+ .containsExactly(3L);
+
+ // Test reset consumer with null snapshot id (delete consumer)
+ catalog.resetConsumer(identifier, "consumer2", null);
+
+ // Verify consumer2 has been deleted
+ consumers =
+ PagedList.listAllFromPagedApi(
+ token ->
+ ((RESTCatalog) catalog)
+ .api()
+ .listConsumersPaged(identifier, null, token));
+ assertThat(consumers).hasSize(1);
+ assertThat(consumers).extracting(ConsumerInfo::getConsumerId).containsExactly("consumer1");
+ }
+
@Test
public void testObjectTable() throws Exception {
// create object table