MACRO NEPAL
Real-time Stream Processing with KSQL and Java
awiskar acharya
November 17, 2025
10:15 am

Introduction

KSQL is the streaming SQL engine for Apache Kafka® that enables real-time data processing using familiar SQL-like syntax. When combined with Java applications, KSQL provides a powerful platform for building real-time stream processing applications, analytics pipelines, and event-driven microservices.

This comprehensive guide explores KSQL integration with Java, covering everything from basic stream processing to advanced real-time analytics.

Architecture Overview

1. KSQL Components

  • KSQL Server: Processes SQL queries on Kafka streams
  • KSQL CLI: Command-line interface for KSQL
  • Streams: Unbounded sequences of structured data
  • Tables: mutable, partitioned collections that represent the latest value for each key
  • Connectors: Integration with external systems

2. Java Integration Patterns

  • KSQL REST API Client: Direct HTTP communication
  • Kafka Streams: Native Java stream processing
  • Spring Cloud Stream: Framework abstraction
  • Custom Clients: Low-level integration

Project Setup and Dependencies

1. Maven Dependencies

<dependencies>
<!-- KSQL DB Client -->
<dependency>
<groupId>io.confluent.ksql</groupId>
<artifactId>ksqldb-api-client</artifactId>
<version>7.5.1</version>
</dependency>
<!-- Kafka Streams -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>3.5.1</version>
</dependency>
<!-- Spring Kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- JSON Processing -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<!-- Testing -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<repositories>
<repository>
<id>confluent</id>
<url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>

2. Application Configuration

# application.yml
spring:
kafka:
bootstrap-servers: localhost:9092
properties:
schema.registry.url: http://localhost:8081
ksql:
server: http://localhost:8088
basic-auth:
username: ${KSQL_USERNAME:}
password: ${KSQL_PASSWORD:}
logging:
level:
io.confluent.ksql: INFO
org.apache.kafka: INFO

KSQL Client Implementation

1. KSQL REST Client Service

@Component
@Slf4j
public class KsqlDbClientService {
private final KsqlClient ksqlClient;
private final ObjectMapper objectMapper;
public KsqlDbClientService(@Value("${ksql.server}") String ksqlServerUrl,
ObjectMapper objectMapper) {
this.ksqlClient = KsqlClient.create(ksqlServerUrl);
this.objectMapper = objectMapper;
}
/**
* Execute KSQL query and return results as list
*/
public <T> List<T> executeQuery(String ksql, Class<T> resultType) {
List<T> results = new ArrayList<>();
try {
StreamedQueryResult streamedResult = ksqlClient.streamQuery(ksql);
while (streamedResult.hasNext()) {
Row row = streamedResult.next();
T result = convertRowToObject(row, resultType);
if (result != null) {
results.add(result);
}
}
} catch (Exception e) {
log.error("Error executing KSQL query: {}", ksql, e);
throw new KsqlExecutionException("Failed to execute KSQL query", e);
}
return results;
}
/**
* Execute KSQL statement (CREATE STREAM, CREATE TABLE, etc.)
*/
public void executeStatement(String ksqlStatement) {
try {
ExecuteStatementResult result = ksqlClient.executeStatement(ksqlStatement);
log.info("KSQL statement executed successfully: {}", ksqlStatement);
} catch (Exception e) {
log.error("Error executing KSQL statement: {}", ksqlStatement, e);
throw new KsqlExecutionException("Failed to execute KSQL statement", e);
}
}
/**
* Execute KSQL with parameters
*/
public <T> List<T> executeParameterizedQuery(String ksql, 
Map<String, Object> parameters, 
Class<T> resultType) {
List<T> results = new ArrayList<>();
try {
StreamedQueryResult streamedResult = ksqlClient.streamQuery(ksql, parameters);
while (streamedResult.hasNext()) {
Row row = streamedResult.next();
T result = convertRowToObject(row, resultType);
if (result != null) {
results.add(result);
}
}
} catch (Exception e) {
log.error("Error executing parameterized KSQL query: {}", ksql, e);
throw new KsqlExecutionException("Failed to execute parameterized KSQL query", e);
}
return results;
}
/**
* Insert values into a stream
*/
public void insertIntoStream(String streamName, Map<String, Object> values) {
try {
ksqlClient.insertInto(streamName, values);
log.debug("Inserted values into stream: {} - {}", streamName, values);
} catch (Exception e) {
log.error("Error inserting into stream: {}", streamName, e);
throw new KsqlExecutionException("Failed to insert into stream", e);
}
}
/**
* Get stream/table description
*/
public DescribeResult describeSource(String sourceName) {
try {
return ksqlClient.describeSource(sourceName);
} catch (Exception e) {
log.error("Error describing source: {}", sourceName, e);
throw new KsqlExecutionException("Failed to describe source", e);
}
}
/**
* List available streams
*/
public List<SourceDescription> listStreams() {
try {
return ksqlClient.listStreams();
} catch (Exception e) {
log.error("Error listing streams", e);
throw new KsqlExecutionException("Failed to list streams", e);
}
}
/**
* List available tables
*/
public List<SourceDescription> listTables() {
try {
return ksqlClient.listTables();
} catch (Exception e) {
log.error("Error listing tables", e);
throw new KsqlExecutionException("Failed to list tables", e);
}
}
/**
* Drop stream
*/
public void dropStream(String streamName) {
try {
ksqlClient.executeStatement("DROP STREAM " + streamName + ";");
log.info("Dropped stream: {}", streamName);
} catch (Exception e) {
log.error("Error dropping stream: {}", streamName, e);
throw new KsqlExecutionException("Failed to drop stream", e);
}
}
/**
* Drop table
*/
public void dropTable(String tableName) {
try {
ksqlClient.executeStatement("DROP TABLE " + tableName + ";");
log.info("Dropped table: {}", tableName);
} catch (Exception e) {
log.error("Error dropping table: {}", tableName, e);
throw new KsqlExecutionException("Failed to drop table", e);
}
}
@SuppressWarnings("unchecked")
private <T> T convertRowToObject(Row row, Class<T> resultType) {
try {
if (row.getRow().isPresent()) {
KsqlObject ksqlObject = row.getRow().get();
Map<String, Object> rowMap = new HashMap<>();
for (String column : ksqlObject.keySet()) {
rowMap.put(column, ksqlObject.get(column));
}
return objectMapper.convertValue(rowMap, resultType);
}
} catch (Exception e) {
log.warn("Failed to convert row to object: {}", row, e);
}
return null;
}
}

