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:05 UTC
[flink-statefun] 07/07: [FLINK-15818] [kafka-io] Add startup
postition configuration for Kafka ingress YAML support
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 56a05851cb2ac15fd90a1ff94be32f8a75f109d0
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Jan 31 16:36:13 2020 +0800
[FLINK-15818] [kafka-io] Add startup postition configuration for Kafka ingress YAML support
Apart from startup configuration, the following configs are also
supported now through YAML:
* Setting consumer group id
* Setting auto offset reset position
Please see ProtobufKafkaSourceProviderTest for an example YAML
definition with the new configurations.
This closes #6.
---
.../io/kafka/ProtobufKafkaSourceProvider.java | 199 +++++++++++++++++----
.../sdk/kafka/KafkaIngressBuilderApiExtension.java | 25 +++
.../src/test/resources/protobuf-kafka-ingress.yaml | 10 +-
.../statefun/sdk/kafka/KafkaIngressBuilder.java | 4 +-
.../statefun/sdk/kafka/KafkaTopicPartition.java | 32 ++++
5 files changed, 236 insertions(+), 34 deletions(-)
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/ProtobufKafkaSourceProvider.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/ProtobufKafkaSourceProvider.java
index 653caf9..47a7ede 100644
--- a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/ProtobufKafkaSourceProvider.java
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/ProtobufKafkaSourceProvider.java
@@ -18,19 +18,31 @@
package org.apache.flink.statefun.flink.io.kafka;
import com.google.protobuf.Message;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Optional;
import java.util.Properties;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonPointer;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.statefun.flink.common.json.Selectors;
import org.apache.flink.statefun.flink.io.spi.JsonIngressSpec;
import org.apache.flink.statefun.flink.io.spi.SourceProvider;
+import org.apache.flink.statefun.sdk.io.IngressIdentifier;
import org.apache.flink.statefun.sdk.io.IngressSpec;
+import org.apache.flink.statefun.sdk.kafka.KafkaIngressAutoResetPosition;
+import org.apache.flink.statefun.sdk.kafka.KafkaIngressBuilder;
+import org.apache.flink.statefun.sdk.kafka.KafkaIngressBuilderApiExtension;
import org.apache.flink.statefun.sdk.kafka.KafkaIngressDeserializer;
+import org.apache.flink.statefun.sdk.kafka.KafkaIngressSpec;
+import org.apache.flink.statefun.sdk.kafka.KafkaIngressStartupPosition;
+import org.apache.flink.statefun.sdk.kafka.KafkaTopicPartition;
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;
final class ProtobufKafkaSourceProvider implements SourceProvider {
@@ -42,51 +54,176 @@ final class ProtobufKafkaSourceProvider implements SourceProvider {
private static final JsonPointer PROPERTIES_POINTER =
JsonPointer.compile("/ingress/spec/properties");
private static final JsonPointer ADDRESS_POINTER = JsonPointer.compile("/ingress/spec/address");
+ private static final JsonPointer GROUP_ID_POINTER =
+ JsonPointer.compile("/ingress/spec/consumerGroupId");
+ private static final JsonPointer AUTO_RESET_POS_POINTER =
+ JsonPointer.compile("/ingress/spec/autoOffsetResetPosition");
+
+ private static final JsonPointer STARTUP_POS_POINTER =
+ JsonPointer.compile("/ingress/spec/startupPosition");
+ private static final JsonPointer STARTUP_POS_TYPE_POINTER =
+ JsonPointer.compile("/ingress/spec/startupPosition/type");
+ private static final JsonPointer STARTUP_SPECIFIC_OFFSETS_POINTER =
+ JsonPointer.compile("/ingress/spec/startupPosition/offsets");
+ private static final JsonPointer STARTUP_DATE_POINTER =
+ JsonPointer.compile("/ingress/spec/startupPosition/date");
+
+ private static final String STARTUP_DATE_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS Z";
+ private static final DateTimeFormatter STARTUP_DATE_FORMATTER =
+ DateTimeFormatter.ofPattern(STARTUP_DATE_PATTERN);
+
+ private final KafkaSourceProvider delegateProvider = new KafkaSourceProvider();
@Override
public <T> SourceFunction<T> forSpec(IngressSpec<T> spec) {
- JsonNode json = asJsonIngressSpec(spec);
- Properties properties = kafkaClientProperties(json);
- List<String> topics = Selectors.textListAt(json, TOPICS_POINTER);
- KafkaDeserializationSchema<T> deserializationSchema = deserializationSchema(json);
- return new FlinkKafkaConsumer<>(topics, deserializationSchema, properties);
+ KafkaIngressSpec<T> kafkaIngressSpec = asKafkaIngressSpec(spec);
+ return delegateProvider.forSpec(kafkaIngressSpec);
}
- private <T> KafkaDeserializationSchema<T> deserializationSchema(JsonNode json) {
- String descriptorSetPath = Selectors.textAt(json, DESCRIPTOR_SET_POINTER);
- String messageType = Selectors.textAt(json, MESSAGE_TYPE_POINTER);
- // this cast is safe since we validate that the produced message type (T) is assignable to a
- // Message.
- // see asJsonIngressSpec()
- @SuppressWarnings("unchecked")
- KafkaIngressDeserializer<T> deserializer =
- (KafkaIngressDeserializer<T>)
- new ProtobufKafkaIngressDeserializer(descriptorSetPath, messageType);
- return new KafkaDeserializationSchemaDelegate<>(deserializer);
+ private static <T> KafkaIngressSpec<T> asKafkaIngressSpec(IngressSpec<T> spec) {
+ if (!(spec instanceof JsonIngressSpec)) {
+ throw new IllegalArgumentException("Wrong type " + spec.type());
+ }
+ JsonIngressSpec<T> casted = (JsonIngressSpec<T>) spec;
+
+ IngressIdentifier<T> id = casted.id();
+ Class<?> producedType = casted.id().producedType();
+ if (!Message.class.isAssignableFrom(producedType)) {
+ throw new IllegalArgumentException(
+ "ProtocolBuffer based ingress is only able to produce types that derive from "
+ + Message.class.getName()
+ + " but "
+ + producedType.getName()
+ + " is provided.");
+ }
+
+ JsonNode json = casted.json();
+
+ KafkaIngressBuilder<T> kafkaIngressBuilder = KafkaIngressBuilder.forIdentifier(id);
+ kafkaIngressBuilder
+ .withKafkaAddress(kafkaAddress(json))
+ .withProperties(kafkaClientProperties(json))
+ .addTopics(topics(json));
+
+ optionalConsumerGroupId(json).ifPresent(kafkaIngressBuilder::withConsumerGroupId);
+ optionalAutoOffsetResetPosition(json).ifPresent(kafkaIngressBuilder::withAutoResetPosition);
+ optionalStartupPosition(json).ifPresent(kafkaIngressBuilder::withStartupPosition);
+
+ KafkaIngressBuilderApiExtension.withDeserializer(kafkaIngressBuilder, deserializer(json));
+
+ return kafkaIngressBuilder.build();
+ }
+
+ private static List<String> topics(JsonNode json) {
+ return Selectors.textListAt(json, TOPICS_POINTER);
}
private static Properties kafkaClientProperties(JsonNode json) {
Map<String, String> kvs = Selectors.propertiesAt(json, PROPERTIES_POINTER);
Properties properties = new Properties();
- properties.put("bootstrap.servers", Selectors.textAt(json, ADDRESS_POINTER));
kvs.forEach(properties::put);
return properties;
}
- private static JsonNode asJsonIngressSpec(IngressSpec<?> spec) {
- if (!(spec instanceof JsonIngressSpec)) {
- throw new IllegalArgumentException("Wrong type " + spec.type());
+ private static String kafkaAddress(JsonNode json) {
+ return Selectors.textAt(json, ADDRESS_POINTER);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> KafkaIngressDeserializer<T> deserializer(JsonNode json) {
+ String descriptorSetPath = Selectors.textAt(json, DESCRIPTOR_SET_POINTER);
+ String messageType = Selectors.textAt(json, MESSAGE_TYPE_POINTER);
+ // this cast is safe since we validate that the produced message type (T) is assignable to a
+ // Message.
+ // see asJsonIngressSpec()
+ return (KafkaIngressDeserializer<T>)
+ new ProtobufKafkaIngressDeserializer(descriptorSetPath, messageType);
+ }
+
+ private static Optional<String> optionalConsumerGroupId(JsonNode json) {
+ return Selectors.optionalTextAt(json, GROUP_ID_POINTER);
+ }
+
+ private static Optional<KafkaIngressAutoResetPosition> optionalAutoOffsetResetPosition(
+ JsonNode json) {
+ Optional<String> conf = Selectors.optionalTextAt(json, AUTO_RESET_POS_POINTER);
+ if (!conf.isPresent()) {
+ return Optional.empty();
}
- JsonIngressSpec<?> casted = (JsonIngressSpec<?>) spec;
- Class<?> producedType = casted.id().producedType();
- if (!Message.class.isAssignableFrom(producedType)) {
+
+ String autoOffsetResetConfig = conf.get().toUpperCase(Locale.ENGLISH);
+
+ try {
+ return Optional.of(KafkaIngressAutoResetPosition.valueOf(autoOffsetResetConfig));
+ } catch (IllegalArgumentException e) {
throw new IllegalArgumentException(
- "ProtocolBuffer based ingress is able to produce types that derive from "
- + Message.class.getName()
- + " but "
- + producedType.getName()
- + " is provided.");
+ "Invalid autoOffsetResetPosition: "
+ + autoOffsetResetConfig
+ + "; valid values are "
+ + Arrays.toString(KafkaIngressAutoResetPosition.values()),
+ e);
}
- return casted.json();
+ }
+
+ private static Optional<KafkaIngressStartupPosition> optionalStartupPosition(JsonNode json) {
+ if (json.at(STARTUP_POS_POINTER).isMissingNode()) {
+ return Optional.empty();
+ }
+
+ String startupType =
+ Selectors.textAt(json, STARTUP_POS_TYPE_POINTER).toLowerCase(Locale.ENGLISH);
+ switch (startupType) {
+ case "group-offsets":
+ return Optional.of(KafkaIngressStartupPosition.fromGroupOffsets());
+ case "earliest":
+ return Optional.of(KafkaIngressStartupPosition.fromEarliest());
+ case "latest":
+ return Optional.of(KafkaIngressStartupPosition.fromLatest());
+ case "specific-offsets":
+ return Optional.of(
+ KafkaIngressStartupPosition.fromSpecificOffsets(specificOffsetsStartupMap(json)));
+ case "date":
+ return Optional.of(KafkaIngressStartupPosition.fromDate(startupDate(json)));
+ default:
+ throw new IllegalArgumentException(
+ "Invalid startup position type: "
+ + startupType
+ + "; valid values are [group-offsets, earliest, latest, specific-offsets, date]");
+ }
+ }
+
+ private static Map<KafkaTopicPartition, Long> specificOffsetsStartupMap(JsonNode json) {
+ Map<String, Long> kvs = Selectors.longPropertiesAt(json, STARTUP_SPECIFIC_OFFSETS_POINTER);
+ Map<KafkaTopicPartition, Long> offsets = new HashMap<>(kvs.size());
+ kvs.forEach(
+ (partition, offset) ->
+ offsets.put(KafkaTopicPartition.fromString(partition), validateOffsetLong(offset)));
+ return offsets;
+ }
+
+ private static ZonedDateTime startupDate(JsonNode json) {
+ String dateStr = Selectors.textAt(json, STARTUP_DATE_POINTER);
+ try {
+ return ZonedDateTime.parse(dateStr, STARTUP_DATE_FORMATTER);
+ } catch (DateTimeParseException e) {
+ throw new IllegalArgumentException(
+ "Unable to parse date string for startup position: "
+ + dateStr
+ + "; the date should conform to the pattern "
+ + STARTUP_DATE_PATTERN,
+ e);
+ }
+ }
+
+ private static Long validateOffsetLong(Long offset) {
+ if (offset < 0) {
+ throw new IllegalArgumentException(
+ "Invalid offset value: "
+ + offset
+ + "; must be a numeric integer with value between 0 and "
+ + Long.MAX_VALUE);
+ }
+
+ return offset;
}
}
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilderApiExtension.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilderApiExtension.java
new file mode 100644
index 0000000..c87edf5
--- /dev/null
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilderApiExtension.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.statefun.sdk.kafka;
+
+public class KafkaIngressBuilderApiExtension {
+ public static <T> void withDeserializer(
+ KafkaIngressBuilder<T> kafkaIngressBuilder, KafkaIngressDeserializer<T> deserializer) {
+ kafkaIngressBuilder.withDeserializer(deserializer);
+ }
+}
diff --git a/statefun-flink/statefun-flink-io-bundle/src/test/resources/protobuf-kafka-ingress.yaml b/statefun-flink/statefun-flink-io-bundle/src/test/resources/protobuf-kafka-ingress.yaml
index b094866..db1c910 100644
--- a/statefun-flink/statefun-flink-io-bundle/src/test/resources/protobuf-kafka-ingress.yaml
+++ b/statefun-flink/statefun-flink-io-bundle/src/test/resources/protobuf-kafka-ingress.yaml
@@ -19,10 +19,18 @@ ingress:
id: com.mycomp.igal/names
spec:
address: kafka-broker:9092
+ consumerGroupId: greeter
topics:
- names
+ autoOffsetResetPosition: earliest
+ startupPosition:
+ type: specific-offsets
+ offsets:
+ - names/0: 91
+ - names/1: 11
+ - names/2: 8
properties:
- - consumer.group: greeter
+ - foo.config: bar
messageType: org.apache.flink.test.SimpleMessage
descriptorSet: classpath:test.desc
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 3ba4a34..c3e7928 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
@@ -130,8 +130,8 @@ public final class KafkaIngressBuilder<T> {
return this;
}
- /** @return A new {@link IngressSpec}. */
- public IngressSpec<T> build() {
+ /** @return A new {@link KafkaIngressSpec}. */
+ public KafkaIngressSpec<T> build() {
Properties properties = resolveKafkaProperties();
return new KafkaIngressSpec<>(id, properties, topics, deserializer, startupPosition);
diff --git a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaTopicPartition.java b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaTopicPartition.java
index 9985baa..5ee2186 100644
--- a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaTopicPartition.java
+++ b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaTopicPartition.java
@@ -25,6 +25,38 @@ public final class KafkaTopicPartition {
private final String topic;
private final int partition;
+ public static KafkaTopicPartition fromString(String topicAndPartition) {
+ Objects.requireNonNull(topicAndPartition);
+ final int pos = topicAndPartition.lastIndexOf("/");
+ if (pos <= 0 || pos == topicAndPartition.length() - 1) {
+ throw new IllegalArgumentException(
+ topicAndPartition + " does not conform to the <topic>/<partition_id> format");
+ }
+
+ String topic = topicAndPartition.substring(0, pos);
+ Integer partitionId;
+ try {
+ partitionId = Integer.valueOf(topicAndPartition.substring(pos + 1));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid topic partition definition: "
+ + topicAndPartition
+ + "; partition id is expected to be an integer with value between 0 and "
+ + Integer.MAX_VALUE,
+ e);
+ }
+
+ if (partitionId < 0) {
+ throw new IllegalArgumentException(
+ "Invalid topic partition definition: "
+ + topicAndPartition
+ + "; partition id is expected to be an integer with value between 0 and "
+ + Integer.MAX_VALUE);
+ }
+
+ return new KafkaTopicPartition(topic, partitionId);
+ }
+
public KafkaTopicPartition(String topic, int partition) {
this.topic = Objects.requireNonNull(topic);