Confluent Cloud Flink UDF: Developer's Journey from Data Inspection to Deployment

The Vision: AI-Powered Simplicity

My goal was simple: Use AI to streamline the entire development process. I wanted to show my data structure, describe what output I needed, and let AI help me get there. No wrestling with Flink documentation, no trial-and-error with Maven configurations, no memorizing Terraform syntax.

Just: "Here's my input, here's what I want, help me build it."

This repository demonstrates that complete developer workflow - creating, testing, and deploying a custom Apache Flink User-Defined Function (UDF) on Confluent Cloud using

• Confluent for VS Code for Confluent Cloud
• Claude Code with the Confluent MCP Server
• Confluent Terraform Provider

The Use Case

As a developer working with sensor data, I discovered that my raw data stream contained temperature readings in Fahrenheit, along with other sensor metrics. I wanted to:

1. Extract only the temperature values
2. Convert them from Fahrenheit to Celsius
3. Process this transformation efficiently in real-time using Flink SQL

Initial Situation

Using the Confluent VS Code extension, I inspected my data and found this structure:

The data looked like this:

{
  "sensorId": "sensor_1",
  "value": [
    {"type": "temperature", "value": 68.0, "unit": "Fahrenheit"},
    {"type": "humidity", "value": 65.5, "unit": "Percent"}
  ],
  "timestamp": "2026-05-04T19:11:36.633Z"
}

Development Process

1. Building the UDF with Claude AI and Confluent MCP Server

I used Claude Code integrated with the Confluent MCP server to:

• Develop the Java UDF to extract temperature and convert to Celsius
• Test the function with unit tests covering edge cases
• Iterate on the implementation with type hints for Flink compatibility

The UDF (ExtractTemperatureCelsius) was designed to:

• Parse the nested array structure
• Find the temperature reading (where type = "temperature")
• Apply the conversion formula: Celsius = (Fahrenheit - 32) * 5/9
• Handle null values gracefully

2. Creating Infrastructure with Terraform

I continued using Claude to generate Terraform configurations for:

• Creating a Flink compute pool
• Uploading the UDF JAR artifact

3. Deployment and Validation

After deploying via Terraform, I used the Confluent VS Code extension again to:

• Register the UDF function in Flink SQL
• Query my data using the new function
• Verify the output matched my expectations

Final Result

Now I can query my data with:

SELECT 
    sensorId,
    ExtractTemperatureCelsius(value) AS temp_celsius,
    timestamp
FROM pneff_raw_table;

And get clean, Celsius-formatted temperature data!

Running This Project Yourself

Prerequisites

• Java 11 or higher
• Maven 3.6+
• Terraform (for deployment)
• Confluent Cloud account with Flink enabled
• VS Code with Confluent extension (optional, for data inspection)

Step 1: Clone and Build

# Clone the repository
git clone <your-repo-url>
cd flink-udf

# Run tests
mvn clean test

# Build the JAR
mvn clean package

The compiled JAR will be at: target/temperature-udf-1.0-SNAPSHOT.jar

Step 2: Configure Your Environment

Create a .env file with your Confluent Cloud credentials so that the Confluent MCP Server can work properly with your CC cluster:

# Kafka Cluster
KAFKA_CLUSTER_ID=your_cluster_id

BOOTSTRAP_SERVERS=bootstrap_server
KAFKA_API_KEY=cluster_api_key
KAFKA_API_SECRET=cluster_api_secret
KAFKA_REST_ENDPOINT=cluster_endpoint
KAFKA_CLUSTER_ID=your_cluster_id
KAFKA_ENV_ID=your_env_id

# Schema Registry Configuration
SCHEMA_REGISTRY_API_KEY=sr_api_key
SCHEMA_REGISTRY_API_SECRET=sr_api_secret
SCHEMA_REGISTRY_ENDPOINT=sr_endpoint

Step 3: Deploy with Terraform

# Create terraform.tfvars from your .env values
cat > terraform.tfvars <<EOF
confluent_cloud_api_key    = "your_cloud_api_key"
confluent_cloud_api_secret = "your_cloud_api_secret"
flink_api_key             = "your_flink_api_key"
flink_api_secret          = "your_flink_api_secret"
organization_id           = "your_org_id"
environment_id            = "your_environment_id"
principal_id              = "your_service_account_id"
udf_jar_path              = "target/temperature-udf-1.0-SNAPSHOT.jar"
EOF

# Initialize Terraform
terraform init

# Preview changes
terraform plan

# Deploy
terraform apply

Step 4: Register the Function in Flink SQL

After Terraform completes, note the udf_artifact_id from the output. Then run this SQL statement in Flink:

CREATE FUNCTION ExtractTemperatureCelsius
AS 'com.confluent.flink.udf.ExtractTemperatureCelsius'
USING JAR '<udf_artifact_id_from_terraform_output>';

Step 5: Use the Function

SELECT 
    sensorId,
    ExtractTemperatureCelsius(value) AS temp_celsius,
    timestamp
FROM your_table_name;

Testing

The project includes comprehensive unit tests:

mvn test

Test coverage includes:

• Temperature extraction from arrays with multiple sensor readings
• Temperature in different array positions
• Missing temperature readings
• Null handling (null arrays, empty arrays, null values)
• Real-world data scenarios
• Various temperature conversion edge cases

Clean Up

To remove all deployed resources:

terraform destroy