2. Domain Models for Stream Processing

// Event Models
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class UserEvent {
private String userId;
private String eventType;
private String pageId;
private Long timestamp;
private Map<String, Object> properties;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class PurchaseEvent {
private String orderId;
private String userId;
private String productId;
private String productName;
private Double price;
private Integer quantity;
private Long timestamp;
private String currency;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ProductViewEvent {
private String userId;
private String productId;
private String productName;
private String category;
private Long viewDurationMs;
private Long timestamp;
}
// KSQL Result Models
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class UserSession {
private String userId;
private Long sessionStart;
private Long sessionEnd;
private Long durationMs;
private Integer pageViewCount;
private String mostViewedPage;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class RealTimeMetrics {
private String metricName;
private Double value;
private Long windowStart;
private Long windowEnd;
private String dimension;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class TopProduct {
private String productId;
private String productName;
private Long viewCount;
private Double totalRevenue;
private Long timestamp;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class FraudAlert {
private String userId;
private String alertType;
private String description;
private Double riskScore;
private Long timestamp;
private String transactionId;
}

Stream Processing Service Implementations

1. Real-time Analytics Service

@Service
@Slf4j
public class RealTimeAnalyticsService {
private final KsqlDbClientService ksqlClient;
private final KafkaTemplate<String, Object> kafkaTemplate;
public RealTimeAnalyticsService(KsqlDbClientService ksqlClient,
KafkaTemplate<String, Object> kafkaTemplate) {
this.ksqlClient = ksqlClient;
this.kafkaTemplate = kafkaTemplate;
}
/**
* Initialize KSQL streams and tables
*/
@PostConstruct
public void initializeKsqlResources() {
createUserEventsStream();
createPurchaseEventsStream();
createProductViewsStream();
createRealTimeAggregations();
createFraudDetectionQueries();
}
private void createUserEventsStream() {
String createStreamSql = """
CREATE STREAM user_events_stream (
user_id VARCHAR,
event_type VARCHAR,
page_id VARCHAR,
timestamp BIGINT,
properties MAP<VARCHAR, VARCHAR>
) WITH (
KAFKA_TOPIC = 'user-events',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
);
""";
ksqlClient.executeStatement(createStreamSql);
log.info("Created user_events_stream");
}
private void createPurchaseEventsStream() {
String createStreamSql = """
CREATE STREAM purchase_events_stream (
order_id VARCHAR,
user_id VARCHAR,
product_id VARCHAR,
product_name VARCHAR,
price DOUBLE,
quantity INT,
timestamp BIGINT,
currency VARCHAR
) WITH (
KAFKA_TOPIC = 'purchase-events',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
);
""";
ksqlClient.executeStatement(createStreamSql);
log.info("Created purchase_events_stream");
}
private void createProductViewsStream() {
String createStreamSql = """
CREATE STREAM product_views_stream (
user_id VARCHAR,
product_id VARCHAR,
product_name VARCHAR,
category VARCHAR,
view_duration_ms BIGINT,
timestamp BIGINT
) WITH (
KAFKA_TOPIC = 'product-views',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
);
""";
ksqlClient.executeStatement(createStreamSql);
log.info("Created product_views_stream");
}
private void createRealTimeAggregations() {
// Real-time revenue by product
String revenueByProductSql = """
CREATE TABLE revenue_by_product AS
SELECT
product_id,
product_name,
SUM(price * quantity) AS total_revenue,
COUNT(*) AS order_count
FROM purchase_events_stream
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY product_id, product_name;
""";
ksqlClient.executeStatement(revenueByProductSql);
// User session aggregation
String userSessionsSql = """
CREATE TABLE user_sessions AS
SELECT
user_id,
WINDOWSTART AS session_start,
WINDOWEND AS session_end,
COUNT(*) AS page_view_count,
AS_VALUE(LATEST_BY_OFFSET(page_id)) AS last_page_viewed
FROM user_events_stream
WINDOW SESSION (30 MINUTES)
GROUP BY user_id;
""";
ksqlClient.executeStatement(userSessionsSql);
// Top products by views
String topProductsSql = """
CREATE TABLE top_products AS
SELECT
product_id,
product_name,
COUNT(*) AS view_count
FROM product_views_stream
WINDOW TUMBLING (SIZE 10 MINUTES)
GROUP BY product_id, product_name
EMIT CHANGES;
""";
ksqlClient.executeStatement(topProductsSql);
log.info("Created real-time aggregations");
}
private void createFraudDetectionQueries() {
// High-frequency purchase detection
String frequentPurchasesSql = """
CREATE STREAM potential_fraud_alerts AS
SELECT
user_id,
'HIGH_FREQUENCY_PURCHASES' AS alert_type,
'User made more than 5 purchases in 10 minutes' AS description,
COUNT(*) AS purchase_count,
0.8 AS risk_score,
ROWTIME AS timestamp
FROM purchase_events_stream
WINDOW TUMBLING (SIZE 10 MINUTES)
GROUP BY user_id
HAVING COUNT(*) > 5;
""";
ksqlClient.executeStatement(frequentPurchasesSql);
// Large amount purchases
String largeAmountSql = """
CREATE STREAM large_amount_alerts AS
SELECT
user_id,
'LARGE_AMOUNT_PURCHASE' AS alert_type,
'Single purchase exceeds $1000' AS description,
(price * quantity) AS purchase_amount,
0.9 AS risk_score,
ROWTIME AS timestamp,
order_id
FROM purchase_events_stream
WHERE (price * quantity) > 1000;
""";
ksqlClient.executeStatement(largeAmountSql);
log.info("Created fraud detection queries");
}
/**
* Get real-time revenue metrics
*/
public List<RealTimeMetrics> getRevenueMetrics(String timeWindow) {
String query = """
SELECT
'REVENUE' AS metric_name,
SUM(total_price) AS value,
WINDOWSTART AS window_start,
WINDOWEND AS window_end
FROM (
SELECT
product_id,
price * quantity AS total_price
FROM purchase_events_stream
)
WINDOW TUMBLING (SIZE %s)
GROUP BY 'REVENUE';
""".formatted(timeWindow);
return ksqlClient.executeQuery(query, RealTimeMetrics.class);
}
/**
* Get top products by views
*/
public List<TopProduct> getTopProducts(int limit) {
String query = """
SELECT
product_id,
product_name,
view_count,
ROWTIME AS timestamp
FROM top_products
ORDER BY view_count DESC
LIMIT %d;
""".formatted(limit);
return ksqlClient.executeQuery(query, TopProduct.class);
}
/**
* Get user session analytics
*/
public List<UserSession> getUserSessions(String userId) {
String query = """
SELECT
user_id,
session_start,
session_end,
(session_end - session_start) AS duration_ms,
page_view_count,
last_page_viewed AS most_viewed_page
FROM user_sessions
WHERE user_id = '%s';
""".formatted(userId);
return ksqlClient.executeQuery(query, UserSession.class);
}
/**
* Get real-time fraud alerts
*/
public List<FraudAlert> getFraudAlerts() {
String query = """
SELECT
user_id,
alert_type,
description,
risk_score,
timestamp,
transaction_id
FROM potential_fraud_alerts
UNION
SELECT
user_id,
alert_type,
description,
risk_score,
timestamp,
order_id AS transaction_id
FROM large_amount_alerts
ORDER BY timestamp DESC
LIMIT 100;
""";
return ksqlClient.executeQuery(query, FraudAlert.class);
}
/**
* Calculate conversion rates
*/
public Double calculateConversionRate(String productId) {
String query = """
SELECT
(purchase_count / view_count) AS conversion_rate
FROM (
SELECT
COUNT(*) AS purchase_count
FROM purchase_events_stream
WHERE product_id = '%s'
WINDOW TUMBLING (SIZE 1 HOUR)
) purchases
JOIN (
SELECT
COUNT(*) AS view_count
FROM product_views_stream
WHERE product_id = '%s'
WINDOW TUMBLING (SIZE 1 HOUR)
) views ON 1=1;
""".formatted(productId, productId);
List<Map<String, Object>> results = ksqlClient.executeQuery(query, Map.class);
if (!results.isEmpty()) {
Object conversionRate = results.get(0).get("CONVERSION_RATE");
return conversionRate != null ? ((Number) conversionRate).doubleValue() : 0.0;
}
return 0.0;
}
}

2. Event Processing Service

@Service
@Slf4j
public class EventProcessingService {
private final KsqlDbClientService ksqlClient;
private final KafkaTemplate<String, Object> kafkaTemplate;
public EventProcessingService(KsqlDbClientService ksqlClient,
KafkaTemplate<String, Object> kafkaTemplate) {
this.ksqlClient = ksqlClient;
this.kafkaTemplate = kafkaTemplate;
}
/**
* Send user event to KSQL stream
*/
public void sendUserEvent(UserEvent userEvent) {
Map<String, Object> eventData = new HashMap<>();
eventData.put("user_id", userEvent.getUserId());
eventData.put("event_type", userEvent.getEventType());
eventData.put("page_id", userEvent.getPageId());
eventData.put("timestamp", userEvent.getTimestamp());
eventData.put("properties", userEvent.getProperties());
ksqlClient.insertIntoStream("user_events_stream", eventData);
// Also send to Kafka topic for other consumers
kafkaTemplate.send("user-events", userEvent.getUserId(), userEvent);
log.debug("Sent user event: {}", userEvent.getUserId());
}
/**
* Send purchase event to KSQL stream
*/
public void sendPurchaseEvent(PurchaseEvent purchaseEvent) {
Map<String, Object> eventData = new HashMap<>();
eventData.put("order_id", purchaseEvent.getOrderId());
eventData.put("user_id", purchaseEvent.getUserId());
eventData.put("product_id", purchaseEvent.getProductId());
eventData.put("product_name", purchaseEvent.getProductName());
eventData.put("price", purchaseEvent.getPrice());
eventData.put("quantity", purchaseEvent.getQuantity());
eventData.put("timestamp", purchaseEvent.getTimestamp());
eventData.put("currency", purchaseEvent.getCurrency());
ksqlClient.insertIntoStream("purchase_events_stream", eventData);
kafkaTemplate.send("purchase-events", purchaseEvent.getUserId(), purchaseEvent);
log.debug("Sent purchase event: {}", purchaseEvent.getOrderId());
}
/**
* Send product view event to KSQL stream
*/
public void sendProductViewEvent(ProductViewEvent productViewEvent) {
Map<String, Object> eventData = new HashMap<>();
eventData.put("user_id", productViewEvent.getUserId());
eventData.put("product_id", productViewEvent.getProductId());
eventData.put("product_name", productViewEvent.getProductName());
eventData.put("category", productViewEvent.getCategory());
eventData.put("view_duration_ms", productViewEvent.getViewDurationMs());
eventData.put("timestamp", productViewEvent.getTimestamp());
ksqlClient.insertIntoStream("product_views_stream", eventData);
kafkaTemplate.send("product-views", productViewEvent.getUserId(), productViewEvent);
log.debug("Sent product view event: {}", productViewEvent.getProductId());
}
/**
* Process real-time user behavior
*/
public void processUserBehavior(String userId) {
String behaviorQuery = """
SELECT
ue.user_id,
COUNT(*) AS total_events,
COUNT_DISTINCT(ue.page_id) AS unique_pages,
MAX(ue.timestamp) AS last_activity,
COALESCE(COUNT(pe.order_id), 0) AS purchase_count
FROM user_events_stream ue
LEFT JOIN purchase_events_stream pe
WITHIN 24 HOURS
ON ue.user_id = pe.user_id
WHERE ue.user_id = '%s'
GROUP BY ue.user_id;
""".formatted(userId);
List<Map<String, Object>> results = ksqlClient.executeQuery(behaviorQuery, Map.class);
if (!results.isEmpty()) {
Map<String, Object> behavior = results.get(0);
log.info("User behavior analysis: {}", behavior);
// Trigger personalized recommendations or alerts
triggerPersonalization(userId, behavior);
}
}
/**
* Generate real-time recommendations
*/
public List<String> generateRecommendations(String userId) {
String recommendationQuery = """
SELECT
pv.product_id,
pv.product_name,
COUNT(*) AS view_count,
COALESCE(SUM(pe.quantity), 0) AS purchase_count
FROM product_views_stream pv
LEFT JOIN purchase_events_stream pe
WITHIN 7 DAYS
ON pv.product_id = pe.product_id
WHERE pv.user_id = '%s'
GROUP BY pv.product_id, pv.product_name
HAVING purchase_count = 0
ORDER BY view_count DESC
LIMIT 10;
""".formatted(userId);
List<Map<String, Object>> results = ksqlClient.executeQuery(recommendationQuery, Map.class);
return results.stream()
.map(result -> (String) result.get("PRODUCT_NAME"))
.collect(Collectors.toList());
}
private void triggerPersonalization(String userId, Map<String, Object> behavior) {
Long totalEvents = (Long) behavior.get("TOTAL_EVENTS");
Long purchaseCount = (Long) behavior.get("PURCHASE_COUNT");
if (totalEvents > 50 && purchaseCount == 0) {
// High engagement but no purchases - trigger special offer
log.info("Triggering special offer for user: {}", userId);
sendPersonalizedOffer(userId);
}
}
private void sendPersonalizedOffer(String userId) {
// Implementation for sending personalized offers
log.info("Sending personalized offer to user: {}", userId);
}
}

Advanced KSQL Patterns

1. Complex Event Processing

@Service
@Slf4j
public class ComplexEventProcessingService {
private final KsqlDbClientService ksqlClient;
public ComplexEventProcessingService(KsqlDbClientService ksqlClient) {
this.ksqlClient = ksqlClient;
}
/**
* Detect abandoned carts
*/
public void setupAbandonedCartDetection() {
String abandonedCartSql = """
CREATE STREAM abandoned_carts AS
SELECT
user_id,
'ABANDONED_CART' AS alert_type,
COLLECT_LIST(product_id) AS viewed_products,
MAX(timestamp) AS last_view_time,
ROWTIME AS alert_time
FROM product_views_stream pv
WHERE NOT EXISTS (
SELECT 1
FROM purchase_events_stream pe
WHERE pv.user_id = pe.user_id
AND pv.product_id = pe.product_id
AND pe.timestamp > pv.timestamp
)
WINDOW SESSION (30 MINUTES)
GROUP BY user_id
HAVING COUNT(*) >= 2;
""";
ksqlClient.executeStatement(abandonedCartSql);
log.info("Setup abandoned cart detection");
}
/**
* Real-time inventory management
*/
public void setupInventoryManagement() {
String inventorySql = """
CREATE TABLE product_inventory AS
SELECT
product_id,
product_name,
SUM(quantity) AS total_sold,
LATEST_BY_OFFSET(price) AS current_price
FROM purchase_events_stream
GROUP BY product_id, product_name;
""";
ksqlClient.executeStatement(inventorySql);
String lowStockAlertSql = """
CREATE STREAM low_stock_alerts AS
SELECT
pi.product_id,
pi.product_name,
pi.total_sold,
'LOW_STOCK_ALERT' AS alert_type,
ROWTIME AS alert_time
FROM product_inventory pi
WHERE pi.total_sold > 100;  -- Threshold for low stock
""";
ksqlClient.executeStatement(lowStockAlertSql);
log.info("Setup inventory management");
}
/**
* Customer segmentation
*/
public void setupCustomerSegmentation() {
String customerSegmentsSql = """
CREATE TABLE customer_segments AS
SELECT
user_id,
CASE
WHEN total_spent > 1000 THEN 'VIP'
WHEN total_spent > 500 THEN 'PREMIUM'
WHEN total_spent > 100 THEN 'STANDARD'
ELSE 'NEW'
END AS segment,
total_spent,
order_count
FROM (
SELECT
user_id,
SUM(price * quantity) AS total_spent,
COUNT(*) AS order_count
FROM purchase_events_stream
GROUP BY user_id
);
""";
ksqlClient.executeStatement(customerSegmentsSql);
log.info("Setup customer segmentation");
}
/**
* Real-time A/B testing
*/
public void setupAbTesting() {
String abTestSql = """
CREATE STREAM ab_test_results AS
SELECT
properties['experiment_id'] AS experiment_id,
properties['variant'] AS variant,
event_type,
COUNT(*) AS event_count,
COUNT_DISTINCT(user_id) AS unique_users
FROM user_events_stream
WHERE event_type IN ('conversion', 'click', 'view')
AND properties['experiment_id'] IS NOT NULL
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY
properties['experiment_id'],
properties['variant'],
event_type;
""";
ksqlClient.executeStatement(abTestSql);
log.info("Setup A/B testing analytics");
}
/**
* Get abandoned cart alerts
*/
public List<Map<String, Object>> getAbandonedCarts() {
String query = """
SELECT
user_id,
viewed_products,
last_view_time,
alert_time
FROM abandoned_carts
ORDER BY alert_time DESC
LIMIT 50;
""";
return ksqlClient.executeQuery(query, Map.class);
}
/**
* Get customer segments
*/
public List<Map<String, Object>> getCustomerSegments() {
String query = """
SELECT
user_id,
segment,
total_spent,
order_count
FROM customer_segments
ORDER BY total_spent DESC
LIMIT 100;
""";
return ksqlClient.executeQuery(query, Map.class);
}
}

2. Windowed Aggregations Service

@Service
@Slf4j
public class WindowedAggregationService {
private final KsqlDbClientService ksqlClient;
public WindowedAggregationService(KsqlDbClientService ksqlClient) {
this.ksqlClient = ksqlClient;
}
/**
* Setup various windowed aggregations
*/
public void setupWindowedAggregations() {
setupTumblingWindowAggregations();
setupHoppingWindowAggregations();
setupSessionWindowAggregations();
}
private void setupTumblingWindowAggregations() {
// Tumbling window - fixed, non-overlapping windows
String tumblingRevenueSql = """
CREATE TABLE tumbling_revenue_1min AS
SELECT
'REVENUE_1MIN' AS metric_type,
SUM(price * quantity) AS total_revenue,
COUNT(*) AS order_count,
WINDOWSTART AS window_start,
WINDOWEND AS window_end
FROM purchase_events_stream
WINDOW TUMBLING (SIZE 1 MINUTE)
GROUP BY 'REVENUE_1MIN';
""";
ksqlClient.executeStatement(tumblingRevenueSql);
String tumblingUserActivitySql = """
CREATE TABLE tumbling_user_activity_5min AS
SELECT
'USER_ACTIVITY_5MIN' AS metric_type,
COUNT_DISTINCT(user_id) AS active_users,
COUNT(*) AS total_events,
WINDOWSTART AS window_start,
WINDOWEND AS window_end
FROM user_events_stream
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY 'USER_ACTIVITY_5MIN';
""";
ksqlClient.executeStatement(tumblingUserActivitySql);
}
private void setupHoppingWindowAggregations() {
// Hopping window - overlapping windows
String hoppingRevenueSql = """
CREATE TABLE hopping_revenue_5min_1min AS
SELECT
'HOPPING_REVENUE' AS metric_type,
SUM(price * quantity) AS total_revenue,
WINDOWSTART AS window_start,
WINDOWEND AS window_end
FROM purchase_events_stream
WINDOW HOPPING (SIZE 5 MINUTES, ADVANCE BY 1 MINUTE)
GROUP BY 'HOPPING_REVENUE';
""";
ksqlClient.executeStatement(hoppingRevenueSql);
}
private void setupSessionWindowAggregations() {
// Session window - periods of activity separated by gaps of inactivity
String userSessionsSql = """
CREATE TABLE user_session_analytics AS
SELECT
user_id,
COUNT(*) AS event_count,
WINDOWSTART AS session_start,
WINDOWEND AS session_end,
(WINDOWEND - WINDOWSTART) AS session_duration_ms
FROM user_events_stream
WINDOW SESSION (30 MINUTES)
GROUP BY user_id;
""";
ksqlClient.executeStatement(userSessionsSql);
}
/**
* Get tumbling window metrics
*/
public List<Map<String, Object>> getTumblingWindowMetrics(String metricType) {
String query = """
SELECT
metric_type,
total_revenue,
order_count,
window_start,
window_end
FROM tumbling_revenue_1min
WHERE metric_type = '%s'
ORDER BY window_end DESC
LIMIT 10;
""".formatted(metricType);
return ksqlClient.executeQuery(query, Map.class);
}
/**
* Get hopping window metrics
*/
public List<Map<String, Object>> getHoppingWindowMetrics() {
String query = """
SELECT
metric_type,
total_revenue,
window_start,
window_end
FROM hopping_revenue_5min_1min
ORDER BY window_end DESC
LIMIT 20;
""";
return ksqlClient.executeQuery(query, Map.class);
}
/**
* Get user session analytics
*/
public List<Map<String, Object>> getUserSessions(String userId) {
String query = """
SELECT
user_id,
event_count,
session_start,
session_end,
session_duration_ms
FROM user_session_analytics
WHERE user_id = '%s'
ORDER BY session_end DESC
LIMIT 10;
""".formatted(userId);
return ksqlClient.executeQuery(query, Map.class);
}
}

REST API Controllers

1. Analytics Controller

@RestController
@RequestMapping("/api/analytics")
@Slf4j
public class AnalyticsController {
private final RealTimeAnalyticsService analyticsService;
private final EventProcessingService eventProcessingService;
private final ComplexEventProcessingService complexEventService;
public AnalyticsController(RealTimeAnalyticsService analyticsService,
EventProcessingService eventProcessingService,
ComplexEventProcessingService complexEventService) {
this.analyticsService = analyticsService;
this.eventProcessingService = eventProcessingService;
this.complexEventService = complexEventService;
}
@GetMapping("/revenue/metrics")
public ResponseEntity<List<RealTimeMetrics>> getRevenueMetrics(
@RequestParam(defaultValue = "1 HOUR") String window) {
try {
List<RealTimeMetrics> metrics = analyticsService.getRevenueMetrics(window);
return ResponseEntity.ok(metrics);
} catch (Exception e) {
log.error("Error fetching revenue metrics", e);
return ResponseEntity.internalServerError().build();
}
}
@GetMapping("/products/top")
public ResponseEntity<List<TopProduct>> getTopProducts(
@RequestParam(defaultValue = "10") int limit) {
try {
List<TopProduct> topProducts = analyticsService.getTopProducts(limit);
return ResponseEntity.ok(topProducts);
} catch (Exception e) {
log.error("Error fetching top products", e);
return ResponseEntity.internalServerError().build();
}
}
@GetMapping("/fraud/alerts")
public ResponseEntity<List<FraudAlert>> getFraudAlerts() {
try {
List<FraudAlert> alerts = analyticsService.getFraudAlerts();
return ResponseEntity.ok(alerts);
} catch (Exception e) {
log.error("Error fetching fraud alerts", e);
return ResponseEntity.internalServerError().build();
}
}
@GetMapping("/users/{userId}/sessions")
public ResponseEntity<List<UserSession>> getUserSessions(@PathVariable String userId) {
try {
List<UserSession> sessions = analyticsService.getUserSessions(userId);
return ResponseEntity.ok(sessions);
} catch (Exception e) {
log.error("Error fetching user sessions for: {}", userId, e);
return ResponseEntity.internalServerError().build();
}
}
@GetMapping("/users/{userId}/recommendations")
public ResponseEntity<List<String>> getUserRecommendations(@PathVariable String userId) {
try {
List<String> recommendations = eventProcessingService.generateRecommendations(userId);
return ResponseEntity.ok(recommendations);
} catch (Exception e) {
log.error("Error generating recommendations for: {}", userId, e);
return ResponseEntity.internalServerError().build();
}
}
@GetMapping("/abandoned-carts")
public ResponseEntity<List<Map<String, Object>>> getAbandonedCarts() {
try {
List<Map<String, Object>> abandonedCarts = complexEventService.getAbandonedCarts();
return ResponseEntity.ok(abandonedCarts);
} catch (Exception e) {
log.error("Error fetching abandoned carts", e);
return ResponseEntity.internalServerError().build();
}
}
@GetMapping("/customer-segments")
public ResponseEntity<List<Map<String, Object>>> getCustomerSegments() {
try {
List<Map<String, Object>> segments = complexEventService.getCustomerSegments();
return ResponseEntity.ok(segments);
} catch (Exception e) {
log.error("Error fetching customer segments", e);
return ResponseEntity.internalServerError().build();
}
}
@PostMapping("/events/user")
public ResponseEntity<Void> sendUserEvent(@RequestBody UserEvent userEvent) {
try {
eventProcessingService.sendUserEvent(userEvent);
return ResponseEntity.accepted().build();
} catch (Exception e) {
log.error("Error sending user event", e);
return ResponseEntity.internalServerError().build();
}
}
@PostMapping("/events/purchase")
public ResponseEntity<Void> sendPurchaseEvent(@RequestBody PurchaseEvent purchaseEvent) {
try {
eventProcessingService.sendPurchaseEvent(purchaseEvent);
return ResponseEntity.accepted().build();
} catch (Exception e) {
log.error("Error sending purchase event", e);
return ResponseEntity.internalServerError().build();
}
}
@PostMapping("/events/product-view")
public ResponseEntity<Void> sendProductViewEvent(@RequestBody ProductViewEvent productViewEvent) {
try {
eventProcessingService.sendProductViewEvent(productViewEvent);
return ResponseEntity.accepted().build();
} catch (Exception e) {
log.error("Error sending product view event", e);
return ResponseEntity.internalServerError().build();
}
}
}

Testing KSQL Integration

1. Integration Tests

@SpringBootTest
@Testcontainers
@ActiveProfiles("test")
class KsqlIntegrationTest {
@Container
static KafkaContainer kafka = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:7.5.1")
);
@Autowired
private KsqlDbClientService ksqlClient;
@Autowired
private RealTimeAnalyticsService analyticsService;
@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
registry.add("ksql.server", () -> "http://localhost:8088");
}
@Test
void testKsqlClientConnection() {
assertDoesNotThrow(() -> {
List<SourceDescription> streams = ksqlClient.listStreams();
assertNotNull(streams);
});
}
@Test
void testStreamCreation() {
String testStreamSql = """
CREATE STREAM test_stream (
id VARCHAR,
value INT,
timestamp BIGINT
) WITH (
KAFKA_TOPIC = 'test-topic',
VALUE_FORMAT = 'JSON',
PARTITIONS = 1
);
""";
assertDoesNotThrow(() -> ksqlClient.executeStatement(testStreamSql));
// Verify stream was created
List<SourceDescription> streams = ksqlClient.listStreams();
boolean streamExists = streams.stream()
.anyMatch(stream -> stream.getName().equals("TEST_STREAM"));
assertTrue(streamExists, "Test stream should be created");
// Cleanup
ksqlClient.dropStream("TEST_STREAM");
}
@Test
void testEventProcessing() {
PurchaseEvent purchaseEvent = PurchaseEvent.builder()
.orderId("test-order-123")
.userId("test-user-456")
.productId("test-product-789")
.productName("Test Product")
.price(99.99)
.quantity(2)
.timestamp(System.currentTimeMillis())
.currency("USD")
.build();
// This would typically be tested with a full KSQL DB setup
assertDoesNotThrow(() -> {
// In a real test, you would verify the event was processed
// and appeared in the appropriate streams/tables
});
}
}

Error Handling and Monitoring

1. Custom Exceptions

public class KsqlExecutionException extends RuntimeException {
public KsqlExecutionException(String message) {
super(message);
}
public KsqlExecutionException(String message, Throwable cause) {
super(message, cause);
}
}
public class StreamProcessingException extends RuntimeException {
public StreamProcessingException(String message) {
super(message);
}
public StreamProcessingException(String message, Throwable cause) {
super(message, cause);
}
}

2. Monitoring and Metrics

@Component
@Slf4j
public class KsqlMetricsMonitor {
private final MeterRegistry meterRegistry;
private final Counter ksqlQueryCounter;
private final Timer ksqlExecutionTimer;
private final Counter ksqlErrorCounter;
public KsqlMetricsMonitor(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
this.ksqlQueryCounter = meterRegistry.counter("ksql.queries.executed");
this.ksqlExecutionTimer = meterRegistry.timer("ksql.execution.time");
this.ksqlErrorCounter = meterRegistry.counter("ksql.errors");
}
public void recordQueryExecution(String queryType, Duration duration, boolean success) {
ksqlQueryCounter.increment();
ksqlExecutionTimer.record(duration);
Tags tags = Tags.of(
Tag.of("query_type", queryType),
Tag.of("success", String.valueOf(success))
);
meterRegistry.counter("ksql.queries.detailed", tags).increment();
meterRegistry.timer("ksql.execution.time.detailed", tags).record(duration);
if (!success) {
ksqlErrorCounter.increment();
}
log.debug("KSQL query executed - type: {}, duration: {}ms, success: {}", 
queryType, duration.toMillis(), success);
}
public void recordStreamEvent(String streamName, String eventType) {
Tags tags = Tags.of(
Tag.of("stream_name", streamName),
Tag.of("event_type", eventType)
);
meterRegistry.counter("ksql.stream.events", tags).increment();
}
}

Configuration Best Practices

1. Application Configuration

# application-prod.yml
spring:
kafka:
bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
properties:
schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
security.protocol: SASL_SSL
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="${KAFKA_USERNAME}" password="${KAFKA_PASSWORD}";
ksql:
server: ${KSQL_SERVER:http://localhost:8088}
basic-auth:
username: ${KSQL_USERNAME:}
password: ${KSQL_PASSWORD:}
management:
endpoints:
web:
exposure:
include: health,metrics,info
endpoint:
health:
show-details: always
metrics:
export:
prometheus:
enabled: true
logging:
level:
io.confluent.ksql: WARN
org.apache.kafka: WARN
com.example.ksql: INFO

Conclusion

KSQL provides a powerful SQL-like interface for real-time stream processing on Kafka, and when combined with Java applications, it enables building sophisticated real-time analytics and event processing systems.

Key Benefits:

  • SQL Familiarity: Leverage existing SQL knowledge for stream processing
  • Real-time Processing: Process data as it arrives with sub-second latency
  • Scalability: Built on Kafka's scalable architecture
  • Integration: Seamless integration with Java applications and microservices

Use Cases:

  • Real-time Analytics: Dashboards, metrics, and business intelligence
  • Fraud Detection: Pattern matching and anomaly detection
  • Recommendation Engines: Real-time personalization
  • IoT Data Processing: Sensor data aggregation and alerting
  • Customer 360: Real-time customer behavior analysis

Best Practices:

  • Start Simple: Begin with basic streams and tables
  • Monitor Performance: Track query performance and resource usage
  • Use Windowing: Leverage tumbling, hopping, and session windows appropriately
  • Handle Errors: Implement robust error handling and retry mechanisms
  • Security: Secure KSQL clusters and implement proper authentication

By combining KSQL's streaming SQL capabilities with Java's enterprise features, you can build powerful, scalable real-time applications that process massive volumes of data with low latency and high reliability.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper