Realtime Compute for Apache Flink:Process Protobuf data with Kafka connector

Step 1: Compile the Protobuf file

1. Create a Protobuf file named order.proto.

   proto3

   syntax = "proto3";
   // Logical package name for importing this file in other .proto files.
   package com.aliyun;
   // Java package name. If not specified, the proto package is used by default.
   option java_package = "com.aliyun";
   // Specifies whether to compile into multiple files. It is recommended to set this to true
   // so that each message generates a separate .java file instead of an inner class.
   option java_multiple_files = true;
   // Java outer class name. When java_multiple_files is true, this class contains metadata
   // such as the file name and does not wrap the message classes.
   option java_outer_classname = "OrderProtoBuf";
   message Order {
     // Proto3 removes the optional/required keywords.
     // Note: Primitive data types (int, long, double) default to 0, and string defaults to an empty string.
     // In Java, a hasOrderId() method will no longer be generated, making it impossible to distinguish
     // between "not set" and "set to 0".
     int32 orderId = 1;
     string orderName = 2;
     double orderPrice = 3;
     int64 orderDate = 4;
   }

   proto2

   syntax = "proto2";
   // The package name of the proto.
   package com.aliyun;
   // The Java package name. If not specified, the proto package is used by default.
   option java_package = "com.aliyun";
   // Specifies whether to compile into multiple files.
   option java_multiple_files = true;
   // The Java wrapper class name.
   option java_outer_classname = "OrderProtoBuf";
   message Order {
     optional int32 orderId = 1;
     optional string orderName= 2;
     optional double orderPrice = 3;
     optional int64 orderDate = 4;
   }

2. Use the Protocol Buffers tool to generate the source code.

   Create an empty Maven project and place the Protobuf file in the src/main/proto directory.

   Directory example

   KafkaProtobuf
   ‒ src
     -main
       -java
       -proto
         -order.proto
   ‒ pom.xml

   pom.xml

   Note

   Align the version with your Flink dependency (for example, 3.21.7) to prevent the generated classes from conflicting with protobuf-java:3.21.7.

   <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
       <modelVersion>4.0.0</modelVersion>
           <groupId>com.aliyun</groupId>
           <artifactId>KafkaProtobuf</artifactId>
           <version>1.0-SNAPSHOT</version>
       <dependencies>
           <dependency>
               <groupId>com.google.protobuf</groupId>
               <artifactId>protobuf-java</artifactId>
               <version>3.21.7</version> <!-- This version must match the Protobuf version used for code generation. -->
           </dependency>
           <dependency>
               <groupId>org.apache.kafka</groupId>
               <artifactId>kafka-clients</artifactId>
               <version>3.3.1</version> <!-- Adjust this based on your Kafka version. -->
           </dependency>
           <dependency>
               <groupId>com.github.javafaker</groupId>
               <artifactId>javafaker</artifactId>
               <version>1.0.2</version>
           </dependency>
       </dependencies>
       <build>
           <finalName>KafkaProtobuf</finalName>
           <plugins>
               <!-- Java compiler -->
               <plugin>
                   <groupId>org.apache.maven.plugins</groupId>
                   <artifactId>maven-compiler-plugin</artifactId>
                   <version>3.13.0</version>
                   <configuration>
                       <source>1.8</source>
                       <target>1.8</target>
                   </configuration>
               </plugin>
               <!-- Use the maven-shade-plugin to create a fat JAR with all required dependencies. -->
               <plugin>
                   <groupId>org.apache.maven.plugins</groupId>
                   <artifactId>maven-shade-plugin</artifactId>
                   <version>3.5.3</version>
               </plugin>
           </plugins>
       </build>
   </project>

   Three classes are generated in the java directory: the Order class, the OrderOrBuilder interface, and the OrderProtobuf outer wrapper class.

   # In the terminal, navigate to the project root directory (where pom.xml is located)
   # and run the following command to generate the source code.
   protoc --java_out=src/main/java --proto_path=src/main/proto src/main/proto/order.proto

