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 2024/10/30 09:31:46 UTC
(paimon) branch master updated: [core] Check snapshot when reset consumer (#4373)
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 8c38c4ba1 [core] Check snapshot when reset consumer (#4373)
8c38c4ba1 is described below
commit 8c38c4ba1a948e798a0d63d73d86c98930034192
Author: xuzifu666 <12...@qq.com>
AuthorDate: Wed Oct 30 17:31:42 2024 +0800
[core] Check snapshot when reset consumer (#4373)
---
.../flink/procedure/ResetConsumerProcedure.java | 1 +
.../paimon/flink/action/ResetConsumerAction.java | 1 +
.../flink/procedure/ResetConsumerProcedure.java | 1 +
.../paimon/flink/action/ConsumerActionITCase.java | 52 ++++++++++++++++++++--
.../spark/procedure/ResetConsumerProcedure.java | 1 +
5 files changed, 52 insertions(+), 4 deletions(-)
diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java
index 0355d6dc1..7777ccda1 100644
--- a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java
+++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java
@@ -49,6 +49,7 @@ public class ResetConsumerProcedure extends ProcedureBase {
throws Catalog.TableNotExistException {
FileStoreTable fileStoreTable =
(FileStoreTable) catalog.getTable(Identifier.fromString(tableId));
+ fileStoreTable.snapshotManager().snapshot(nextSnapshotId);
ConsumerManager consumerManager =
new ConsumerManager(
fileStoreTable.fileIO(),
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ResetConsumerAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ResetConsumerAction.java
index 615b448ec..6db8ab4fe 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ResetConsumerAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ResetConsumerAction.java
@@ -57,6 +57,7 @@ public class ResetConsumerAction extends TableActionBase {
if (Objects.isNull(nextSnapshotId)) {
consumerManager.deleteConsumer(consumerId);
} else {
+ dataTable.snapshotManager().snapshot(nextSnapshotId);
consumerManager.resetConsumer(consumerId, new Consumer(nextSnapshotId));
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java
index 5bd4cbaaf..934ce182a 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ResetConsumerProcedure.java
@@ -67,6 +67,7 @@ public class ResetConsumerProcedure extends ProcedureBase {
fileStoreTable.location(),
fileStoreTable.snapshotManager().branch());
if (nextSnapshotId != null) {
+ fileStoreTable.snapshotManager().snapshot(nextSnapshotId);
consumerManager.resetConsumer(consumerId, new Consumer(nextSnapshotId));
} else {
consumerManager.deleteConsumer(consumerId);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ConsumerActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ConsumerActionITCase.java
index ef921ad66..e2243ddf2 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ConsumerActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ConsumerActionITCase.java
@@ -27,6 +27,7 @@ import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.flink.table.api.TableException;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -39,6 +40,7 @@ import static org.apache.flink.table.planner.factories.TestValuesTableFactory.ch
import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.init;
import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.testStreamingRead;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/** IT cases for consumer management actions. */
public class ConsumerActionITCase extends ActionITCaseBase {
@@ -144,6 +146,48 @@ public class ConsumerActionITCase extends ActionITCaseBase {
}
Optional<Consumer> consumer3 = consumerManager.consumer("myid");
assertThat(consumer3).isNotPresent();
+
+ // reset consumer to a not exist snapshot id
+ List<String> args1 =
+ Arrays.asList(
+ "reset_consumer",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--consumer_id",
+ "myid",
+ "--next_snapshot",
+ "10");
+ switch (invoker) {
+ case "action":
+ assertThrows(
+ RuntimeException.class,
+ () -> createAction(ResetConsumerAction.class, args1).run());
+ break;
+ case "procedure_indexed":
+ assertThrows(
+ TableException.class,
+ () ->
+ executeSQL(
+ String.format(
+ "CALL sys.reset_consumer('%s.%s', 'myid', 10)",
+ database, tableName)));
+ break;
+ case "procedure_named":
+ assertThrows(
+ TableException.class,
+ () ->
+ executeSQL(
+ String.format(
+ "CALL sys.reset_consumer(`table` => '%s.%s', consumer_id => 'myid', next_snapshot_id => cast(10 as bigint))",
+ database, tableName)));
+ break;
+ default:
+ throw new UnsupportedOperationException(invoker);
+ }
}
@ParameterizedTest
@@ -206,7 +250,7 @@ public class ConsumerActionITCase extends ActionITCaseBase {
"--consumer_id",
"myid",
"--next_snapshot",
- "1");
+ "3");
// reset consumer
switch (invoker) {
case "action":
@@ -215,13 +259,13 @@ public class ConsumerActionITCase extends ActionITCaseBase {
case "procedure_indexed":
executeSQL(
String.format(
- "CALL sys.reset_consumer('%s.%s', 'myid', 1)",
+ "CALL sys.reset_consumer('%s.%s', 'myid', 3)",
database, branchTableName));
break;
case "procedure_named":
executeSQL(
String.format(
- "CALL sys.reset_consumer(`table` => '%s.%s', consumer_id => 'myid', next_snapshot_id => cast(1 as bigint))",
+ "CALL sys.reset_consumer(`table` => '%s.%s', consumer_id => 'myid', next_snapshot_id => cast(3 as bigint))",
database, branchTableName));
break;
default:
@@ -229,7 +273,7 @@ public class ConsumerActionITCase extends ActionITCaseBase {
}
Optional<Consumer> consumer2 = consumerManager.consumer("myid");
assertThat(consumer2).isPresent();
- assertThat(consumer2.get().nextSnapshot()).isEqualTo(1);
+ assertThat(consumer2.get().nextSnapshot()).isEqualTo(3);
// delete consumer
switch (invoker) {
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ResetConsumerProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ResetConsumerProcedure.java
index a13227e95..0f7fabd05 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ResetConsumerProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ResetConsumerProcedure.java
@@ -90,6 +90,7 @@ public class ResetConsumerProcedure extends BaseProcedure {
if (nextSnapshotId == null) {
consumerManager.deleteConsumer(consumerId);
} else {
+ fileStoreTable.snapshotManager().snapshot(nextSnapshotId);
consumerManager.resetConsumer(consumerId, new Consumer(nextSnapshotId));
}