You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/06/05 07:52:22 UTC
[flink] branch release-1.11 updated: [FLINK-18142][hive] Wrong
state names in HiveContinuousMonitoringFunction
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 78b3f48 [FLINK-18142][hive] Wrong state names in HiveContinuousMonitoringFunction
78b3f48 is described below
commit 78b3f48d5d13cb64cecccd833cc1dd18c8d520e9
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Jun 5 15:50:54 2020 +0800
[FLINK-18142][hive] Wrong state names in HiveContinuousMonitoringFunction
This closes #12497
---
.../flink/connectors/hive/read/HiveContinuousMonitoringFunction.java | 4 ++--
.../java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java | 1 +
2 files changed, 3 insertions(+), 2 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
index cdd1e76..893e880 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
@@ -171,13 +171,13 @@ public class HiveContinuousMonitoringFunction
public void initializeState(FunctionInitializationContext context) throws Exception {
this.currReadTimeState = context.getOperatorStateStore().getListState(
new ListStateDescriptor<>(
- "partition-monitoring-state",
+ "current-read-time-state",
LongSerializer.INSTANCE
)
);
this.distinctPartsState = context.getOperatorStateStore().getListState(
new ListStateDescriptor<>(
- "partition-monitoring-state",
+ "distinct-partitions-state",
new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
)
);
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
index 317943a..32a5cd4 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
@@ -464,6 +464,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
final String dbName = "source_db";
final String tblName = "stream_test";
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(100);
StreamTableEnvironment tEnv = HiveTestUtils.createTableEnvWithBlinkPlannerStreamMode(env, SqlDialect.HIVE);
tEnv.registerCatalog(catalogName, hiveCatalog);
tEnv.useCatalog(catalogName);