You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by jq...@apache.org on 2021/03/29 23:53:37 UTC
[flink] 12/12: [FLINK-20114][connector/kafka] Remove duplicated
warning and remove redundant default value for
partition.discovery.interval.ms
This is an automated email from the ASF dual-hosted git repository.
jqin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8d2fa1bb81eff790bdc21a69d15a17aa074010ae
Author: Dong Lin <li...@gmail.com>
AuthorDate: Wed Mar 24 15:26:23 2021 +0800
[FLINK-20114][connector/kafka] Remove duplicated warning and remove redundant default value for partition.discovery.interval.ms
---
.../apache/flink/connector/kafka/source/KafkaSourceBuilder.java | 8 ++------
.../apache/flink/connector/kafka/source/KafkaSourceOptions.java | 2 +-
.../streaming/runtime/io/StreamMultipleInputProcessorFactory.java | 6 +++++-
3 files changed, 8 insertions(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
index 8ede378..77f844f 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
@@ -442,14 +442,10 @@ public class KafkaSourceBuilder<OUT> {
true);
// If the source is bounded, do not run periodic partition discovery.
- if (maybeOverride(
+ maybeOverride(
KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
"-1",
- boundedness == Boundedness.BOUNDED)) {
- LOG.warn(
- "{} property is overridden to -1 because the source is bounded.",
- KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS);
- }
+ boundedness == Boundedness.BOUNDED);
// If the client id prefix is not set, reuse the consumer group id as the client id prefix.
maybeOverride(
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
index db09fa5..98c735f 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
@@ -36,7 +36,7 @@ public class KafkaSourceOptions {
public static final ConfigOption<Long> PARTITION_DISCOVERY_INTERVAL_MS =
ConfigOptions.key("partition.discovery.interval.ms")
.longType()
- .defaultValue(30000L)
+ .noDefaultValue()
.withDescription(
"The interval in milliseconds for the Kafka source to discover "
+ "the new partitions. A non-positive value disables the partition discovery.");
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java
index a70b112..db351bc 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessorFactory.java
@@ -351,7 +351,11 @@ public class StreamMultipleInputProcessorFactory {
WatermarkGauge inputWatermarkGauge,
MultiStreamStreamStatusTracker streamStatusTracker,
int inputIndex) {
- super(chainedSourceOutput, streamStatusMaintainer, new SimpleCounter(), inputWatermarkGauge);
+ super(
+ chainedSourceOutput,
+ streamStatusMaintainer,
+ new SimpleCounter(),
+ inputWatermarkGauge);
this.streamStatusTracker = streamStatusTracker;
this.inputIndex = inputIndex;
}