Configuring Apache Flink SQL Kafka Connector with SCRAM-SHA-256 and Protobuf

This document provides a step-by-step guide to configuring the Apache Flink SQL Kafka connector with username/password authentication (SCRAM-SHA-256) and Protobuf deserialization in a Podman-based setup.

1. Install Required Dependencies

Ensure the following dependencies are available in Flink’s /lib directory:

Flink SQL Kafka Connector JAR Example: flink-sql-connector-kafka-<flink-version>.jar.
Protobuf JARs
- protobuf-java-<version>.jar (Core Protobuf library).
- Your Protobuf-generated classes JAR (if applicable).
- If using Confluent's Protobuf format, include kafka-protobuf-serializer and kafka-clients JARs.
Flink SQL Protobuf JAR Use flink-sql-protobuf-<flink-version>.jar if available, or provide a custom Protobuf deserialization JAR.

2. Create a Kafka Table in SQL

Define the Kafka source table with SCRAM-SHA-256 authentication and Protobuf serialization.

SQL for Kafka Source Table

CREATE TABLE KafkaSource (
    key STRING,
    value ROW(
        field1 STRING,
        field2 INT,
        field3 TIMESTAMP(3)
    ),
    sourcetime AS PROCTIME() -- Optional: Extract processing time
) WITH (
    'connector' = 'kafka',
    'topic' = 'your-topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'flink-group',

    -- Authentication properties
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'SCRAM-SHA-256',
    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username="your-username" password="your-password";',

    -- Serialization
    'format' = 'protobuf',
    'protobuf.message-class-name' = 'your.protobuf.MessageClassName'
);

3. Use the Kafka Table

Once the Kafka source table is created, you can perform SQL queries on it.

Basic Query Example

SELECT * FROM KafkaSource;

Streaming with Aggregation

For example, if field2 is a numeric value, calculate real-time statistics:

SELECT
    TUMBLE_START(sourcetime, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(sourcetime, INTERVAL '1' MINUTE) AS window_end,
    COUNT(*) AS message_count,
    AVG(value.field2) AS average_value
FROM KafkaSource
GROUP BY TUMBLE(sourcetime, INTERVAL '1' MINUTE);

4. Custom Protobuf Configuration (Optional)

If you’re not using Confluent’s Protobuf deserializer, register a custom Protobuf deserialization format.

Custom Format JAR

Write a custom Protobuf deserialization format, package it into a JAR, and add it to Flink’s /lib directory.

5. Run the SQL Job

Use the Flink SQL CLI to execute the SQL queries.

Start Flink SQL CLI

./bin/sql-client.sh

Load and Run the Query

Register the Table
Paste the CREATE TABLE statement into the SQL CLI.
Run the Query
Execute any SQL query on the table:
```
SELECT * FROM KafkaSource;
```

6. Debug and Test

Verify your Flink logs to ensure successful connectivity and message consumption.
Test the Protobuf deserialization by inspecting the schema and ensuring no errors occur when parsing messages.

This configuration ensures your Flink SQL jobs can securely connect to Kafka using SCRAM-SHA-256 and deserialize Protobuf messages efficiently. Let me know if you need further assistance!

If you're using your own Protobuf file in Flink for deserialization, you’ll need to ensure the appropriate deserialization mechanism is configured. Here’s how to handle it effectively depending on your setup:

1. Protobuf Deserialization in Flink

Flink supports Protobuf deserialization, but the setup depends on how you’re using it:

SQL Configuration with Built-in Protobuf Support: If you’re using Flink’s SQL connector with Protobuf, you can define the Protobuf message class in the CREATE TABLE statement using the protobuf.message-class-name option.
Example:
```
sql
CREATE TABLE KafkaSource (
    key STRING,
    value ROW(
        field1 STRING,
        field2 INT
    )
) WITH (
    'connector' = 'kafka',
    'topic' = 'your-topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'protobuf',
    'protobuf.message-class-name' = 'your.protobuf.MessageClassName'
);
```
In this case:
- Flink uses the Protobuf deserializer for your specified Protobuf class (your.protobuf.MessageClassName).
- No additional custom serialization is required unless you want to customize or override the default behavior.
Custom Deserialization (Optional or Advanced Use Cases): If your Protobuf schema has unique requirements or you need to handle deserialization manually, you’ll need to create a custom DeserializationSchema.

2. Setting Up Your Protobuf Class

Use the protoc compiler to generate Java classes from your .proto file.
Include the generated Java classes in a JAR and place it in the Flink /lib directory.
```
bash
protoc --java_out=output/path your_file.proto
```

3. Custom Serialization/Deserialization (If Required)

If the built-in Protobuf deserializer doesn't meet your needs, you can write a custom DeserializationSchema. Here’s an example:

Custom Protobuf Deserialization Schema

java
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import com.google.protobuf.InvalidProtocolBufferException;
import your.protobuf.MessageClassName;

public class ProtobufDeserializationSchema implements DeserializationSchema<MessageClassName> {

    @Override
    public MessageClassName deserialize(byte[] message) throws InvalidProtocolBufferException {
        return MessageClassName.parseFrom(message);
    }

    @Override
    public boolean isEndOfStream(MessageClassName nextElement) {
        return false;
    }

    @Override
    public TypeInformation<MessageClassName> getProducedType() {
        return TypeInformation.of(MessageClassName.class);
    }
}

Add this custom deserialization schema when creating the Kafka source.

4. Do You Need Special Serialization?

No Special Serialization Required in Most Cases: If you’re consuming Protobuf messages with Flink SQL and your generated Protobuf classes are available, Flink’s built-in Protobuf deserializer will handle the serialization seamlessly.
Custom Serialization Needed for Complex Logic: If you need custom logic (e.g., modifying fields or handling schema evolution), you’ll need to implement a custom serialization/deserialization schema.

5. Testing Your Setup

Validate the Protobuf deserialization by running a small test job with your Kafka topic.
Test schema compatibility and make sure all fields from the Protobuf messages are correctly mapped to Flink SQL table fields.

If your Protobuf schema has any specific complexities (like nested messages or custom types), let me know, and I can tailor the guidance further!

Tips to Improve Knowledge

Monday, 5 August 2024

Apache Flink SQL Kafka Connector