You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2024/07/11 01:22:51 UTC
(flink) branch master updated (c00fcfb293d -> d04c70fae03)
This is an automated email from the ASF dual-hosted git repository.
ron pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from c00fcfb293d [FLINK-33730][doc] Update the Flink upgrade savepoint compatibility table doc for chinese version
new 140d96c0041 [FLINK-35754][e2e] Fix SqlGatewayE2ECase.testMaterializedTableInFullMode failed due to Internal Server Error
new d04c70fae03 [FLINK-35754][doc] Fix typo in materialized table docs
The 2 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.
Summary of changes:
.../dev/table/materialized-table/statements.md | 2 +-
.../docs/dev/table/materialized-table/overview.md | 4 +--
.../dev/table/materialized-table/statements.md | 8 +++---
.../flink/table/gateway/SqlGatewayE2ECase.java | 13 +++++----
.../flink/core/testutils/CommonTestUtils.java | 33 ++++++++++++++++++++++
5 files changed, 47 insertions(+), 13 deletions(-)
(flink) 01/02: [FLINK-35754][e2e] Fix SqlGatewayE2ECase.testMaterializedTableInFullMode failed due to Internal Server Error
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 140d96c00410d16d56f293b321e73475532878c6
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Fri Jul 5 15:56:52 2024 +0800
[FLINK-35754][e2e] Fix SqlGatewayE2ECase.testMaterializedTableInFullMode failed due to Internal Server Error
---
.../flink/table/gateway/SqlGatewayE2ECase.java | 13 +++++----
.../flink/core/testutils/CommonTestUtils.java | 33 ++++++++++++++++++++++
2 files changed, 40 insertions(+), 6 deletions(-)
diff --git a/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java b/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java
index d65ed2f5f5d..0d7387b5803 100644
--- a/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java
+++ b/flink-end-to-end-tests/flink-sql-gateway-test/src/test/java/org/apache/flink/table/gateway/SqlGatewayE2ECase.java
@@ -187,7 +187,7 @@ public class SqlGatewayE2ECase extends TestLogger {
gatewayRestClient.executeStatementWithResult("SET 'execution.runtime-mode' = 'batch'");
// verify the result
- CommonTestUtils.waitUtil(
+ CommonTestUtils.waitUntilIgnoringExceptions(
() -> {
List<RowData> result =
gatewayRestClient.executeStatementWithResult(
@@ -219,7 +219,7 @@ public class SqlGatewayE2ECase extends TestLogger {
"ALTER MATERIALIZED TABLE my_materialized_table_in_continuous_mode RESUME");
// verify the result
- CommonTestUtils.waitUtil(
+ CommonTestUtils.waitUntilIgnoringExceptions(
() -> {
List<RowData> result =
gatewayRestClient.executeStatementWithResult(
@@ -312,7 +312,7 @@ public class SqlGatewayE2ECase extends TestLogger {
// verify the materialized table should auto refresh the today partition or tomorrow
// partition
- CommonTestUtils.waitUtil(
+ CommonTestUtils.waitUntilIgnoringExceptions(
() -> {
List<RowData> result =
gatewayRestClient.executeStatementWithResult(
@@ -345,9 +345,10 @@ public class SqlGatewayE2ECase extends TestLogger {
gatewayRestClient.executeStatementWithResult(
"ALTER MATERIALIZED TABLE my_materialized_table_in_full_mode RESUME");
- // wait until the materialized table is updated and verify only today or tomorrow data
+ // wait until the materialized table is updated and verify only today or tomorrow
+ // data
// should be updated
- CommonTestUtils.waitUtil(
+ CommonTestUtils.waitUntilIgnoringExceptions(
() -> {
List<RowData> result =
gatewayRestClient.executeStatementWithResult(
@@ -378,7 +379,7 @@ public class SqlGatewayE2ECase extends TestLogger {
+ "')");
// verify the materialized table that all partitions are updated
- CommonTestUtils.waitUtil(
+ CommonTestUtils.waitUntilIgnoringExceptions(
() -> {
List<RowData> result =
gatewayRestClient.executeStatementWithResult(
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java
index 7466648a904..70e474e9e86 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java
@@ -19,6 +19,8 @@
package org.apache.flink.core.testutils;
import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
@@ -42,6 +44,8 @@ import static org.hamcrest.MatcherAssert.assertThat;
/** This class contains reusable utility methods for unit tests. */
public class CommonTestUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(CommonTestUtils.class);
+
/**
* Creates a copy of an object via Java Serialization.
*
@@ -215,6 +219,35 @@ public class CommonTestUtils {
}
}
+ /**
+ * Wait until the given condition is met or timeout, ignoring any exceptions thrown by the
+ * condition.
+ *
+ * @param condition the condition to wait for.
+ * @param timeout the maximum time to wait for the condition to become true.
+ * @param pause delay between condition checks.
+ * @param errorMsg the error message to include in the <code>TimeoutException</code> if the
+ * condition was not met before timeout.
+ * @throws TimeoutException if the condition is not met before timeout.
+ * @throws InterruptedException if the thread is interrupted.
+ */
+ @SuppressWarnings("BusyWait")
+ public static void waitUntilIgnoringExceptions(
+ Supplier<Boolean> condition, Duration timeout, Duration pause, String errorMsg)
+ throws TimeoutException, InterruptedException {
+ Supplier<Boolean> safeCondition =
+ () -> {
+ try {
+ return condition.get();
+ } catch (Exception ignored) {
+ LOG.warn("Exception thrown while evaluating condition", ignored);
+ return false;
+ }
+ };
+
+ waitUtil(safeCondition, timeout, pause, errorMsg);
+ }
+
/**
* Wait util the given condition is met or timeout.
*
(flink) 02/02: [FLINK-35754][doc] Fix typo in materialized table docs
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d04c70fae033bb04ed7a3bb69832be55b4425700
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Mon Jul 8 10:06:36 2024 +0800
[FLINK-35754][doc] Fix typo in materialized table docs
---
docs/content.zh/docs/dev/table/materialized-table/statements.md | 2 +-
docs/content/docs/dev/table/materialized-table/overview.md | 4 ++--
docs/content/docs/dev/table/materialized-table/statements.md | 8 ++++----
3 files changed, 7 insertions(+), 7 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/materialized-table/statements.md b/docs/content.zh/docs/dev/table/materialized-table/statements.md
index ceda0c92364..1838845d708 100644
--- a/docs/content.zh/docs/dev/table/materialized-table/statements.md
+++ b/docs/content.zh/docs/dev/table/materialized-table/statements.md
@@ -96,7 +96,7 @@ CREATE MATERIALIZED TABLE my_materialized_table
如上例所示，我们为 `ds` 分区列指定了 `date-formatter` 选项。每次调度时，调度时间将转换为相应的 `ds` 分区值。例如，在 `2024-01-01 00:00:00` 的调度时间下，只有分区 `ds = '2024-01-01'` 会被刷新。
<span class="label label-danger">注意</span>
-- `partition.fields.#.date-formatter` 选项仅适用于全量模式。
+- [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) 选项仅适用于全量模式。
- [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) 中的字段必须是有效的字符串类型分区字段。
## FRESHNESS
diff --git a/docs/content/docs/dev/table/materialized-table/overview.md b/docs/content/docs/dev/table/materialized-table/overview.md
index a2e6c830773..d49492e314a 100644
--- a/docs/content/docs/dev/table/materialized-table/overview.md
+++ b/docs/content/docs/dev/table/materialized-table/overview.md
@@ -45,8 +45,8 @@ Data freshness is a crucial attribute of a materialized table, serving two main
- CONTINUOUS mode: Launches a Flink streaming job that continuously refreshes the materialized table data.
- FULL mode: The workflow scheduler periodically triggers a Flink batch job to refresh the materialized table data.
- **Determining the Refresh Frequency**.
- - In CONTINUOUS mode, data freshness is converted into the `checkpoint` interval of the Flink streaming job currently.
- - In FULL mode, data freshness is converted into the scheduling cycle of the workflow, e.g. cron expression.
+ - In CONTINUOUS mode, data freshness is converted into the `checkpoint` interval of the Flink streaming job.
+ - In FULL mode, data freshness is converted into the scheduling cycle of the workflow, e.g., a cron expression.
## Refresh Mode
diff --git a/docs/content/docs/dev/table/materialized-table/statements.md b/docs/content/docs/dev/table/materialized-table/statements.md
index 18d90675bb9..490652d0732 100644
--- a/docs/content/docs/dev/table/materialized-table/statements.md
+++ b/docs/content/docs/dev/table/materialized-table/statements.md
@@ -28,7 +28,7 @@ under the License.
Flink SQL supports the following Materialized Table statements for now:
- [CREATE MATERIALIZED TABLE](#create-materialized-table)
-- [Alter MATERIALIZED TABLE](#alter-materialized-table)
+- [ALTER MATERIALIZED TABLE](#alter-materialized-table)
- [DROP MATERIALIZED TABLE](#drop-materialized-table)
# CREATE MATERIALIZED TABLE
@@ -60,7 +60,7 @@ AS <select_statement>
## PARTITIONED BY
-`PARTITIONED BY` define an optional list of columns to partition the materialized table. A directory is created for each partition if this materialized table is used as a filesystem sink.
+`PARTITIONED BY` defines an optional list of columns to partition the materialized table. A directory is created for each partition if this materialized table is used as a filesystem sink.
**Example:**
@@ -96,12 +96,12 @@ CREATE MATERIALIZED TABLE my_materialized_table
As shown in the above example, we specified the date-formatter option for the `ds` partition column. During each scheduling, the scheduling time will be converted to the ds partition value. For example, for a scheduling time of `2024-01-01 00:00:00`, only the partition `ds = '2024-01-01'` will be refreshed.
<span class="label label-danger">Note</span>
-- The `partition.fields.#.date-formatter` option only works in full mode.
+- The [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) option only works in full mode.
- The field in the [partition.fields.#.date-formatter]({{< ref "docs/dev/table/config" >}}#partition-fields-date-formatter) must be a valid string type partition field.
## FRESHNESS
-`FRESHNESS` define the data freshness of a materialized table.
+`FRESHNESS` defines the data freshness of a materialized table.
**FRESHNESS and Refresh Mode Relationship**