Add examples and update constraints for Kafka API (#13492)

Co-authored-by: anton-bobkov <anton-bobkov@ydb.tech>
Co-authored-by: Ivan Blinkov <ivan@ydb.tech>

Author: 3 people

ydb/docs/_includes/bash/kafka-api-console-read-no-auth.md

@@ -0,0 +1,9 @@
+
+```bash
+kafka-console-consumer --bootstrap-server localhost:9092 \
+    --topic my-topic  \
+    --group my-group \
+    --from-beginning \
+    --consumer-property check.crcs=false \
+    --consumer-property partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor
+```

ydb/docs/_includes/bash/kafka-api-console-read-with-sasl-creds-on-prem.md

@@ -0,0 +1,12 @@
+
+```bash
+kafka-console-consumer --bootstrap-server localhost:9092 \
+--topic <topic-name>  \
+--group <consumer-name> \
+--from-beginning \
+--consumer-property check.crcs=false \
+--consumer-property partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor \
+--consumer-property security.protocol=SASL_PLAINTEXT \
+--consumer-property sasl.mechanism=PLAIN \
+--consumer-property "sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"<username>\" password=\"<password>\";"
+```

ydb/docs/_includes/bash/kafka-api-console-write-no-auth.md

@@ -0,0 +1,4 @@
+
+```bash
+kafka-console-producer --broker-list localhost:9092 --topic my-topic
+```

ydb/docs/_includes/bash/kafka-api-kcat-read-no-auth.md

@@ -0,0 +1,8 @@
+
+```bash
+kcat -C \
+  -b <ydb-endpoint> \
+  -X check.crcs=false \
+  -X partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor \
+  -G <consumer-name> <topic-name>
+```

ydb/docs/_includes/bash/kafka-api-kcat-read-with-sasl-creds-on-prem.md

@@ -0,0 +1,12 @@
+
+```bash
+kcat -C \
+  -b localhost:9092 \
+  -X security.protocol=SASL_PLAINTEXT \
+  -X sasl.mechanism=PLAIN \
+  -X sasl.username="<username>" \
+  -X sasl.password="<password>" \
+  -X check.crcs=false \
+  -X partition.assignment.strategy=roundrobin \
+  -G <consumer-name> <topic-name>
+```

ydb/docs/_includes/bash/kafka-api-kcat-write-no-auth.md

@@ -0,0 +1,7 @@
+
+```bash
+echo "test message" | kcat -P \
+    -b <ydb-endpoint> \
+    -t <topic-name> \
+    -k key
+```

ydb/docs/_includes/java/kafka-api-flink-read-no-auth.md

@@ -0,0 +1,29 @@
+
+```java
+public class YdbKafkaApiReadExample {
+
+    public static void main(String[] args) throws Exception {
+        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment()
+                .enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
+
+        Configuration config = new Configuration();
+        config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
+        config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "file:///path/to/your/checkpoints");
+        env.configure(config);
+
+        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
+                .setBootstrapServers("localhost:9092")
+                .setProperty(ConsumerConfig.CHECK_CRCS_CONFIG, "false")
+                .setGroupId("flink-demo-consumer")
+                .setTopics("my-topic")
+                .setStartingOffsets(OffsetsInitializer.earliest())
+                .setBounded(OffsetsInitializer.latest())
+                .setValueOnlyDeserializer(new SimpleStringSchema())
+                        .build();
+
+        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source").print();
+
+        env.execute("YDB Kafka API example read app");
+    }
+}
+```

ydb/docs/_includes/java/kafka-api-flink-write-no-auth.md

@@ -0,0 +1,29 @@
+
+```java
+public class YdbKafkaApiProduceExample {
+  private static final String TOPIC = "my-topic";
+
+  public static void main(String[] args) throws Exception {
+      final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+      Sink<String> kafkaSink = KafkaSink.<String>builder()
+              .setBootstrapServers("localhost:9092") // assuming ydb is running locally with kafka proxy on 9092 port
+              .setRecordSerializer(KafkaRecordSerializationSchema.builder()
+                      .setTopic(TOPIC)
+                      .setValueSerializationSchema(new SimpleStringSchema())
+                      .setKeySerializationSchema(new SimpleStringSchema())
+                      .build())
+              .setRecordSerializer((el, ctx, ts) -> new ProducerRecord<>(TOPIC, el.getBytes()))
+              .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
+                      .build();
+
+      env.setParallelism(1)
+              .fromSequence(0, 10)
+              .map(i -> i + "")
+              .sinkTo(kafkaSink);
+
+      // Execute program, beginning computation.
+      env.execute("ydb_kafka_api_write_example");
+  }
+}
+```

ydb/docs/_includes/java/kafka-api-java-read-no-auth.md

@@ -0,0 +1,27 @@
+
+```java
+String HOST = "<ydb-endpoint>";
+String TOPIC = "<topic-name>";
+String CONSUMER = "<consumer-name>";
+
+Properties props = new Properties();
+
+props.put("bootstrap.servers", HOST);
+
+props.put("key.deserializer", StringDeserializer.class.getName());
+props.put("value.deserializer", StringDeserializer.class.getName());
+
+props.put("check.crcs", false);
+props.put("partition.assignment.strategy", RoundRobinAssignor.class.getName());
+
+props.put("group.id", CONSUMER);
+Consumer<String, String> consumer = new KafkaConsumer<>(props);
+consumer.subscribe(Arrays.asList(new String[] {TOPIC}));
+
+while (true) {
+  ConsumerRecords<String, String> records = consumer.poll(10000); // timeout 10 sec
+  for (ConsumerRecord<String, String> record : records) {
+      System.out.println(record.key() + ":" + record.value());
+  }
+}
+```

ydb/docs/_includes/java/kafka-api-java-read-with-sasl-creds-on-prem.md

@@ -0,0 +1,30 @@
+
+```java
+String TOPIC = "<topic-name>";
+String CONSUMER = "<consumer-name>";
+
+Properties props = new Properties();
+
+props.put("bootstrap.servers", "localhost:9092");
+
+props.put("key.deserializer", StringDeserializer.class.getName());
+props.put("value.deserializer", StringDeserializer.class.getName());
+
+props.put("check.crcs", false);
+props.put("partition.assignment.strategy", RoundRobinAssignor.class.getName());
+
+props.put("security.protocol", "SASL_PLAINTEXT");
+props.put("sasl.mechanism", "PLAIN");
+props.put("sasl.jaas.config", "sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"<username>\" password=\"<password>\";");
+
+props.put("group.id", CONSUMER);
+Consumer<String, String> consumer = new KafkaConsumer<>(props);
+consumer.subscribe(Arrays.asList(new String[] {TOPIC}));
+
+while (true) {
+  ConsumerRecords<String, String> records = consumer.poll(10000); // timeout 10 sec
+  for (ConsumerRecord<String, String> record : records) {
+      System.out.println(record.key() + ":" + record.value());
+  }
+}
+```