3. Test the serialization and deserialization.

   package com.aliyun;
   public class OrderTest {
       public static void main(String[] args) {
           // Create an Order object and set its field values.
           Order order = Order.newBuilder()
                   .setOrderId(8513)
                   .setOrderName("flink")
                   .setOrderPrice(99.99)
                   .setOrderDate(System.currentTimeMillis())
                   .build();
           // Serialize to a byte array.
           byte[] serializedBytes = order.toByteArray();
           System.out.println("Byte length after serialization: " + serializedBytes.length);
           // Deserialize the byte array into a new Order object.
           Order deserializedOrder;
           try {
               deserializedOrder = Order.parseFrom(serializedBytes);
           } catch (Exception e) {
               System.err.println("Deserialization failed: " + e.getMessage());
               return;
           }
           System.out.println("Original object: \n" + order);
           // Verify that the fields of the deserialized object match the original object.
           if (order.getOrderId() == deserializedOrder.getOrderId() &&
                   order.getOrderName().equals(deserializedOrder.getOrderName()) &&
               order.getOrderPrice() == deserializedOrder.getOrderPrice() &&
               order.getOrderDate() == deserializedOrder.getOrderDate()) {
               System.out.println("Serialization and deserialization test passed!");
           } else {
               System.out.println("Serialization and deserialization test failed!");
           }
       }
   }

Step 2: Write test data to Kafka

This example uses ApsaraMQ for Kafka as the operating environment.

1. Download the SSL root certificate. This certificate is required if you connect by using an SSL endpoint.

2. Use the username and password for your instance.

   • If an access control list (ACL) is not enabled for the instance, you can obtain the default username and password from the Configuration Information section of the Instance Details page in the ApsaraMQ for Kafka console.

   • If an access control list (ACL) is enabled for the instance, ensure the SASL user uses the PLAIN type and has permission to send and receive messages. For more information, see Use an ACL for access control.

