You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/12/07 12:42:41 UTC
[flink] branch master updated: [FLINK-20486][hive] Hive temporal
join shouldn't require a minimal 1h monitor interval
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 09ec2ae [FLINK-20486][hive] Hive temporal join shouldn't require a minimal 1h monitor interval
09ec2ae is described below
commit 09ec2ae626acac0c17374a77d49f290ae5289f7d
Author: Leonard Xu <xb...@163.com>
AuthorDate: Mon Dec 7 20:38:32 2020 +0800
[FLINK-20486][hive] Hive temporal join shouldn't require a minimal 1h monitor interval
This closes #14310
---
.../connectors/hive/HiveLookupTableSource.java | 23 +++++++++-------
.../hive/HiveDynamicTableFactoryTest.java | 31 ++++++++++------------
2 files changed, 27 insertions(+), 27 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java
index f6c3e2b..0d2bf1e 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveLookupTableSource.java
@@ -42,6 +42,8 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.ArrayList;
@@ -67,6 +69,7 @@ import static org.apache.flink.table.filesystem.FileSystemOptions.STREAMING_SOUR
*/
public class HiveLookupTableSource extends HiveTableSource implements LookupTableSource {
+ private static final Logger LOG = LoggerFactory.getLogger(HiveLookupTableSource.class);
private static final Duration DEFAULT_LOOKUP_MONITOR_INTERVAL = Duration.ofHours(1L);
private final Configuration configuration;
private Duration hiveTableReloadInterval;
@@ -114,16 +117,16 @@ public class HiveLookupTableSource extends HiveTableSource implements LookupTabl
Duration monitorInterval = configuration.get(STREAMING_SOURCE_MONITOR_INTERVAL) == null
? DEFAULT_LOOKUP_MONITOR_INTERVAL
: configuration.get(STREAMING_SOURCE_MONITOR_INTERVAL);
- Preconditions.checkArgument(
- monitorInterval.toMillis() >= DEFAULT_LOOKUP_MONITOR_INTERVAL.toMillis(),
- String.format(
- "Currently the value of '%s' is required bigger or equal to default value '%s' " +
- "when set '%s' to 'latest', but actual is '%s'",
- STREAMING_SOURCE_MONITOR_INTERVAL.key(),
- DEFAULT_LOOKUP_MONITOR_INTERVAL.toMillis(),
- STREAMING_SOURCE_PARTITION_INCLUDE.key(),
- monitorInterval.toMillis())
- );
+
+ if (monitorInterval.toMillis() < DEFAULT_LOOKUP_MONITOR_INTERVAL.toMillis()) {
+ LOG.warn(String.format(
+ "Currently the recommended value of '%s' is at least '%s' when set '%s' to 'latest'," +
+ " but actual is '%s', this may produce big pressure to hive metastore.",
+ STREAMING_SOURCE_MONITOR_INTERVAL.key(),
+ DEFAULT_LOOKUP_MONITOR_INTERVAL.toMillis(),
+ STREAMING_SOURCE_PARTITION_INCLUDE.key(),
+ monitorInterval.toMillis()));
+ }
hiveTableReloadInterval = monitorInterval;
} else {
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDynamicTableFactoryTest.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDynamicTableFactoryTest.java
index 1edd19f..3baa533 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDynamicTableFactoryTest.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDynamicTableFactoryTest.java
@@ -188,26 +188,23 @@ public class HiveDynamicTableFactoryTest {
assertEquals(configuration1.get(STREAMING_SOURCE_PARTITION_ORDER), "partition-time");
assertEquals(configuration1.get(PARTITION_TIME_EXTRACTOR_KIND), "custom");
assertEquals(configuration1.get(PARTITION_TIME_EXTRACTOR_CLASS), "path.to..TimeExtractor");
- }
- @Test
- public void testInvalidOptions() {
tableEnv.executeSql(String.format(
- "create table table8 (x int, y string, z int)" +
- " tblproperties ('%s' = 'true', '%s' = 'latest', '%s' = '5min')",
- STREAMING_SOURCE_ENABLE.key(),
- STREAMING_SOURCE_PARTITION_INCLUDE.key(),
- STREAMING_SOURCE_MONITOR_INTERVAL.key()));
- try {
- getTableSource("table8");
- } catch (Throwable t) {
- assertTrue(ExceptionUtils.findThrowableWithMessage(t,
- "Currently the value of 'streaming-source.monitor-interval' is required bigger or " +
- "equal to default value '3600000' when set 'streaming-source.partition.include' to 'latest'," +
- " but actual is '300000'")
- .isPresent());
- }
+ "create table table8 (x int, y string, z int)" +
+ " tblproperties ('%s' = 'true', '%s' = 'latest', '%s' = '5min')",
+ STREAMING_SOURCE_ENABLE.key(),
+ STREAMING_SOURCE_PARTITION_INCLUDE.key(),
+ STREAMING_SOURCE_MONITOR_INTERVAL.key()));
+ DynamicTableSource tableSource4 = getTableSource("table8");
+ assertTrue(tableSource4 instanceof HiveLookupTableSource);
+ HiveLookupTableSource lookupTableSource4 = (HiveLookupTableSource) tableSource4;
+ Configuration configuration4 = new Configuration();
+ lookupTableSource4.catalogTable.getOptions().forEach(configuration4::setString);
+ assertEquals(configuration4.get(STREAMING_SOURCE_MONITOR_INTERVAL), Duration.ofMinutes(5L));
+ }
+ @Test
+ public void testInvalidOptions() throws Exception {
tableEnv.executeSql(String.format(
"create table table9 (x int, y string, z int)" +
" tblproperties ('%s' = 'true', '%s' = 'latest', '%s' = '120min', '%s' = '1970-00-01')",