You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2020/01/31 16:08:01 UTC
[flink-statefun] 03/07: [FLINK-15769] [kafka-io] Move properties
resolution to KafkaIngressBuilder
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit 33d91182ec72b8c4e6fa32fd49808027596dd01d
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Thu Jan 30 12:03:19 2020 +0800
[FLINK-15769] [kafka-io] Move properties resolution to KafkaIngressBuilder
Since we now introduce named methods for configuring consumer group id
and auto offset reset position, it would make sense to move the
resolution of the Kafka client properties earlier since the values
provided by the named methods may overwrite any configuration provided
directly via the properties.
---
.../flink/io/kafka/KafkaSourceProvider.java | 18 ++-------------
.../statefun/sdk/kafka/KafkaIngressBuilder.java | 25 ++++++++++++++-------
.../flink/statefun/sdk/kafka/KafkaIngressSpec.java | 26 +---------------------
3 files changed, 20 insertions(+), 49 deletions(-)
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaSourceProvider.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaSourceProvider.java
index 79718d9..1c1793b 100644
--- a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaSourceProvider.java
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaSourceProvider.java
@@ -17,7 +17,6 @@
*/
package org.apache.flink.statefun.flink.io.kafka;
-import java.util.Properties;
import org.apache.flink.statefun.flink.io.common.ReflectionUtil;
import org.apache.flink.statefun.flink.io.spi.SourceProvider;
import org.apache.flink.statefun.sdk.io.IngressSpec;
@@ -26,16 +25,15 @@ import org.apache.flink.statefun.sdk.kafka.KafkaIngressSpec;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
public class KafkaSourceProvider implements SourceProvider {
@Override
public <T> SourceFunction<T> forSpec(IngressSpec<T> ingressSpec) {
KafkaIngressSpec<T> spec = asKafkaSpec(ingressSpec);
- Properties properties = createPropertiesFromSpec(spec);
- return new FlinkKafkaConsumer<>(spec.topics(), deserializationSchemaFromSpec(spec), properties);
+ return new FlinkKafkaConsumer<>(
+ spec.topics(), deserializationSchemaFromSpec(spec), spec.properties());
}
private static <T> KafkaIngressSpec<T> asKafkaSpec(IngressSpec<T> ingressSpec) {
@@ -48,18 +46,6 @@ public class KafkaSourceProvider implements SourceProvider {
throw new IllegalArgumentException(String.format("Wrong type %s", ingressSpec.type()));
}
- private static <T> Properties createPropertiesFromSpec(KafkaIngressSpec<T> spec) {
- Properties properties = new Properties();
- properties.putAll(spec.properties());
- properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.kafkaAddress());
- properties.put(
- ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, spec.autoResetPosition().asKafkaConfig());
- spec.consumerGroupId()
- .ifPresent(groupId -> properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId));
-
- return properties;
- }
-
private <T> KafkaDeserializationSchema<T> deserializationSchemaFromSpec(
KafkaIngressSpec<T> spec) {
KafkaIngressDeserializer<T> ingressDeserializer =
diff --git a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilder.java b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilder.java
index 599292d..d5e5bd5 100644
--- a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilder.java
+++ b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilder.java
@@ -23,6 +23,7 @@ import java.util.Objects;
import java.util.Properties;
import org.apache.flink.statefun.sdk.io.IngressIdentifier;
import org.apache.flink.statefun.sdk.io.IngressSpec;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
/**
* A builder for creating an {@link IngressSpec} for consuming data from Apache Kafka.
@@ -110,13 +111,21 @@ public final class KafkaIngressBuilder<T> {
/** @return A new {@link IngressSpec}. */
public IngressSpec<T> build() {
- return new KafkaIngressSpec<>(
- id,
- kafkaAddress,
- properties,
- topics,
- deserializerClass,
- autoResetPosition,
- consumerGroupId);
+ Properties properties = resolveKafkaProperties();
+
+ return new KafkaIngressSpec<>(id, properties, topics, deserializerClass);
+ }
+
+ private Properties resolveKafkaProperties() {
+ Properties resultProps = new Properties();
+ resultProps.putAll(properties);
+ resultProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);
+ resultProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoResetPosition.asKafkaConfig());
+
+ if (consumerGroupId != null) {
+ resultProps.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
+ }
+
+ return resultProps;
}
}
diff --git a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressSpec.java b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressSpec.java
index 0a69797..94ece25 100644
--- a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressSpec.java
+++ b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressSpec.java
@@ -19,38 +19,26 @@ package org.apache.flink.statefun.sdk.kafka;
import java.util.List;
import java.util.Objects;
-import java.util.Optional;
import java.util.Properties;
-import javax.annotation.Nullable;
import org.apache.flink.statefun.sdk.IngressType;
import org.apache.flink.statefun.sdk.io.IngressIdentifier;
import org.apache.flink.statefun.sdk.io.IngressSpec;
public class KafkaIngressSpec<T> implements IngressSpec<T> {
- private final String kafkaAddress;
private final Properties properties;
private final List<String> topics;
private final Class<? extends KafkaIngressDeserializer<T>> deserializerClass;
- private final KafkaIngressAutoResetPosition autoResetPosition;
private final IngressIdentifier<T> ingressIdentifier;
- @Nullable private final String consumerGroupId;
-
KafkaIngressSpec(
IngressIdentifier<T> id,
- String kafkaAddress,
Properties properties,
List<String> topics,
- Class<? extends KafkaIngressDeserializer<T>> deserializerClass,
- KafkaIngressAutoResetPosition autoResetPosition,
- String consumerGroupId) {
- this.kafkaAddress = Objects.requireNonNull(kafkaAddress);
+ Class<? extends KafkaIngressDeserializer<T>> deserializerClass) {
this.properties = Objects.requireNonNull(properties);
this.topics = Objects.requireNonNull(topics);
this.deserializerClass = Objects.requireNonNull(deserializerClass);
this.ingressIdentifier = Objects.requireNonNull(id);
- this.autoResetPosition = Objects.requireNonNull(autoResetPosition);
- this.consumerGroupId = consumerGroupId;
}
@Override
@@ -63,10 +51,6 @@ public class KafkaIngressSpec<T> implements IngressSpec<T> {
return Constants.KAFKA_INGRESS_TYPE;
}
- public String kafkaAddress() {
- return kafkaAddress;
- }
-
public Properties properties() {
return properties;
}
@@ -78,12 +62,4 @@ public class KafkaIngressSpec<T> implements IngressSpec<T> {
public Class<? extends KafkaIngressDeserializer<T>> deserializerClass() {
return deserializerClass;
}
-
- public KafkaIngressAutoResetPosition autoResetPosition() {
- return autoResetPosition;
- }
-
- public Optional<String> consumerGroupId() {
- return (consumerGroupId == null) ? Optional.empty() : Optional.of(consumerGroupId);
- }
}