package com.aliyun;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
import com.github.javafaker.Faker; // Import the Faker library to generate a random test dataset.
public class ProtoBufToKafkaTest {
    public static void main(String[] args) {
        Properties props = new Properties();
        // Set the endpoint. Obtain the endpoint for the topic from the Kafka console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "<bootstrap_servers>");
        // Set the access protocol to SASL_SSL.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        // Set the absolute path of the SSL root certificate. Do not package this file into the JAR.
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "../only.4096.client.truststore.jks");
        // The password for the root certificate truststore. Use the default value.
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        // The SASL authentication method. Use the default value.
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        // The serialization method for ApsaraMQ for Kafka message keys and values.
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        // The maximum wait time for a request, in milliseconds.
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        // The number of retries for the client.
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        // The retry interval for the client, in milliseconds.
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
        // Disable hostname verification by setting the value to an empty string.
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"aliyun_flink\" password=\"123456\";");
        // Construct a Producer object. This object is thread-safe.
        // In general, one Producer object per process is sufficient.
        // To improve performance, you can create more objects, but do not exceed five.
        KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);
        String topic = "test";
        // Create a list to store three messages.
        List<ProducerRecord<String, byte[]>> messages = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            byte[] value = getProtoTestData();
            ProducerRecord<String, byte[]> kafkaMessage = new ProducerRecord<>(topic, value);
            messages.add(kafkaMessage);
        }
        try {
            // Send messages in a batch.
            List<Future<RecordMetadata>> futures = new ArrayList<>();
            for (ProducerRecord<String, byte[]> message : messages) {
                Future<RecordMetadata> metadataFuture = producer.send(message);
                futures.add(metadataFuture);
            }
            producer.flush();
            // Synchronously get the result of the Future objects.
            for (Future<RecordMetadata> future : futures) {
                try {
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("Produce ok:" + recordMetadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // If the message still fails to send after retries, the business logic must handle this error.
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }
    private static byte[] getProtoTestData() {
        // Use Faker to generate random data.
        Faker faker = new Faker();
        int orderId = faker.number().numberBetween(1000, 9999); // Generate a random order ID.
        String orderName = faker.commerce().productName(); // Generate a random order name.
        double orderPrice = faker.number().randomDouble(2, 10, 1000); // Generate a random order price.
        long orderDate = System.currentTimeMillis(); // Use the current time as the order date.
        // Create an object based on the defined data structure.
        Order order = Order.newBuilder()
                .setOrderId(orderId)
                .setOrderName(orderName)
                .setOrderPrice(orderPrice)
                .setOrderDate(orderDate)
                .build();
        // Serialize the data: convert the object data to a byte array.
        return order.toByteArray();
    }
}

3. Run the test code to write three Protobuf-formatted messages to the test topic in Kafka.

   Produce ok:test-1@3
   Produce ok:test-1@4
   Produce ok:test-1@5

Step 3: Build and upload artifact

Upload the compiled and packaged KafkaProtobuf.jar file.

In the left-side navigation pane, click Artifact Management. On the Artifacts tab, click Upload Artifact.

Step 4: Read data with Flink SQL

1. For reference, see the following SQL example.

   Set the protobuf.message-class-name parameter to the full class name of your Protobuf message. For more information about protobuf parameters, see Flink-Protobuf.

   CREATE TEMPORARY TABLE KafkaSource (
     orderId INT,
     orderName STRING,
     orderPrice DOUBLE,
     orderDate BIGINT
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'test',
     'properties.group.id' = 'my-group',  -- The ID of the consumer group.
     'properties.bootstrap.servers' = '<bootstrap_servers>', -- Enter the address of the Kafka broker.
     'format' = 'protobuf',              -- The data format for the value part.
     'protobuf.message-class-name' = 'com.aliyun.Order',    -- The message class for the message body.
     'scan.startup.mode' = 'earliest-offset'  -- Read from the earliest offset of the Kafka partition.
   );
   CREATE TEMPORARY TABLE KafkaSink (
     orderId INT,
     orderName STRING,
     orderPrice DOUBLE,
     orderDate BIGINT
   ) WITH (
       'connector' = 'print'
   );
   INSERT INTO KafkaSink
   SELECT * FROM KafkaSource
   ;

2. Reference additional dependencies.

   On the More Configurations panel, upload KafkaProtobuf.jar in the Additional Dependencies section. Use a CREATE TEMPORARY TABLE statement to define the KafkaSource table with the fields: orderId INT, orderName STRING, orderPrice DOUBLE, and orderDate BIGINT. In the WITH clause, set connector to kafka, format to protobuf, and protobuf.message-class-name to com.aliyun.Order. After the job runs successfully, the consumed data appears on the KafkaSink results tab.

3. Debug the SQL code.

   Click Debug and run the following SQL statement to read from the Kafka Protobuf source table.

   CREATE TEMPORARY TABLE KafkaSource (
       orderId INT,
       orderName STRING,
       orderPrice DOUBLE,
       orderDate BIGINT
   ) WITH (
       'connector' = 'kafka',
       'topic' = 'test',
       'properties.group.id' = 'my-group',
       'properties.bootstrap.servers' = 'alikafka-serverless-cn-xxx',
       'format' = 'protobuf',
       'protobuf.message-class-name' = 'com.aliyun.Order',
       'scan.startup.mode' = 'earliest-offset'
   );

4. View the output after the job starts running.

   On the job details page, go to the Job Log tab, then the Running Task Managers sub-tab. Click your Task Manager ID, and then select the Stdout tab. In the Stdout output, you can see data records that start with +I (for example, [6066, Small Marble Car, 150.83, 1745561043465]). This indicates that Flink SQL has successfully read Protobuf-formatted data from Kafka.