tzulitai commented on a change in pull request #63: [FLINK-16557][docs]
Document YAML-ized Kafka egresses / ingresses in …
URL: https://github.com/apache/flink-statefun/pull/63#discussion_r396063018
##########
File path: statefun-docs/docs/io_module/kafka.rst
##########
@@ -36,44 +37,178 @@ To use the Kafka I/O Module, please include the following
dependency in your pom
<scope>provided</scope>
</dependency>
-Kafka Ingress Builder
-=====================
+Kafka Ingress Spec
+==================
-A ``KafkaIngressBuilder`` declares an ingress spec for consuming from Kafka
cluster.
+A ``KafkaIngressSpec`` declares an ingress spec for consuming from Kafka
cluster.
It accepts the following arguments:
1) The ingress identifier associated with this ingress
2) The topic name / list of topic names
3) The address of the bootstrap servers
4) The consumer group id to use
-5) A ``KafkaIngressDeserializer`` for deserializing data from Kafka
+5) A ``KafkaIngressDeserializer`` for deserializing data from Kafka (Java only)
6) The position to start consuming from
-.. literalinclude::
../../src/main/java/org/apache/flink/statefun/docs/io/kafka/IngressSpecs.java
- :language: java
- :lines: 18-
+.. tabs::
-The ingress allows configuring the startup position to be one of the following:
+ .. group-tab:: Java
-* ``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.
+ .. literalinclude::
../../src/main/java/org/apache/flink/statefun/docs/io/kafka/IngressSpecs.java
+ :language: java
+ :lines: 18-
-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
-using
``KafkaIngressBuilder#withAutoOffsetResetPosition(KafkaIngressAutoResetPosition)``.
By default, this is set to be the latest position.
+ .. group-tab:: Yaml
+
+ .. code-block:: yaml
+
+ version: "1.0"
+
+ module:
+ meta:
+ type: remote
+ spec:
+ ingresses:
+ - ingress:
+ meta:
+ type: statefun.kafka.io/routable-protobuf-ingress
+ id: example/user-ingress
+ spec:
+ address: kafka-broker:9092
+ consumerGroupId: routable-kafka-e2e
+ startupPosition:
+ type: earliest
+ topics:
+ - topic: messages-1
Review comment:
Indentation:
```suggestion
topics:
- topic: messages-1
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
