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/02/04 03:59:39 UTC
[flink-statefun] branch master updated (55f2ee0 -> 1cdd50b)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git.
from 55f2ee0 [hotfix] [build] Add build status badge to README.md
add a129b5a [FLINK-15839] [kafka-io] Introduce OptionalConfig
add 7735a2d [FLINK-15839] [kafka-io] Use OptionalConfig in KafkaIngressBuilder to handle config resolution
new 7fc0055 [FLINK-15839] [kafka-io] Strengthen Kafka ingress spec validation
new d23aa62 [hotfix] [kafka-io] Omit get-* prefix from getters to conform with code styles
new fb2fb9a [hotfix] [docs] Fix documentation build
new 1cdd50b [hotfix] [doc] Documentation build README mentions incorrect generated PDF name
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
statefun-docs/README.md | 2 +-
statefun-docs/docs/_templates/footer.html | 1 -
.../docs/api_concepts/io_module/kafka.rst | 10 +-
.../docs/api_concepts/io_module/source_sink.rst | 2 +-
statefun-docs/docs/index.rst | 4 +-
.../flink/io/kafka/KafkaSourceProvider.java | 4 +-
statefun-kafka-io/pom.xml | 14 ++
.../sdk/kafka/KafkaIngressAutoResetPosition.java | 3 +-
.../statefun/sdk/kafka/KafkaIngressBuilder.java | 40 ++---
.../flink/statefun/sdk/kafka/KafkaIngressSpec.java | 46 ++++--
.../sdk/kafka/KafkaIngressStartupPosition.java | 91 ++++++++++-
.../flink/statefun/sdk/kafka/OptionalConfig.java | 73 +++++++++
.../sdk/kafka/KafkaIngressBuilderTest.java | 172 +++++++++++++++++++++
.../statefun/sdk/kafka/testutils/Matchers.java | 56 +++++++
14 files changed, 472 insertions(+), 46 deletions(-)
create mode 100644 statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/OptionalConfig.java
create mode 100644 statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressBuilderTest.java
create mode 100644 statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
[flink-statefun] 02/04: [hotfix] [kafka-io] Omit get-* prefix from
getters to conform with code styles
Posted by tz...@apache.org.
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 d23aa629149f1d4169fe98fef94f58da36ea053f
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sat Feb 1 15:20:07 2020 +0800
[hotfix] [kafka-io] Omit get-* prefix from getters to conform with code styles
---
.../org/apache/flink/statefun/flink/io/kafka/KafkaSourceProvider.java | 4 ++--
.../apache/flink/statefun/sdk/kafka/KafkaIngressStartupPosition.java | 4 ++--
.../java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java | 2 --
3 files changed, 4 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 f42f762..1fc61dc 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
@@ -63,10 +63,10 @@ public class KafkaSourceProvider implements SourceProvider {
KafkaIngressStartupPosition.SpecificOffsetsPosition offsetsPosition =
startupPosition.asSpecificOffsets();
consumer.setStartFromSpecificOffsets(
- convertKafkaTopicPartitionMap(offsetsPosition.getSpecificOffsets()));
+ convertKafkaTopicPartitionMap(offsetsPosition.specificOffsets()));
} else if (startupPosition.isDate()) {
KafkaIngressStartupPosition.DatePosition datePosition = startupPosition.asDate();
- consumer.setStartFromTimestamp(datePosition.getEpochMilli());
+ consumer.setStartFromTimestamp(datePosition.epochMilli());
} else {
throw new IllegalStateException("Safe guard; should not occur");
}
diff --git a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressStartupPosition.java b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressStartupPosition.java
index 42754dd..c6cf2f0 100644
--- a/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressStartupPosition.java
+++ b/statefun-kafka-io/src/main/java/org/apache/flink/statefun/sdk/kafka/KafkaIngressStartupPosition.java
@@ -178,7 +178,7 @@ public class KafkaIngressStartupPosition {
this.specificOffsets = specificOffsets;
}
- public Map<KafkaTopicPartition, Long> getSpecificOffsets() {
+ public Map<KafkaTopicPartition, Long> specificOffsets() {
return specificOffsets;
}
@@ -212,7 +212,7 @@ public class KafkaIngressStartupPosition {
this.date = date;
}
- public long getEpochMilli() {
+ public long epochMilli() {
return date.toInstant().toEpochMilli();
}
diff --git a/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java b/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
index 198d946..71ddce9 100644
--- a/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
+++ b/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
@@ -28,8 +28,6 @@ public final class Matchers {
private Matchers() {}
- private Matchers() {}
-
public static <K, V> Matcher<Map<K, V>> isMapOfSize(int size) {
return new TypeSafeMatcher<Map<K, V>>() {
@Override
[flink-statefun] 01/04: [FLINK-15839] [kafka-io] Strengthen Kafka
ingress spec validation
Posted by tz...@apache.org.
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 7fc005555bff7642d50f4672fe0bb18ffb9cb6b1
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sat Feb 1 15:16:23 2020 +0800
[FLINK-15839] [kafka-io] Strengthen Kafka ingress spec validation
With the changes to how we resolve Kafka properties in the
KafkaIngressBuilder, instantiation of a KafkaIngressSpec would be the
time when all properties and spec definitions are "final", and where we
should properly validate them.
This closes #7.
---
.../flink/statefun/sdk/kafka/KafkaIngressSpec.java | 46 +++++++++++++++++-----
.../statefun/sdk/kafka/testutils/Matchers.java | 4 +-
2 files changed, 39 insertions(+), 11 deletions(-)
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 4f5db36..f088c5b 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
@@ -20,7 +20,6 @@ package org.apache.flink.statefun.sdk.kafka;
import java.util.List;
import java.util.Objects;
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;
@@ -39,14 +38,12 @@ public class KafkaIngressSpec<T> implements IngressSpec<T> {
List<String> topics,
KafkaIngressDeserializer<T> deserializer,
KafkaIngressStartupPosition startupPosition) {
- this.properties = Objects.requireNonNull(properties);
- this.topics = Objects.requireNonNull(topics);
+ this.properties = requireValidProperties(properties);
+ this.topics = requireValidTopics(topics);
+ this.startupPosition = requireValidStartupPosition(startupPosition, properties);
+
this.deserializer = Objects.requireNonNull(deserializer);
this.ingressIdentifier = Objects.requireNonNull(id);
-
- validateStartupPosition(
- startupPosition, properties.getProperty(ConsumerConfig.GROUP_ID_CONFIG));
- this.startupPosition = Objects.requireNonNull(startupPosition);
}
@Override
@@ -75,12 +72,41 @@ public class KafkaIngressSpec<T> implements IngressSpec<T> {
return startupPosition;
}
- private static void validateStartupPosition(
- KafkaIngressStartupPosition startupPosition, @Nullable String groupIdConfig) {
- if (startupPosition.isGroupOffsets() && groupIdConfig == null) {
+ private static Properties requireValidProperties(Properties properties) {
+ Objects.requireNonNull(properties);
+
+ if (!properties.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
+ throw new IllegalArgumentException("Missing setting for Kafka address.");
+ }
+
+ // TODO: we eventually want to make the ingress work out-of-the-box without the need to set the
+ // consumer group id
+ if (!properties.containsKey(ConsumerConfig.GROUP_ID_CONFIG)) {
+ throw new IllegalArgumentException("Missing setting for consumer group id.");
+ }
+
+ return properties;
+ }
+
+ private static List<String> requireValidTopics(List<String> topics) {
+ Objects.requireNonNull(topics);
+
+ if (topics.isEmpty()) {
+ throw new IllegalArgumentException("Must define at least one Kafka topic to consume from.");
+ }
+
+ return topics;
+ }
+
+ private static KafkaIngressStartupPosition requireValidStartupPosition(
+ KafkaIngressStartupPosition startupPosition, Properties properties) {
+ if (startupPosition.isGroupOffsets()
+ && !properties.containsKey(ConsumerConfig.GROUP_ID_CONFIG)) {
throw new IllegalStateException(
"The ingress is configured to start from committed consumer group offsets in Kafka, but no consumer group id was set.\n"
+ "Please set the group id with the withConsumerGroupId(String) method.");
}
+
+ return startupPosition;
}
}
diff --git a/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java b/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
index 784e69c..198d946 100644
--- a/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
+++ b/statefun-kafka-io/src/test/java/org/apache/flink/statefun/sdk/kafka/testutils/Matchers.java
@@ -24,7 +24,9 @@ import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeMatcher;
-public class Matchers {
+public final class Matchers {
+
+ private Matchers() {}
private Matchers() {}
[flink-statefun] 04/04: [hotfix] [doc] Documentation build README
mentions incorrect generated PDF name
Posted by tz...@apache.org.
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 1cdd50bb063224249fffa82adeb8e3eb80a5338e
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Feb 4 11:58:06 2020 +0800
[hotfix] [doc] Documentation build README mentions incorrect generated PDF name
---
statefun-docs/README.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/statefun-docs/README.md b/statefun-docs/README.md
index 0e84ca5..98fc261 100644
--- a/statefun-docs/README.md
+++ b/statefun-docs/README.md
@@ -15,4 +15,4 @@ If the build fails or you experience any issues, try doing a `make clean` to rem
# Export the documentation locally as PDF
-`make docker-latexpdf` will export the current brach/commit of the documentation asd PDF. The final documentation will be located under `_build/pdf/latex/ApplicationManager.pdf`.
+`make docker-latexpdf` will export the current brach/commit of the documentation asd PDF. The final documentation will be located under `_build/pdf/latex/StatefulFunctions.pdf`.
[flink-statefun] 03/04: [hotfix] [docs] Fix documentation build
Posted by tz...@apache.org.
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 fb2fb9a1d78b2edd6be2d142f0aeb0c3589eea0e
Author: Seth Wiesman <sj...@gmail.com>
AuthorDate: Mon Feb 3 10:48:58 2020 -0600
[hotfix] [docs] Fix documentation build
This closes #11.
---
statefun-docs/docs/_templates/footer.html | 1 -
statefun-docs/docs/api_concepts/io_module/kafka.rst | 10 +++++-----
statefun-docs/docs/api_concepts/io_module/source_sink.rst | 2 +-
statefun-docs/docs/index.rst | 4 ++--
4 files changed, 8 insertions(+), 9 deletions(-)
diff --git a/statefun-docs/docs/_templates/footer.html b/statefun-docs/docs/_templates/footer.html
index db1a1a0..e3f5748 100644
--- a/statefun-docs/docs/_templates/footer.html
+++ b/statefun-docs/docs/_templates/footer.html
@@ -4,6 +4,5 @@
<p style="font-size:75%;margin-bottom:1px">© Copyright 2019, Apache Software Foundation.</p>
<p style="font-size:75%;margin-bottom:1px">Apache Flink, Flink®, Apache®, the squirrel logo, and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation. </p>
-<p style="font-size:75%;margin-bottom:1px"><a href="mailto:marta@ververica.com">Report an issue with this documentation page</a></p>
{% endblock %}
diff --git a/statefun-docs/docs/api_concepts/io_module/kafka.rst b/statefun-docs/docs/api_concepts/io_module/kafka.rst
index e623f44..c3061df 100644
--- a/statefun-docs/docs/api_concepts/io_module/kafka.rst
+++ b/statefun-docs/docs/api_concepts/io_module/kafka.rst
@@ -56,11 +56,11 @@ It accepts the following arguments:
The ingress allows configuring the startup position to be one of the following:
-*``KafkaIngressStartupPosition#fromGroupOffsets()`` (default): starts from offsets that were committed to Kafka for the specified consumer group.
-*``KafkaIngressStartupPosition#fromEarliest()``: starts from the earliest offset.
-*``KafkaIngressStartupPosition#fromLatest()``: starts from the latest offset.
-*``KafkaIngressStartupPosition#fromSpecificOffsets(Map)``: starts from specific offsets, defined as a map of partitions to their target starting offset.
-*``KafkaIngressStartupPosition#fromDate(Date)``: starts from offsets that have an ingestion time larger than or equal to a specified date.
+* ``KafkaIngressStartupPosition#fromGroupOffsets()`` (default): starts from offsets that were committed to Kafka for the specified consumer group.
+* ``KafkaIngressStartupPosition#fromEarliest()``: starts from the earliest offset.
+* ``KafkaIngressStartupPosition#fromLatest()``: starts from the latest offset.
+* ``KafkaIngressStartupPosition#fromSpecificOffsets(Map)``: starts from specific offsets, defined as a map of partitions to their target starting offset.
+* ``KafkaIngressStartupPosition#fromDate(Date)``: starts from offsets that have an ingestion time larger than or equal to a specified date.
On startup, if the specified startup offset for a partition is out-of-range or does not exist (which may be the case if the ingress is configured to
start from group offsets, specific offsets, or from a date), then the ingress will fallback to using the position configured
diff --git a/statefun-docs/docs/api_concepts/io_module/source_sink.rst b/statefun-docs/docs/api_concepts/io_module/source_sink.rst
index 6fbf10f..6fc963d 100644
--- a/statefun-docs/docs/api_concepts/io_module/source_sink.rst
+++ b/statefun-docs/docs/api_concepts/io_module/source_sink.rst
@@ -44,7 +44,7 @@ A source function spec creates an ingress from a Flink source function.
:lines: 18-
Sink Spec
-===========
+=========
A sink function spec creates an egress from a Flink sink function.
diff --git a/statefun-docs/docs/index.rst b/statefun-docs/docs/index.rst
index 92aa2da..6feab78 100644
--- a/statefun-docs/docs/index.rst
+++ b/statefun-docs/docs/index.rst
@@ -23,9 +23,9 @@
roadmap
contribute
-############################################
+###########################################
Lightweight, Stateful Applications at Scale
-############################################
+###########################################
**Stateful Functions** is an open source framework that reduces the complexity of building and orchestrating distributed stateful applications at scale. It brings together the benefits of stream processing with Apache Flink® and Function-as-a-Service (FaaS) to provide a powerful abstraction for the next generation of event-driven architectures.