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/03/03 07:45:09 UTC
[flink-statefun] branch master updated (e2305e8 -> 1bfd242)
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 e2305e8 [FLINK-16274] Consolidate constants for Kafka bootstrap server configuration
add 14e264f [FLINK-16123] [e2e] Add statefun-routable-kafka-e2e E2E test module
add 1eabdfb [FLINK-16123] Add RoutableKafkaVerificationModule application for E2E testing
add 9babd73 [FLINK-16123] Support adding classpath files to app image build context in StatefulFunctionsAppContainers
new 1fe23c0 [FLINK-16123] [e2e] Add RoutableKafkaE2E
new 1df8b49 [hotfix] Fix some last dangling incorrect mentions of *ITCase
new fb88505 [FLINK-16395] [legal] Add OkHttp to NOTICE of statefun-flink-distribution
new 1bfd242 [FLINK-16396] [kafka] Add Protobuf message for YAML-ized Kafka egress
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-e2e-tests/pom.xml | 3 +-
.../e2e/common/StatefulFunctionsAppContainers.java | 32 ++++-
.../pom.xml | 10 +-
.../statefun/e2e/routablekafka}/Constants.java | 20 +--
.../e2e/routablekafka/FnSelfAddressTagger.java | 56 ++++++++
.../flink/statefun/e2e/routablekafka/KafkaIO.java | 58 +++++++++
.../RoutableKafkaVerificationModule.java} | 39 ++----
.../protobuf/routable-kafka-verification.proto | 39 +++---
.../e2e/routablekafka/RoutableKafkaE2E.java | 142 +++++++++++++++++++++
.../src/test/resources/Dockerfile | 5 +-
.../src/test/resources/log4j.properties | 0
.../routable-kafka-ingress-module/module.yaml | 39 ++++++
.../flink/statefun/e2e/sanity/Constants.java | 9 +-
.../src/main/resources/META-INF/NOTICE | 1 +
.../src/main/protobuf/kafka-egress.proto | 18 +--
15 files changed, 376 insertions(+), 95 deletions(-)
copy statefun-e2e-tests/{statefun-sanity-e2e => statefun-routable-kafka-e2e}/pom.xml (92%)
copy statefun-e2e-tests/{statefun-sanity-e2e/src/main/java/org/apache/flink/statefun/e2e/sanity => statefun-routable-kafka-e2e/src/main/java/org/apache/flink/statefun/e2e/routablekafka}/Constants.java (57%)
create mode 100644 statefun-e2e-tests/statefun-routable-kafka-e2e/src/main/java/org/apache/flink/statefun/e2e/routablekafka/FnSelfAddressTagger.java
create mode 100644 statefun-e2e-tests/statefun-routable-kafka-e2e/src/main/java/org/apache/flink/statefun/e2e/routablekafka/KafkaIO.java
copy statefun-e2e-tests/{statefun-sanity-e2e/src/main/java/org/apache/flink/statefun/e2e/sanity/SanityVerificationModule.java => statefun-routable-kafka-e2e/src/main/java/org/apache/flink/statefun/e2e/routablekafka/RoutableKafkaVerificationModule.java} (54%)
copy statefun-flink/statefun-flink-core/src/main/protobuf/stateful-functions.proto => statefun-e2e-tests/statefun-routable-kafka-e2e/src/main/protobuf/routable-kafka-verification.proto (65%)
create mode 100644 statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/java/org/apache/flink/statefun/e2e/routablekafka/RoutableKafkaE2E.java
copy statefun-e2e-tests/{statefun-sanity-e2e => statefun-routable-kafka-e2e}/src/test/resources/Dockerfile (78%)
copy statefun-e2e-tests/{statefun-sanity-e2e => statefun-routable-kafka-e2e}/src/test/resources/log4j.properties (100%)
create mode 100644 statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/routable-kafka-ingress-module/module.yaml
copy statefun-examples/statefun-greeter-example/src/main/protobuf/greeter.proto => statefun-flink/statefun-flink-io/src/main/protobuf/kafka-egress.proto (77%)
[flink-statefun] 04/04: [FLINK-16396] [kafka] Add Protobuf message
for YAML-ized Kafka egress
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 1bfd242bc471c16e680f145eb54b85da39120cae
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Mar 3 11:28:42 2020 +0800
[FLINK-16396] [kafka] Add Protobuf message for YAML-ized Kafka egress
This closes #45.
---
.../src/main/protobuf/kafka-egress.proto | 29 ++++++++++++++++++++++
1 file changed, 29 insertions(+)
diff --git a/statefun-flink/statefun-flink-io/src/main/protobuf/kafka-egress.proto b/statefun-flink/statefun-flink-io/src/main/protobuf/kafka-egress.proto
new file mode 100644
index 0000000..9284e1c
--- /dev/null
+++ b/statefun-flink/statefun-flink-io/src/main/protobuf/kafka-egress.proto
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+syntax = "proto3";
+
+package org.apache.flink.statefun.flink.io;
+option java_package = "org.apache.flink.statefun.flink.io.generated";
+option java_multiple_files = true;
+
+message KafkaProducerRecord {
+ string key = 1;
+ bytes value_bytes = 2;
+ string topic = 3;
+}
[flink-statefun] 02/04: [hotfix] Fix some last dangling incorrect
mentions of *ITCase
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 1df8b4919dd160801022b393ccdf4c6c73a66da4
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Wed Feb 26 18:44:40 2020 +0800
[hotfix] Fix some last dangling incorrect mentions of *ITCase
---
statefun-e2e-tests/pom.xml | 2 +-
.../java/org/apache/flink/statefun/e2e/sanity/Constants.java | 9 ++++-----
2 files changed, 5 insertions(+), 6 deletions(-)
diff --git a/statefun-e2e-tests/pom.xml b/statefun-e2e-tests/pom.xml
index eaee050..ab70e22 100644
--- a/statefun-e2e-tests/pom.xml
+++ b/statefun-e2e-tests/pom.xml
@@ -38,7 +38,7 @@ under the License.
<profile>
<!--
Build profile for running end-to-end tests in this module.
- This first builds the base Stateful Functions image before running the ITCases.
+ This first builds the base Stateful Functions image before running the E2E tests.
Docker must be installed for this profile to work.
-->
<id>run-e2e-tests</id>
diff --git a/statefun-e2e-tests/statefun-sanity-e2e/src/main/java/org/apache/flink/statefun/e2e/sanity/Constants.java b/statefun-e2e-tests/statefun-sanity-e2e/src/main/java/org/apache/flink/statefun/e2e/sanity/Constants.java
index a17f229..2f868e9 100644
--- a/statefun-e2e-tests/statefun-sanity-e2e/src/main/java/org/apache/flink/statefun/e2e/sanity/Constants.java
+++ b/statefun-e2e-tests/statefun-sanity-e2e/src/main/java/org/apache/flink/statefun/e2e/sanity/Constants.java
@@ -31,14 +31,13 @@ final class Constants {
static final String KAFKA_BOOTSTRAP_SERVERS_CONF = "kafka-bootstrap-servers";
static final IngressIdentifier<Command> COMMAND_INGRESS_ID =
- new IngressIdentifier<>(Command.class, "org.apache.flink.itcases.sanity", "commands");
+ new IngressIdentifier<>(Command.class, "org.apache.flink.e2e.sanity", "commands");
static final EgressIdentifier<StateSnapshot> STATE_SNAPSHOT_EGRESS_ID =
- new EgressIdentifier<>(
- "org.apache.flink.itcases.sanity", "state-snapshots", StateSnapshot.class);
+ new EgressIdentifier<>("org.apache.flink.e2e.sanity", "state-snapshots", StateSnapshot.class);
static final FunctionType[] FUNCTION_TYPES =
new FunctionType[] {
- new FunctionType("org.apache.flink.itcases.sanity", "t0"),
- new FunctionType("org.apache.flink.itcases.sanity", "t1")
+ new FunctionType("org.apache.flink.e2e.sanity", "t0"),
+ new FunctionType("org.apache.flink.e2e.sanity", "t1")
};
}
[flink-statefun] 03/04: [FLINK-16395] [legal] Add OkHttp to NOTICE
of statefun-flink-distribution
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 fb88505213085e576f11794c1d9f34ec3a4fcb10
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Mar 3 11:31:22 2020 +0800
[FLINK-16395] [legal] Add OkHttp to NOTICE of statefun-flink-distribution
This closes #44.
---
.../statefun-flink-distribution/src/main/resources/META-INF/NOTICE | 1 +
1 file changed, 1 insertion(+)
diff --git a/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE b/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE
index d795bb7..6fad4ea 100644
--- a/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE
+++ b/statefun-flink/statefun-flink-distribution/src/main/resources/META-INF/NOTICE
@@ -15,6 +15,7 @@ This project bundles the following dependencies under the Apache Software Licens
- org.apache.kafka:kafka-clients:2.2.0
- org.lz4:lz4-java:1.5.0
- org.objenesis:objenesis:2.1
+- com.squareup.okhttp3:okhttp:3.14.6
This project bundles the following dependencies under the BSD license.
See bundled license files under "META-INF/licenses" for details.
[flink-statefun] 01/04: [FLINK-16123] [e2e] Add RoutableKafkaE2E
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 1fe23c0ffd915cfe7090502505dee2e4379365b1
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Wed Feb 26 18:19:06 2020 +0800
[FLINK-16123] [e2e] Add RoutableKafkaE2E
End-to-end test based on the {@link RoutableKafkaVerificationModule} application.
This test writes some records to Kafka, with target function id as key
(UTF8 String) and MessageWithAddress messages as value, without the
"from" field set. The routable Kafka ingress should automatically route
them to the correct function instances, which tag the input messages
with their own address, and then forwards it back to Kafka. The test
verifies that the tagged outputs written back to Kafka are correct.
This closes #37.
---
.../e2e/routablekafka/RoutableKafkaE2E.java | 142 +++++++++++++++++++++
.../src/test/resources/Dockerfile | 21 +++
.../src/test/resources/log4j.properties | 24 ++++
.../routable-kafka-ingress-module/module.yaml | 39 ++++++
4 files changed, 226 insertions(+)
diff --git a/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/java/org/apache/flink/statefun/e2e/routablekafka/RoutableKafkaE2E.java b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/java/org/apache/flink/statefun/e2e/routablekafka/RoutableKafkaE2E.java
new file mode 100644
index 0000000..9f4eda1
--- /dev/null
+++ b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/java/org/apache/flink/statefun/e2e/routablekafka/RoutableKafkaE2E.java
@@ -0,0 +1,142 @@
+/*
+ * 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.e2e.routablekafka;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.util.Collections;
+import java.util.Properties;
+import org.apache.flink.statefun.e2e.common.StatefulFunctionsAppContainers;
+import org.apache.flink.statefun.e2e.common.kafka.KafkaIOVerifier;
+import org.apache.flink.statefun.e2e.common.kafka.KafkaProtobufSerializer;
+import org.apache.flink.statefun.e2e.routablekafka.generated.RoutableKafkaVerification.FnAddress;
+import org.apache.flink.statefun.e2e.routablekafka.generated.RoutableKafkaVerification.MessageWithAddress;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Rule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+
+/**
+ * End-to-end test based on the {@link RoutableKafkaVerificationModule} application.
+ *
+ * <p>This test writes some records to Kafka, with target function id as key (UTF8 String) and
+ * {@link MessageWithAddress} messages as value, without the {@code from} field set. The routable
+ * Kafka ingress should automatically route them to the correct function instances, which tag the
+ * input messages with their own address, and then forwards it back to Kafka (see {@link
+ * FnSelfAddressTagger} function). The test verifies that the tagged outputs written back to Kafka
+ * are correct.
+ */
+public class RoutableKafkaE2E {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RoutableKafkaE2E.class);
+
+ private static final String CONFLUENT_PLATFORM_VERSION = "5.0.3";
+ private static final String KAFKA_HOST = "kafka-broker";
+
+ @Rule
+ public KafkaContainer kafka =
+ new KafkaContainer(CONFLUENT_PLATFORM_VERSION).withNetworkAliases(KAFKA_HOST);
+
+ @Rule
+ public StatefulFunctionsAppContainers verificationApp =
+ new StatefulFunctionsAppContainers("routable-kafka-verification", 1)
+ .dependsOn(kafka)
+ .exposeMasterLogs(LOG)
+ .withBuildContextFileFromClasspath(
+ "routable-kafka-ingress-module", "/routable-kafka-ingress-module/")
+ .withModuleGlobalConfiguration(
+ Constants.KAFKA_BOOTSTRAP_SERVERS_CONF, KAFKA_HOST + ":9092");
+
+ @Test(timeout = 60_000L)
+ public void run() {
+ final String kafkaAddress = kafka.getBootstrapServers();
+
+ final Producer<String, MessageWithAddress> messageProducer =
+ kafkaKeyedMessagesProducer(kafkaAddress);
+ final Consumer<String, MessageWithAddress> taggedMessageConsumer =
+ kafkaTaggedMessagesConsumer(kafkaAddress);
+
+ final KafkaIOVerifier<String, MessageWithAddress, String, MessageWithAddress> verifier =
+ new KafkaIOVerifier<>(messageProducer, taggedMessageConsumer);
+
+ assertThat(
+ verifier.sending(
+ producerRecord("messages-1", "key-1", message("foo")),
+ producerRecord("messages-1", "key-2", message("bar")),
+ producerRecord("messages-2", "key-1", message("hello"))),
+ verifier.resultsInOrder(
+ is(taggedMessage(fnAddress(Constants.FUNCTION_NAMESPACE, "t0", "key-1"), "foo")),
+ is(taggedMessage(fnAddress(Constants.FUNCTION_NAMESPACE, "t1", "key-1"), "foo")),
+ is(taggedMessage(fnAddress(Constants.FUNCTION_NAMESPACE, "t0", "key-2"), "bar")),
+ is(taggedMessage(fnAddress(Constants.FUNCTION_NAMESPACE, "t1", "key-2"), "bar")),
+ is(taggedMessage(fnAddress(Constants.FUNCTION_NAMESPACE, "t1", "key-1"), "hello"))));
+ }
+
+ private static Producer<String, MessageWithAddress> kafkaKeyedMessagesProducer(
+ String bootstrapServers) {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", bootstrapServers);
+
+ return new KafkaProducer<>(
+ props, new StringSerializer(), new KafkaProtobufSerializer<>(MessageWithAddress.parser()));
+ }
+
+ private Consumer<String, MessageWithAddress> kafkaTaggedMessagesConsumer(
+ String bootstrapServers) {
+ Properties consumerProps = new Properties();
+ consumerProps.setProperty("bootstrap.servers", bootstrapServers);
+ consumerProps.setProperty("group.id", "routable-kafka");
+ consumerProps.setProperty("auto.offset.reset", "earliest");
+
+ KafkaConsumer<String, MessageWithAddress> consumer =
+ new KafkaConsumer<>(
+ consumerProps,
+ new StringDeserializer(),
+ new KafkaProtobufSerializer<>(MessageWithAddress.parser()));
+ consumer.subscribe(Collections.singletonList(KafkaIO.TAGGED_MESSAGES_TOPIC_NAME));
+
+ return consumer;
+ }
+
+ private static ProducerRecord<String, MessageWithAddress> producerRecord(
+ String topic, String key, MessageWithAddress message) {
+ return new ProducerRecord<>(topic, key, message);
+ }
+
+ private static MessageWithAddress message(String message) {
+ return MessageWithAddress.newBuilder().setMessage(message).build();
+ }
+
+ private static MessageWithAddress taggedMessage(FnAddress fromTag, String message) {
+ return MessageWithAddress.newBuilder().setFrom(fromTag).setMessage(message).build();
+ }
+
+ private static FnAddress fnAddress(String namespace, String type, String id) {
+ return FnAddress.newBuilder().setNamespace(namespace).setType(type).setId(id).build();
+ }
+}
diff --git a/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/Dockerfile b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/Dockerfile
new file mode 100644
index 0000000..2b025c7
--- /dev/null
+++ b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/Dockerfile
@@ -0,0 +1,21 @@
+# 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.
+
+FROM statefun
+
+RUN mkdir -p /opt/statefun/modules/statefun-routable-kafka-e2e
+COPY statefun-routable-kafka-e2e*.jar /opt/statefun/modules/statefun-routable-kafka-e2e/
+COPY routable-kafka-ingress-module/ /opt/statefun/modules/statefun-routable-kafka-e2e/
+COPY flink-conf.yaml $FLINK_HOME/conf/flink-conf.yaml
diff --git a/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/log4j.properties b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/log4j.properties
new file mode 100644
index 0000000..fb965d3
--- /dev/null
+++ b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/log4j.properties
@@ -0,0 +1,24 @@
+################################################################################
+# 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.
+################################################################################
+
+log4j.rootLogger=INFO, console
+
+# Log all infos in the given file
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
diff --git a/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/routable-kafka-ingress-module/module.yaml b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/routable-kafka-ingress-module/module.yaml
new file mode 100644
index 0000000..6e0c7f5
--- /dev/null
+++ b/statefun-e2e-tests/statefun-routable-kafka-e2e/src/test/resources/routable-kafka-ingress-module/module.yaml
@@ -0,0 +1,39 @@
+# 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.
+
+module:
+ meta:
+ type: remote
+ spec:
+ ingresses:
+ - ingress:
+ meta:
+ type: org.apache.flink.statefun.sdk.kafka/routable-protobuf-kafka-connector
+ id: org.apache.flink.statefun.e2e/messages
+ spec:
+ address: kafka-broker:9092
+ consumerGroupId: routable-kafka-e2e
+ startupPosition:
+ type: earliest
+ topics:
+ - topic: messages-1
+ typeUrl: com.googleapis/org.apache.flink.statefun.e2e.routablekafka.MessageWithAddress
+ targets:
+ - org.apache.flink.e2e.routablekafka/t0
+ - org.apache.flink.e2e.routablekafka/t1
+ - topic: messages-2
+ typeUrl: com.googleapis/org.apache.flink.statefun.e2e.routablekafka.MessageWithAddress
+ targets:
+ - org.apache.flink.e2e.routablekafka/t1