Merge pull request #270909 from sreekzz/Flink-batch4

Flink updates Batch 4

Author: PMEds28

articles/hdinsight-aks/flink/azure-service-bus-demo.md

@@ -1,9 +1,9 @@
 ---
 title: Use Apache Flink on HDInsight on AKS with Azure Service Bus 
-description: Use Apache Flink DataStream API on HDInsight on AKS with Azure Service Bus
+description: Use Apache Flink DataStream API on HDInsight on AKS with Azure Service Bus.
 ms.service: hdinsight-aks
 ms.topic: how-to
-ms.date: 11/27/2023
+ms.date: 04/02/2024
 ---
 # Use Apache Flink on HDInsight on AKS with Azure Service Bus
 
@@ -13,7 +13,7 @@ This article provides an overview and demonstration of Apache Flink DataStream A
 
 ## Prerequisites
 
--  [Flink Cluster 1.16.0 on HDInsight on AKS](./flink-create-cluster-portal.md)
+-  [Flink Cluster 1.17.0 on HDInsight on AKS](./flink-create-cluster-portal.md)
 -  For this demonstration, we use a Window VM as maven project develop env in the same VNET as HDInsight on AKS.
 -  During the [creation](./flink-create-cluster-portal.md) of the Flink cluster, you are required to ensure that SSH access is selected. This enables you to access the cluster using Secure Shell (SSH).
 -  Set up an [Azure Service Bus](/azure/service-bus-messaging/service-bus-messaging-overview) instance.
@@ -76,7 +76,7 @@ In the POM.xml file, we define the project's dependencies using Maven, ensuring
     <properties>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <flink.version>1.16.0</flink.version>
+        <flink.version>1.17.0</flink.version>
         <java.version>1.8</java.version>
     </properties>
     <dependencies>

articles/hdinsight-aks/flink/change-data-capture-connectors-for-apache-flink.md

@@ -3,7 +3,7 @@ title: How to perform Change Data Capture of SQL Server with Apache Flink® Data
 description: Learn how to perform Change Data Capture of SQL Server with Apache Flink® DataStream API and DataStream Source.
 ms.service: hdinsight-aks
 ms.topic: how-to
-ms.date: 03/22/2024
+ms.date: 04/02/2024
 ---
 
 # Change Data Capture of SQL Server with Apache Flink® DataStream API and DataStream Source on HDInsight on AKS
@@ -138,7 +138,7 @@ GO
 ```
 ##### Maven source code on IdeaJ
 
-In the below snippet, we use HDInsight Kafka 2.4.1. Based on your usage, update the version of Kafka on `<kafka.version>`. 
+In the below snippet, we use Kafka 2.4.1. Based on your usage, update the version of Kafka on `<kafka.version>`. 
 
 **maven pom.xml**
 
@@ -158,7 +158,7 @@ In the below snippet, we use HDInsight Kafka 2.4.1. Based on your usage, update
          <flink.version>1.17.0</flink.version>
          <java.version>1.8</java.version>
          <scala.binary.version>2.12</scala.binary.version>
-         <kafka.version>3.2.0</kafka.version> // Replace with 3.2 if you're using HDInsight Kafka 3.2
+         <kafka.version>3.2.0</kafka.version> //
      </properties>
      <dependencies>
          <dependency>

articles/hdinsight-aks/flink/flink-how-to-setup-event-hub.md

@@ -3,7 +3,7 @@ title: How to connect Apache Flink® on HDInsight on AKS with Azure Event Hubs f
 description: Learn how to connect Apache Flink® on HDInsight on AKS with Azure Event Hubs for Apache Kafka®
 ms.service: hdinsight-aks
 ms.topic: how-to
-ms.date: 08/29/2023
+ms.date: 04/02/2024
 ---
 
 # Connect Apache Flink® on HDInsight on AKS with Azure Event Hubs for Apache Kafka®
@@ -53,60 +53,66 @@ In this article, we explore how to connect [Azure Event Hubs](/azure/event-hubs/
 ## Packaging the JAR for Flink 
 1. Package com.example.app;
    
-   ```
+    ```
+       package contoso.example;
+     
     import org.apache.flink.api.common.functions.MapFunction;
     import org.apache.flink.api.common.serialization.SimpleStringSchema;
+     
+    import org.apache.flink.api.java.utils.ParameterTool;
+    import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
+    import org.apache.flink.connector.kafka.sink.KafkaSink;
+     
     import org.apache.flink.streaming.api.datastream.DataStream;
     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;     //v0.11.0.0
-    import java.io.FileNotFoundException;
+     
     import java.io.FileReader;
     import java.util.Properties;
-
-    public class FlinkTestProducer {
-
-    private static final String TOPIC = "test";
-    private static final String FILE_PATH = "src/main/resources/producer.config";
-
-    public static void main(String... args) {
-        try {
-            Properties properties = new Properties();
-            properties.load(new FileReader(FILE_PATH));
-
-            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-            DataStream stream = createStream(env);
-            FlinkKafkaProducer<String> myProducer = new FlinkKafkaProducer<>(
-                    TOPIC,    
-                    new SimpleStringSchema(),   // serialization schema
-                    properties);
-
-            stream.addSink(myProducer);
-            env.execute("Testing flink print");
-
-        } catch(FileNotFoundException e){
-            System.out.println("FileNotFoundException: " + e);
-        } catch (Exception e) {
-            System.out.println("Failed with exception:: " + e);
-        }
-    }
-
-    public static DataStream createStream(StreamExecutionEnvironment env){
-        return env.generateSequence(0, 200)
-            .map(new MapFunction<Long, String>() {
-                @Override
-                public String map(Long in) {
-                    return "FLINK PRODUCE " + in;
-                }
-            });
-      }
-   }
-   ```
- 
+     
+    public class AzureEventHubDemo {
+     
+       public static void main(String[] args) throws Exception {
+           // 1. get stream execution environment
+           StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
+           ParameterTool parameters = ParameterTool.fromArgs(args);
+           String input = parameters.get("input");
+           Properties properties = new Properties();
+           properties.load(new FileReader(input));
+     
+           // 2. generate stream input
+           DataStream<String> stream = createStream(env);
+     
+           // 3. sink to eventhub
+           KafkaSink<String> sink = KafkaSink.<String>builder().setKafkaProducerConfig(properties)
+                  .setRecordSerializer(KafkaRecordSerializationSchema.builder()
+                          .setTopic("topic1")
+                          .setValueSerializationSchema(new SimpleStringSchema())
+                           .build())
+                   .build();
+     
+           stream.sinkTo(sink);
+     
+           // 4. execute the stream
+           env.execute("Produce message to Azure event hub");
+       }
+     
+       public static DataStream<String> createStream(StreamExecutionEnvironment env){
+           return env.generateSequence(0, 200)
+                   .map(new MapFunction<Long, String>() {
+                       @Override
+                       public String map(Long in) {
+                           return "FLINK PRODUCE " + in;
+                       }
+                   });
+       }
+    } 
+    ```
+     
 1. Add the snippet to run the Flink Producer.
 
    :::image type="content" source="./media/flink-eventhub/testing-flink.png" alt-text="Screenshot showing how to test Flink in Event Hubs." border="true" lightbox="./media/flink-eventhub/testing-flink.png":::
 
-1. Once the code is executed, the events are stored in the topic **“TEST”**
+1. Once the code is executed, the events are stored in the topic **"topic1"**
 
    :::image type="content" source="./media/flink-eventhub/events-stored-in-topic.png" alt-text="Screenshot showing Event Hubs stored in topic." border="true" lightbox="./media/flink-eventhub/events-stored-in-topic.png":::