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:07:59 UTC
[flink-statefun] 01/07: [FLINK-15769] [kafka-io] Add named method
for configuring Kafka group id
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 766d0f1276099c5bffc460614fc1409abc9330eb
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sun Jan 26 17:55:59 2020 +0800
[FLINK-15769] [kafka-io] Add named method for configuring Kafka group id
---
.../statefun/flink/io/kafka/KafkaSourceProvider.java | 16 ++++++++++++----
.../flink/statefun/sdk/kafka/KafkaIngressBuilder.java | 10 +++++++++-
.../flink/statefun/sdk/kafka/KafkaIngressSpec.java | 12 +++++++++++-
3 files changed, 32 insertions(+), 6 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 cbb8a15..d165341 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
@@ -26,16 +26,14 @@ 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 = new Properties();
- properties.putAll(spec.properties());
- properties.put("bootstrap.servers", spec.kafkaAddress());
+ Properties properties = createPropertiesFromSpec(spec);
return new FlinkKafkaConsumer<>(spec.topics(), deserializationSchemaFromSpec(spec), properties);
}
@@ -50,6 +48,16 @@ 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());
+ 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 a997492..fb95ad0 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
@@ -34,6 +34,7 @@ public final class KafkaIngressBuilder<T> {
private final IngressIdentifier<T> id;
private final List<String> topics = new ArrayList<>();
private final Properties properties = new Properties();
+ private String consumerGroupId;
private Class<? extends KafkaIngressDeserializer<T>> deserializerClass;
private String kafkaAddress;
@@ -50,6 +51,12 @@ public final class KafkaIngressBuilder<T> {
return new KafkaIngressBuilder<>(id);
}
+ /** @param consumerGroupId the consumer group id to use. */
+ public KafkaIngressBuilder<T> withConsumerGroupId(String consumerGroupId) {
+ this.consumerGroupId = Objects.requireNonNull(consumerGroupId);
+ return this;
+ }
+
/** @param kafkaAddress Comma separated addresses of the brokers. */
public KafkaIngressBuilder<T> withKafkaAddress(String kafkaAddress) {
this.kafkaAddress = Objects.requireNonNull(kafkaAddress);
@@ -92,6 +99,7 @@ public final class KafkaIngressBuilder<T> {
/** @return A new {@link IngressSpec}. */
public IngressSpec<T> build() {
- return new KafkaIngressSpec<>(id, kafkaAddress, properties, topics, deserializerClass);
+ return new KafkaIngressSpec<>(
+ id, kafkaAddress, properties, topics, deserializerClass, consumerGroupId);
}
}
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 24db6ac..6c4bdd5 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,7 +19,9 @@ 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;
@@ -31,17 +33,21 @@ public class KafkaIngressSpec<T> implements IngressSpec<T> {
private final Class<? extends KafkaIngressDeserializer<T>> deserializerClass;
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) {
+ Class<? extends KafkaIngressDeserializer<T>> deserializerClass,
+ String consumerGroupId) {
this.kafkaAddress = Objects.requireNonNull(kafkaAddress);
this.properties = Objects.requireNonNull(properties);
this.topics = Objects.requireNonNull(topics);
this.deserializerClass = Objects.requireNonNull(deserializerClass);
this.ingressIdentifier = Objects.requireNonNull(id);
+ this.consumerGroupId = consumerGroupId;
}
@Override
@@ -69,4 +75,8 @@ public class KafkaIngressSpec<T> implements IngressSpec<T> {
public Class<? extends KafkaIngressDeserializer<T>> deserializerClass() {
return deserializerClass;
}
+
+ public Optional<String> consumerGroupId() {
+ return (consumerGroupId == null) ? Optional.empty() : Optional.of(consumerGroupId);
+ }
}