[flink] Improve Exception message for consumer without expire time

apache · Nov 13, 2024 · 787a981 · 787a981
1 parent 72c25d5
commit 787a981
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 deletions.
diff --git a/.../paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java b/.../paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -259,7 +259,9 @@ public DataStream<RowData> build() {
         if (conf.contains(CoreOptions.CONSUMER_ID)
                 && !conf.contains(CoreOptions.CONSUMER_EXPIRATION_TIME)) {
             throw new IllegalArgumentException(
-                    "consumer.expiration-time should be specified when using consumer-id.");
+                    "You need to configure 'consumer.expiration-time' (ALTER TABLE) and restart your write job for it"
+                            + " to take effect, when you need consumer-id feature. This is to prevent consumers from leaving"
+                            + " too many snapshots that could pose a risk to the file system.");
         }
 
         if (sourceBounded) {

diff --git a/...n-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/...n-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -972,7 +972,8 @@ public void testConsumerIdExpInBatchMode() {
                                         "SELECT * FROM T /*+ OPTIONS('consumer-id' = 'test-id') */ WHERE a = 1"))
                 .rootCause()
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessage("consumer.expiration-time should be specified when using consumer-id.");
+                .hasMessageContaining(
+                        "You need to configure 'consumer.expiration-time' (ALTER TABLE) and restart your write job for it");
     }
 
     @Test
@@ -985,7 +986,8 @@ public void testConsumerIdExpInStreamingMode() {
                                 streamSqlIter(
                                         "SELECT * FROM T /*+ OPTIONS('consumer-id'='test-id') */"))
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessage("consumer.expiration-time should be specified when using consumer-id.");
+                .hasMessageContaining(
+                        "You need to configure 'consumer.expiration-time' (ALTER TABLE) and restart your write job for it");
     }
 
     @Test