MACRO NEPAL
CloudEvents SDK in Java: Building Event-Driven Applications
awiskar acharya
November 18, 2025
9:44 am

CloudEvents is a specification for describing event data in a common way, providing interoperability across services, platforms, and systems. The CloudEvents SDK for Java helps create, send, receive, and process CloudEvents.

Core Concepts

What are CloudEvents?

  • Standardized format for event data
  • Platform-agnostic event description
  • Enables interoperability between event producers and consumers
  • Supports multiple transport protocols (HTTP, Kafka, AMQP, etc.)

CloudEvents Structure:

  • Required Attributes: specversion, id, source, type
  • Optional Attributes: datacontenttype, dataschema, subject, time, data
  • Extensions: Custom attributes for specific use cases

Dependencies and Setup

1. Maven Dependencies
<properties>
<cloudevents.version>3.0.0</cloudevents.version>
<jackson.version>2.15.2</jackson.version>
<spring-boot.version>3.1.0</spring-boot.version>
<kafka.version>3.4.0</kafka.version>
</properties>
<dependencies>
<!-- CloudEvents SDK -->
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-core</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<!-- CloudEvents HTTP Transport -->
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-http-vertx</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<!-- CloudEvents Kafka Transport -->
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-kafka</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<!-- CloudEvents Jackson JSON Binding -->
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-json-jackson</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<!-- Spring Boot (Optional) -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>${spring-boot.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<version>${spring-boot.version}</version>
<optional>true</optional>
</dependency>
<!-- Kafka (Optional) -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<!-- Jackson for JSON processing -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>${jackson.version}</version>
</dependency>
</dependencies>

Core Implementation

1. CloudEvents Factory and Builder
package com.example.cloudevents.core;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.data.PojoCloudEventData;
import io.cloudevents.jackson.PojoCloudEventDataMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
import java.time.OffsetDateTime;
import java.util.UUID;
public class CloudEventFactory {
private static final Logger LOG = LoggerFactory.getLogger(CloudEventFactory.class);
private final String eventSource;
private final ObjectMapper objectMapper;
public CloudEventFactory(String eventSource, ObjectMapper objectMapper) {
this.eventSource = eventSource;
this.objectMapper = objectMapper;
}
public CloudEvent createEvent(String eventType, Object data) {
return createEvent(eventType, data, null, null);
}
public CloudEvent createEvent(String eventType, Object data, String subject) {
return createEvent(eventType, data, subject, null);
}
public CloudEvent createEvent(String eventType, Object data, String subject, 
OffsetDateTime eventTime) {
LOG.debug("Creating CloudEvent of type: {} from source: {}", eventType, eventSource);
try {
CloudEventBuilder builder = CloudEventBuilder.v1()
.withId(generateEventId())
.withSource(URI.create(eventSource))
.withType(eventType)
.withDataContentType("application/json");
if (subject != null) {
builder.withSubject(subject);
}
if (eventTime != null) {
builder.withTime(eventTime);
} else {
builder.withTime(OffsetDateTime.now());
}
// Add data
if (data != null) {
PojoCloudEventData<Object> eventData = PojoCloudEventData.wrap(
data, 
objectMapper::writeValueAsBytes
);
builder.withData(eventData);
}
return builder.build();
} catch (Exception e) {
LOG.error("Failed to create CloudEvent: {}", e.getMessage(), e);
throw new CloudEventException("Failed to create CloudEvent", e);
}
}
public <T> T extractData(CloudEvent cloudEvent, Class<T> dataType) {
try {
if (cloudEvent.getData() == null) {
return null;
}
PojoCloudEventData<T> eventData = PojoCloudEventDataMapper.from(
objectMapper, 
dataType
).map(cloudEvent.getData());
return eventData.getValue();
} catch (Exception e) {
LOG.error("Failed to extract data from CloudEvent: {}", e.getMessage(), e);
throw new CloudEventException("Failed to extract event data", e);
}
}
public EventMetadata extractMetadata(CloudEvent cloudEvent) {
return EventMetadata.builder()
.id(cloudEvent.getId())
.source(cloudEvent.getSource().toString())
.type(cloudEvent.getType())
.subject(cloudEvent.getSubject())
.time(cloudEvent.getTime())
.dataContentType(cloudEvent.getDataContentType())
.build();
}
private String generateEventId() {
return UUID.randomUUID().toString();
}
// Domain-specific event creators
public CloudEvent createUserCreatedEvent(UserCreatedData data) {
return createEvent("com.example.user.created", data, data.getUserId());
}
public CloudEvent createUserUpdatedEvent(UserUpdatedData data) {
return createEvent("com.example.user.updated", data, data.getUserId());
}
public CloudEvent createOrderPlacedEvent(OrderPlacedData data) {
return createEvent("com.example.order.placed", data, data.getOrderId());
}
public CloudEvent createPaymentProcessedEvent(PaymentProcessedData data) {
return createEvent("com.example.payment.processed", data, data.getPaymentId());
}
}
class CloudEventException extends RuntimeException {
public CloudEventException(String message) {
super(message);
}
public CloudEventException(String message, Throwable cause) {
super(message, cause);
}
}
2. Event Data Models
package com.example.cloudevents.model;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonInclude;
import java.time.LocalDateTime;
import java.util.Map;
// Base event data class
@JsonInclude(JsonInclude.Include.NON_NULL)
public abstract class BaseEventData {
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
private LocalDateTime timestamp;
private String correlationId;
private Map<String, Object> metadata;
public BaseEventData() {
this.timestamp = LocalDateTime.now();
}
// Getters and setters
public LocalDateTime getTimestamp() { return timestamp; }
public void setTimestamp(LocalDateTime timestamp) { this.timestamp = timestamp; }
public String getCorrelationId() { return correlationId; }
public void setCorrelationId(String correlationId) { this.correlationId = correlationId; }
public Map<String, Object> getMetadata() { return metadata; }
public void setMetadata(Map<String, Object> metadata) { this.metadata = metadata; }
}
// Domain-specific event data models
public class UserCreatedData extends BaseEventData {
private String userId;
private String email;
private String firstName;
private String lastName;
private UserStatus status;
// Constructors
public UserCreatedData() {}
public UserCreatedData(String userId, String email, String firstName, String lastName) {
this.userId = userId;
this.email = email;
this.firstName = firstName;
this.lastName = lastName;
this.status = UserStatus.ACTIVE;
}
// Getters and setters
public String getUserId() { return userId; }
public void setUserId(String userId) { this.userId = userId; }
public String getEmail() { return email; }
public void setEmail(String email) { this.email = email; }
public String getFirstName() { return firstName; }
public void setFirstName(String firstName) { this.firstName = firstName; }
public String getLastName() { return lastName; }
public void setLastName(String lastName) { this.lastName = lastName; }
public UserStatus getStatus() { return status; }
public void setStatus(UserStatus status) { this.status = status; }
}
public class UserUpdatedData extends BaseEventData {
private String userId;
private String email;
private String firstName;
private String lastName;
private UserStatus previousStatus;
private UserStatus newStatus;
private Map<String, Object> updatedFields;
// Constructors, getters, and setters
public UserUpdatedData() {}
public UserUpdatedData(String userId, Map<String, Object> updatedFields) {
this.userId = userId;
this.updatedFields = updatedFields;
}
public String getUserId() { return userId; }
public void setUserId(String userId) { this.userId = userId; }
public String getEmail() { return email; }
public void setEmail(String email) { this.email = email; }
public String getFirstName() { return firstName; }
public void setFirstName(String firstName) { this.firstName = firstName; }
public String getLastName() { return lastName; }
public void setLastName(String lastName) { this.lastName = lastName; }
public UserStatus getPreviousStatus() { return previousStatus; }
public void setPreviousStatus(UserStatus previousStatus) { this.previousStatus = previousStatus; }
public UserStatus getNewStatus() { return newStatus; }
public void setNewStatus(UserStatus newStatus) { this.newStatus = newStatus; }
public Map<String, Object> getUpdatedFields() { return updatedFields; }
public void setUpdatedFields(Map<String, Object> updatedFields) { this.updatedFields = updatedFields; }
}
public class OrderPlacedData extends BaseEventData {
private String orderId;
private String customerId;
private BigDecimal totalAmount;
private List<OrderItem> items;
private String currency;
private ShippingAddress shippingAddress;
// Constructors, getters, and setters
public OrderPlacedData() {}
public OrderPlacedData(String orderId, String customerId, BigDecimal totalAmount) {
this.orderId = orderId;
this.customerId = customerId;
this.totalAmount = totalAmount;
this.currency = "USD";
}
public String getOrderId() { return orderId; }
public void setOrderId(String orderId) { this.orderId = orderId; }
public String getCustomerId() { return customerId; }
public void setCustomerId(String customerId) { this.customerId = customerId; }
public BigDecimal getTotalAmount() { return totalAmount; }
public void setTotalAmount(BigDecimal totalAmount) { this.totalAmount = totalAmount; }
public List<OrderItem> getItems() { return items; }
public void setItems(List<OrderItem> items) { this.items = items; }
public String getCurrency() { return currency; }
public void setCurrency(String currency) { this.currency = currency; }
public ShippingAddress getShippingAddress() { return shippingAddress; }
public void setShippingAddress(ShippingAddress shippingAddress) { this.shippingAddress = shippingAddress; }
}
public class PaymentProcessedData extends BaseEventData {
private String paymentId;
private String orderId;
private BigDecimal amount;
private String currency;
private PaymentStatus status;
private String gatewayTransactionId;
private LocalDateTime processedAt;
// Constructors, getters, and setters
public PaymentProcessedData() {}
public PaymentProcessedData(String paymentId, String orderId, BigDecimal amount, PaymentStatus status) {
this.paymentId = paymentId;
this.orderId = orderId;
this.amount = amount;
this.status = status;
this.currency = "USD";
this.processedAt = LocalDateTime.now();
}
public String getPaymentId() { return paymentId; }
public void setPaymentId(String paymentId) { this.paymentId = paymentId; }
public String getOrderId() { return orderId; }
public void setOrderId(String orderId) { this.orderId = orderId; }
public BigDecimal getAmount() { return amount; }
public void setAmount(BigDecimal amount) { this.amount = amount; }
public String getCurrency() { return currency; }
public void setCurrency(String currency) { this.currency = currency; }
public PaymentStatus getStatus() { return status; }
public void setStatus(PaymentStatus status) { this.status = status; }
public String getGatewayTransactionId() { return gatewayTransactionId; }
public void setGatewayTransactionId(String gatewayTransactionId) { this.gatewayTransactionId = gatewayTransactionId; }
public LocalDateTime getProcessedAt() { return processedAt; }
public void setProcessedAt(LocalDateTime processedAt) { this.processedAt = processedAt; }
}
// Supporting models
public class OrderItem {
private String productId;
private String productName;
private int quantity;
private BigDecimal unitPrice;
// Constructors, getters, and setters
public OrderItem() {}
public OrderItem(String productId, String productName, int quantity, BigDecimal unitPrice) {
this.productId = productId;
this.productName = productName;
this.quantity = quantity;
this.unitPrice = unitPrice;
}
public String getProductId() { return productId; }
public void setProductId(String productId) { this.productId = productId; }
public String getProductName() { return productName; }
public void setProductName(String productName) { this.productName = productName; }
public int getQuantity() { return quantity; }
public void setQuantity(int quantity) { this.quantity = quantity; }
public BigDecimal getUnitPrice() { return unitPrice; }
public void setUnitPrice(BigDecimal unitPrice) { this.unitPrice = unitPrice; }
}
public class ShippingAddress {
private String street;
private String city;
private String state;
private String zipCode;
private String country;
// Constructors, getters, and setters
public ShippingAddress() {}
public ShippingAddress(String street, String city, String state, String zipCode, String country) {
this.street = street;
this.city = city;
this.state = state;
this.zipCode = zipCode;
this.country = country;
}
public String getStreet() { return street; }
public void setStreet(String street) { this.street = street; }
public String getCity() { return city; }
public void setCity(String city) { this.city = city; }
public String getState() { return state; }
public void setState(String state) { this.state = state; }
public String getZipCode() { return zipCode; }
public void setZipCode(String zipCode) { this.zipCode = zipCode; }
public String getCountry() { return country; }
public void setCountry(String country) { this.country = country; }
}
enum UserStatus {
ACTIVE, INACTIVE, SUSPENDED, PENDING_VERIFICATION
}
enum PaymentStatus {
SUCCEEDED, FAILED, PENDING, REFUNDED, CANCELLED
}
3. Event Metadata and Validation
package com.example.cloudevents.model;
import java.time.OffsetDateTime;
public class EventMetadata {
private final String id;
private final String source;
private final String type;
private final String subject;
private final OffsetDateTime time;
private final String dataContentType;
private EventMetadata(Builder builder) {
this.id = builder.id;
this.source = builder.source;
this.type = builder.type;
this.subject = builder.subject;
this.time = builder.time;
this.dataContentType = builder.dataContentType;
}
// Getters
public String getId() { return id; }
public String getSource() { return source; }
public String getType() { return type; }
public String getSubject() { return subject; }
public OffsetDateTime getTime() { return time; }
public String getDataContentType() { return dataContentType; }
public static Builder builder() {
return new Builder();
}
public static class Builder {
private String id;
private String source;
private String type;
private String subject;
private OffsetDateTime time;
private String dataContentType;
public Builder id(String id) { this.id = id; return this; }
public Builder source(String source) { this.source = source; return this; }
public Builder type(String type) { this.type = type; return this; }
public Builder subject(String subject) { this.subject = subject; return this; }
public Builder time(OffsetDateTime time) { this.time = time; return this; }
public Builder dataContentType(String dataContentType) { this.dataContentType = dataContentType; return this; }
public EventMetadata build() {
return new EventMetadata(this);
}
}
}
package com.example.cloudevents.core;
import io.cloudevents.CloudEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.List;
public class CloudEventValidator {
private static final Logger LOG = LoggerFactory.getLogger(CloudEventValidator.class);
public ValidationResult validate(CloudEvent cloudEvent) {
List<String> errors = new ArrayList<>();
List<String> warnings = new ArrayList<>();
// Required field validation
if (cloudEvent.getId() == null || cloudEvent.getId().trim().isEmpty()) {
errors.add("Event ID is required");
}
if (cloudEvent.getSource() == null) {
errors.add("Event source is required");
} else {
try {
URI sourceUri = cloudEvent.getSource();
if (!sourceUri.isAbsolute()) {
warnings.add("Event source should be an absolute URI");
}
} catch (Exception e) {
errors.add("Event source is not a valid URI: " + e.getMessage());
}
}
if (cloudEvent.getType() == null || cloudEvent.getType().trim().isEmpty()) {
errors.add("Event type is required");
}
if (cloudEvent.getSpecVersion() == null) {
errors.add("Spec version is required");
}
// Time validation
if (cloudEvent.getTime() != null) {
OffsetDateTime eventTime = cloudEvent.getTime();
OffsetDateTime now = OffsetDateTime.now();
if (eventTime.isAfter(now.plusMinutes(5))) {
warnings.add("Event time is in the future");
}
if (eventTime.isBefore(now.minusDays(7))) {
warnings.add("Event time is more than 7 days in the past");
}
}
// Data validation
if (cloudEvent.getData() != null && cloudEvent.getDataContentType() == null) {
warnings.add("Data content type should be specified when data is present");
}
boolean isValid = errors.isEmpty();
String message = isValid ? "Validation successful" : "Validation failed with " + errors.size() + " errors";
return ValidationResult.builder()
.valid(isValid)
.message(message)
.errors(errors)
.warnings(warnings)
.build();
}
public boolean isValid(CloudEvent cloudEvent) {
return validate(cloudEvent).isValid();
}
}
4. HTTP Transport Implementation
package com.example.cloudevents.transport.http;
import io.cloudevents.CloudEvent;
import io.cloudevents.http.vertx.VertxMessageFactory;
import io.cloudevents.core.message.MessageWriter;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.ext.web.client.HttpRequest;
import io.vertx.ext.web.client.WebClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
public class CloudEventHttpClient {
private static final Logger LOG = LoggerFactory.getLogger(CloudEventHttpClient.class);
private final WebClient webClient;
private final String baseUrl;
public CloudEventHttpClient(Vertx vertx, String baseUrl) {
this.webClient = WebClient.create(vertx);
this.baseUrl = baseUrl;
}
public CompletableFuture<HttpResponse> sendEvent(CloudEvent cloudEvent, String path) {
return sendEvent(cloudEvent, path, Map.of());
}
public CompletableFuture<HttpResponse> sendEvent(CloudEvent cloudEvent, String path, 
Map<String, String> headers) {
CompletableFuture<HttpResponse> future = new CompletableFuture<>();
try {
HttpRequest<Buffer> request = webClient.postAbs(baseUrl + path);
// Add custom headers
headers.forEach(request::putHeader);
// Write CloudEvent to HTTP request
VertxMessageFactory.createWriter(request)
.writeBinary(cloudEvent)
.onSuccess(response -> {
LOG.debug("CloudEvent sent successfully to {}: {}", path, response.statusCode());
future.complete(new HttpResponse(response.statusCode(), response.bodyAsString()));
})
.onFailure(error -> {
LOG.error("Failed to send CloudEvent to {}: {}", path, error.getMessage());
future.completeExceptionally(new CloudEventTransportException(
"Failed to send CloudEvent", error));
});
} catch (Exception e) {
LOG.error("Error preparing CloudEvent HTTP request: {}", e.getMessage());
future.completeExceptionally(e);
}
return future;
}
public CloudEvent receiveEvent(io.vertx.core.http.HttpServerRequest request) {
try {
return VertxMessageFactory.createReader(request)
.toEvent()
.orElseThrow(() -> new CloudEventTransportException("Failed to read CloudEvent from request"));
} catch (Exception e) {
LOG.error("Failed to receive CloudEvent: {}", e.getMessage());
throw new CloudEventTransportException("Failed to receive CloudEvent", e);
}
}
}
class CloudEventTransportException extends RuntimeException {
public CloudEventTransportException(String message) {
super(message);
}
public CloudEventTransportException(String message, Throwable cause) {
super(message, cause);
}
}
class HttpResponse {
private final int statusCode;
private final String body;
public HttpResponse(int statusCode, String body) {
this.statusCode = statusCode;
this.body = body;
}
public int getStatusCode() { return statusCode; }
public String getBody() { return body; }
public boolean isSuccess() {
return statusCode >= 200 && statusCode < 300;
}
}
5. Kafka Transport Implementation
package com.example.cloudevents.transport.kafka;
import io.cloudevents.CloudEvent;
import io.cloudevents.kafka.CloudEventSerializer;
import io.cloudevents.kafka.CloudEventDeserializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Future;
public class CloudEventKafkaTransport {
private static final Logger LOG = LoggerFactory.getLogger(CloudEventKafkaTransport.class);
private final KafkaProducer<String, CloudEvent> producer;
private final KafkaConsumer<String, CloudEvent> consumer;
private final String topic;
public CloudEventKafkaTransport(String bootstrapServers, String topic) {
this.topic = topic;
// Producer configuration
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", bootstrapServers);
producerProps.put("key.serializer", StringSerializer.class.getName());
producerProps.put("value.serializer", CloudEventSerializer.class.getName());
producerProps.put("acks", "all");
producerProps.put("retries", 3);
this.producer = new KafkaProducer<>(producerProps);
// Consumer configuration
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", bootstrapServers);
consumerProps.put("group.id", "cloudevents-consumer");
consumerProps.put("key.deserializer", StringDeserializer.class.getName());
consumerProps.put("value.deserializer", CloudEventDeserializer.class.getName());
consumerProps.put("auto.offset.reset", "earliest");
consumerProps.put("enable.auto.commit", "false");
this.consumer = new KafkaConsumer<>(consumerProps);
}
public Future<Void> sendEvent(CloudEvent cloudEvent) {
return sendEvent(cloudEvent, null);
}
public Future<Void> sendEvent(CloudEvent cloudEvent, String key) {
CompletableFuture<Void> future = new CompletableFuture<>();
try {
ProducerRecord<String, CloudEvent> record = new ProducerRecord<>(
topic, key, cloudEvent
);
producer.send(record, (metadata, exception) -> {
if (exception != null) {
LOG.error("Failed to send CloudEvent to Kafka: {}", exception.getMessage());
future.completeExceptionally(
new CloudEventTransportException("Kafka send failed", exception));
} else {
LOG.debug("CloudEvent sent to Kafka topic {} [partition: {}, offset: {}]", 
metadata.topic(), metadata.partition(), metadata.offset());
future.complete(null);
}
});
} catch (Exception e) {
LOG.error("Error sending CloudEvent to Kafka: {}", e.getMessage());
future.completeExceptionally(e);
}
return future;
}
public void subscribe() {
consumer.subscribe(Collections.singletonList(topic));
LOG.info("Subscribed to Kafka topic: {}", topic);
}
public ConsumerRecords<String, CloudEvent> pollEvents(Duration timeout) {
return consumer.poll(timeout);
}
public void commitSync() {
consumer.commitSync();
}
public void close() {
if (producer != null) {
producer.close();
}
if (consumer != null) {
consumer.close();
}
LOG.info("Kafka transport closed");
}
}

Spring Boot Integration

1. Spring Boot Configuration
package com.example.cloudevents.config;
import com.example.cloudevents.core.CloudEventFactory;
import com.example.cloudevents.core.CloudEventValidator;
import com.example.cloudevents.transport.http.CloudEventHttpClient;
import com.example.cloudevents.transport.kafka.CloudEventKafkaTransport;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import io.vertx.core.Vertx;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
@Configuration
@EnableKafka
public class CloudEventsConfig {
@Value("${app.events.source:https://example.com/app}")
private String eventSource;
@Value("${app.kafka.bootstrap-servers:localhost:9092}")
private String kafkaBootstrapServers;
@Value("${app.kafka.topic:cloudevents}")
private String kafkaTopic;
@Value("${app.http.event-endpoint:http://localhost:8080/events}")
private String httpEventEndpoint;
@Bean
public ObjectMapper objectMapper() {
ObjectMapper mapper = new ObjectMapper();
mapper.registerModule(new JavaTimeModule());
return mapper;
}
@Bean
public CloudEventFactory cloudEventFactory(ObjectMapper objectMapper) {
return new CloudEventFactory(eventSource, objectMapper);
}
@Bean
public CloudEventValidator cloudEventValidator() {
return new CloudEventValidator();
}
@Bean
public Vertx vertx() {
return Vertx.vertx();
}
@Bean
public CloudEventHttpClient cloudEventHttpClient(Vertx vertx) {
return new CloudEventHttpClient(vertx, httpEventEndpoint);
}
@Bean
public CloudEventKafkaTransport cloudEventKafkaTransport() {
return new CloudEventKafkaTransport(kafkaBootstrapServers, kafkaTopic);
}
}
2. Application Properties
# application.yml
app:
events:
source: "https://user-service.example.com"
kafka:
bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
topic: "user-events"
consumer-group: "user-service-consumer"
http:
event-endpoint: ${EVENT_ENDPOINT:http://event-broker:8080/events}
spring:
kafka:
bootstrap-servers: ${app.kafka.bootstrap-servers}
properties:
security.protocol: ${KAFKA_SECURITY_PROTOCOL:PLAINTEXT}
consumer:
group-id: ${app.kafka.consumer-group}
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: io.cloudevents.kafka.CloudEventDeserializer
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: io.cloudevents.kafka.CloudEventSerializer
server:
port: 8080
logging:
level:
com.example.cloudevents: DEBUG
3. Spring REST Controller for CloudEvents
package com.example.cloudevents.controller;
import com.example.cloudevents.core.CloudEventFactory;
import com.example.cloudevents.core.CloudEventValidator;
import com.example.cloudevents.model.*;
import com.example.cloudevents.service.UserService;
import com.example.cloudevents.transport.http.CloudEventHttpClient;
import io.cloudevents.CloudEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.concurrent.CompletableFuture;
@RestController
@RequestMapping("/api/events")
public class CloudEventController {
private static final Logger LOG = LoggerFactory.getLogger(CloudEventController.class);
private final CloudEventFactory eventFactory;
private final CloudEventValidator eventValidator;
private final CloudEventHttpClient httpClient;
private final UserService userService;
public CloudEventController(CloudEventFactory eventFactory,
CloudEventValidator eventValidator,
CloudEventHttpClient httpClient,
UserService userService) {
this.eventFactory = eventFactory;
this.eventValidator = eventValidator;
this.httpClient = httpClient;
this.userService = userService;
}
@PostMapping("/receive")
public ResponseEntity<EventReceipt> receiveEvent(@RequestBody CloudEvent cloudEvent) {
LOG.info("Received CloudEvent: {} from {}", cloudEvent.getType(), cloudEvent.getSource());
// Validate the event
ValidationResult validation = eventValidator.validate(cloudEvent);
if (!validation.isValid()) {
LOG.warn("Invalid CloudEvent received: {}", validation.getMessage());
return ResponseEntity.badRequest()
.body(EventReceipt.failure("Invalid event: " + validation.getMessage()));
}
try {
// Process based on event type
String eventType = cloudEvent.getType();
switch (eventType) {
case "com.example.user.created":
UserCreatedData userCreated = eventFactory.extractData(cloudEvent, UserCreatedData.class);
userService.handleUserCreated(userCreated);
break;
case "com.example.user.updated":
UserUpdatedData userUpdated = eventFactory.extractData(cloudEvent, UserUpdatedData.class);
userService.handleUserUpdated(userUpdated);
break;
case "com.example.order.placed":
OrderPlacedData orderPlaced = eventFactory.extractData(cloudEvent, OrderPlacedData.class);
userService.handleOrderPlaced(orderPlaced);
break;
default:
LOG.warn("Unhandled event type: {}", eventType);
return ResponseEntity.status(HttpStatus.NOT_IMPLEMENTED)
.body(EventReceipt.failure("Unhandled event type: " + eventType));
}
LOG.info("Successfully processed CloudEvent: {}", cloudEvent.getId());
return ResponseEntity.ok(EventReceipt.success(cloudEvent.getId()));
} catch (Exception e) {
LOG.error("Failed to process CloudEvent {}: {}", cloudEvent.getId(), e.getMessage(), e);
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body(EventReceipt.failure("Processing failed: " + e.getMessage()));
}
}
@PostMapping("/users/{userId}/created")
public CompletableFuture<ResponseEntity<EventReceipt>> publishUserCreated(
@PathVariable String userId,
@RequestBody UserCreatedData userData) {
LOG.info("Publishing user created event for: {}", userId);
CloudEvent event = eventFactory.createUserCreatedEvent(userData);
return httpClient.sendEvent(event, "/api/events/receive")
.thenApply(response -> {
if (response.isSuccess()) {
return ResponseEntity.ok(EventReceipt.success(event.getId()));
} else {
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body(EventReceipt.failure("Failed to publish event: " + response.getBody()));
}
})
.exceptionally(throwable -> {
LOG.error("Failed to publish user created event: {}", throwable.getMessage());
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body(EventReceipt.failure("Publishing failed: " + throwable.getMessage()));
});
}
@GetMapping("/health")
public ResponseEntity<String> health() {
return ResponseEntity.ok("CloudEvents service is healthy");
}
}
class EventReceipt {
private final boolean success;
private final String eventId;
private final String message;
private final long timestamp;
private EventReceipt(boolean success, String eventId, String message) {
this.success = success;
this.eventId = eventId;
this.message = message;
this.timestamp = System.currentTimeMillis();
}
public static EventReceipt success(String eventId) {
return new EventReceipt(true, eventId, "Event processed successfully");
}
public static EventReceipt failure(String message) {
return new EventReceipt(false, null, message);
}
// Getters
public boolean isSuccess() { return success; }
public String getEventId() { return eventId; }
public String getMessage() { return message; }
public long getTimestamp() { return timestamp; }
}
4. Kafka Event Consumer
package com.example.cloudevents.consumer;
import com.example.cloudevents.core.CloudEventFactory;
import com.example.cloudevents.core.CloudEventValidator;
import com.example.cloudevents.model.*;
import com.example.cloudevents.service.UserService;
import io.cloudevents.CloudEvent;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class CloudEventKafkaConsumer {
private static final Logger LOG = LoggerFactory.getLogger(CloudEventKafkaConsumer.class);
private final CloudEventFactory eventFactory;
private final CloudEventValidator eventValidator;
private final UserService userService;
public CloudEventKafkaConsumer(CloudEventFactory eventFactory,
CloudEventValidator eventValidator,
UserService userService) {
this.eventFactory = eventFactory;
this.eventValidator = eventValidator;
this.userService = userService;
}
@KafkaListener(topics = "${app.kafka.topic:cloudevents}")
public void consumeCloudEvent(ConsumerRecord<String, CloudEvent> record, Acknowledgment ack) {
CloudEvent cloudEvent = record.value();
LOG.info("Received CloudEvent from Kafka: {} [key: {}, partition: {}, offset: {}]", 
cloudEvent.getType(), record.key(), record.partition(), record.offset());
try {
// Validate event
if (!eventValidator.isValid(cloudEvent)) {
LOG.warn("Invalid CloudEvent received from Kafka, skipping: {}", cloudEvent.getId());
ack.acknowledge();
return;
}
// Process event based on type
processCloudEvent(cloudEvent);
// Acknowledge message
ack.acknowledge();
LOG.debug("Successfully processed CloudEvent: {}", cloudEvent.getId());
} catch (Exception e) {
LOG.error("Failed to process CloudEvent from Kafka {}: {}", 
cloudEvent.getId(), e.getMessage(), e);
// In production, you might want to implement retry or dead-letter queue logic
}
}
private void processCloudEvent(CloudEvent cloudEvent) {
String eventType = cloudEvent.getType();
switch (eventType) {
case "com.example.user.created":
UserCreatedData userCreated = eventFactory.extractData(cloudEvent, UserCreatedData.class);
userService.handleUserCreated(userCreated);
break;
case "com.example.user.updated":
UserUpdatedData userUpdated = eventFactory.extractData(cloudEvent, UserUpdatedData.class);
userService.handleUserUpdated(userUpdated);
break;
case "com.example.order.placed":
OrderPlacedData orderPlaced = eventFactory.extractData(cloudEvent, OrderPlacedData.class);
userService.handleOrderPlaced(orderPlaced);
break;
case "com.example.payment.processed":
PaymentProcessedData paymentProcessed = eventFactory.extractData(cloudEvent, PaymentProcessedData.class);
userService.handlePaymentProcessed(paymentProcessed);
break;
default:
LOG.warn("Unhandled CloudEvent type: {}", eventType);
}
}
}
5. Service Layer for Event Processing
package com.example.cloudevents.service;
import com.example.cloudevents.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@Service
public class UserService {
private static final Logger LOG = LoggerFactory.getLogger(UserService.class);
public void handleUserCreated(UserCreatedData userData) {
LOG.info("Processing user creation: {} - {}", userData.getUserId(), userData.getEmail());
// Business logic for user creation
// This could involve:
// - Creating user in database
// - Sending welcome email
// - Setting up user preferences
// - Publishing integration events
try {
// Simulate processing
Thread.sleep(100);
LOG.info("Successfully processed user creation for: {}", userData.getUserId());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.error("User creation processing interrupted for: {}", userData.getUserId());
}
}
public void handleUserUpdated(UserUpdatedData userData) {
LOG.info("Processing user update: {}", userData.getUserId());
// Business logic for user updates
// - Update user in database
// - Sync with external systems
// - Update search indexes
LOG.info("Successfully processed user update for: {}", userData.getUserId());
}
public void handleOrderPlaced(OrderPlacedData orderData) {
LOG.info("Processing order placement: {} for user: {}", 
orderData.getOrderId(), orderData.getCustomerId());
// Business logic for order placement
// - Update user order history
// - Calculate loyalty points
// - Send order confirmation
LOG.info("Successfully processed order placement: {}", orderData.getOrderId());
}
public void handlePaymentProcessed(PaymentProcessedData paymentData) {
LOG.info("Processing payment: {} for order: {}", 
paymentData.getPaymentId(), paymentData.getOrderId());
// Business logic for payment processing
// - Update order status
// - Trigger shipping
// - Update user account balance
if (paymentData.getStatus() == PaymentStatus.SUCCEEDED) {
LOG.info("Payment succeeded: {}", paymentData.getPaymentId());
} else {
LOG.warn("Payment failed: {}", paymentData.getPaymentId());
}
}
}

Advanced Features

1. Event Correlation and Tracing
package com.example.cloudevents.correlation;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;
import java.util.UUID;
@Component
public class EventCorrelationService {
private static final String CORRELATION_ID_HEADER = "ce-correlationid";
private static final String TRACE_ID_HEADER = "traceparent";
public CloudEvent addCorrelationContext(CloudEvent cloudEvent) {
CloudEventBuilder builder = CloudEventBuilder.from(cloudEvent);
// Add correlation ID if not present
if (cloudEvent.getExtension(CORRELATION_ID_HEADER) == null) {
String correlationId = getOrGenerateCorrelationId();
builder.withExtension(CORRELATION_ID_HEADER, correlationId);
}
// Add trace context if available
String traceId = MDC.get("traceId");
if (traceId != null && cloudEvent.getExtension(TRACE_ID_HEADER) == null) {
builder.withExtension(TRACE_ID_HEADER, traceId);
}
return builder.build();
}
public void setupCorrelationContext(CloudEvent cloudEvent) {
// Extract correlation ID from event
String correlationId = (String) cloudEvent.getExtension(CORRELATION_ID_HEADER);
if (correlationId != null) {
MDC.put("correlationId", correlationId);
}
// Extract trace ID from event
String traceId = (String) cloudEvent.getExtension(TRACE_ID_HEADER);
if (traceId != null) {
MDC.put("traceId", traceId);
}
}
public void clearCorrelationContext() {
MDC.remove("correlationId");
MDC.remove("traceId");
}
private String getOrGenerateCorrelationId() {
String correlationId = MDC.get("correlationId");
if (correlationId == null) {
correlationId = UUID.randomUUID().toString();
MDC.put("correlationId", correlationId);
}
return correlationId;
}
}
2. Event Schema Validation
package com.example.cloudevents.validation;
import com.example.cloudevents.model.ValidationResult;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.networknt.schema.JsonSchema;
import com.networknt.schema.JsonSchemaFactory;
import com.networknt.schema.SpecVersion;
import com.networknt.schema.ValidationMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;
import java.io.InputStream;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@Component
public class EventSchemaValidator {
private static final Logger LOG = LoggerFactory.getLogger(EventSchemaValidator.class);
private final ObjectMapper objectMapper;
private final ConcurrentHashMap<String, JsonSchema> schemaCache;
private final JsonSchemaFactory schemaFactory;
public EventSchemaValidator(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
this.schemaCache = new ConcurrentHashMap<>();
this.schemaFactory = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V7);
}
public ValidationResult validateAgainstSchema(String eventType, Object eventData) {
try {
JsonSchema schema = getSchemaForEventType(eventType);
if (schema == null) {
LOG.warn("No schema found for event type: {}", eventType);
return ValidationResult.valid(); // No schema means no validation
}
JsonNode dataNode = objectMapper.valueToTree(eventData);
Set<ValidationMessage> validationMessages = schema.validate(dataNode);
if (validationMessages.isEmpty()) {
return ValidationResult.valid();
} else {
return ValidationResult.invalid(
validationMessages.stream()
.map(ValidationMessage::getMessage)
.toList()
);
}
} catch (Exception e) {
LOG.error("Schema validation failed for event type {}: {}", eventType, e.getMessage());
return ValidationResult.invalid("Schema validation error: " + e.getMessage());
}
}
private JsonSchema getSchemaForEventType(String eventType) {
return schemaCache.computeIfAbsent(eventType, this::loadSchema);
}
private JsonSchema loadSchema(String eventType) {
try {
String schemaPath = String.format("schemas/%s.schema.json", 
eventType.replace('.', '/'));
ClassPathResource resource = new ClassPathResource(schemaPath);
if (!resource.exists()) {
LOG.debug("No schema file found at: {}", schemaPath);
return null;
}
try (InputStream inputStream = resource.getInputStream()) {
return schemaFactory.getSchema(inputStream);
}
} catch (Exception e) {
LOG.error("Failed to load schema for event type {}: {}", eventType, e.getMessage());
return null;
}
}
}
3. Event Retry and Dead Letter Queue
package com.example.cloudevents.retry;
import io.cloudevents.CloudEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.time.OffsetDateTime;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@Component
public class EventRetryHandler {
private static final Logger LOG = LoggerFactory.getLogger(EventRetryHandler.class);
private final KafkaTemplate<String, CloudEvent> kafkaTemplate;
private final ConcurrentHashMap<String, AtomicInteger> retryCounts;
@Value("${app.kafka.dlq-topic:cloudevents-dlq}")
private String dlqTopic;
@Value("${app.kafka.retry-topic:cloudevents-retry}")
private String retryTopic;
@Value("${app.events.max-retries:3}")
private int maxRetries;
public EventRetryHandler(KafkaTemplate<String, CloudEvent> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
this.retryCounts = new ConcurrentHashMap<>();
}
public boolean shouldRetry(CloudEvent cloudEvent, Exception error) {
String eventId = cloudEvent.getId();
int retryCount = retryCounts.computeIfAbsent(eventId, k -> new AtomicInteger(0)).get();
if (retryCount >= maxRetries) {
LOG.warn("Event {} reached maximum retry count ({}), sending to DLQ", eventId, maxRetries);
sendToDlq(cloudEvent, error);
retryCounts.remove(eventId);
return false;
}
return true;
}
public void scheduleRetry(CloudEvent cloudEvent) {
String eventId = cloudEvent.getId();
int retryCount = retryCounts.get(eventId).incrementAndGet();
LOG.info("Scheduling retry {} for event: {}", retryCount, eventId);
// Add retry delay based on retry count (exponential backoff)
long delayMs = calculateRetryDelay(retryCount);
// In production, you might use a delayed message queue or scheduler
// For simplicity, we're sending to a retry topic
kafkaTemplate.send(retryTopic, eventId, cloudEvent);
}
public void sendToDlq(CloudEvent cloudEvent, Exception error) {
// Add error information to the event
CloudEvent dlqEvent = io.cloudevents.core.builder.CloudEventBuilder.from(cloudEvent)
.withExtension("dlq-reason", error.getMessage())
.withExtension("dlq-timestamp", OffsetDateTime.now().toString())
.build();
kafkaTemplate.send(dlqTopic, cloudEvent.getId(), dlqEvent);
LOG.info("Event {} sent to DLQ: {}", cloudEvent.getId(), error.getMessage());
}
private long calculateRetryDelay(int retryCount) {
// Exponential backoff: 1s, 2s, 4s, 8s, etc.
return (long) Math.pow(2, retryCount - 1) * 1000;
}
public void clearRetryCount(String eventId) {
retryCounts.remove(eventId);
}
}

Testing

1. Unit Tests
package com.example.cloudevents.core;
import com.example.cloudevents.model.UserCreatedData;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.cloudevents.CloudEvent;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
class CloudEventFactoryTest {
private CloudEventFactory eventFactory;
private ObjectMapper objectMapper;
@BeforeEach
void setUp() {
objectMapper = new ObjectMapper();
eventFactory = new CloudEventFactory("https://test.example.com", objectMapper);
}
@Test
void testCreateEventWithData() {
// Given
UserCreatedData userData = new UserCreatedData("user123", "[email protected]", "John", "Doe");
// When
CloudEvent event = eventFactory.createUserCreatedEvent(userData);
// Then
assertNotNull(event);
assertEquals("com.example.user.created", event.getType());
assertEquals("https://test.example.com", event.getSource().toString());
assertNotNull(event.getId());
assertNotNull(event.getTime());
assertEquals("user123", event.getSubject());
}
@Test
void testExtractDataFromEvent() {
// Given
UserCreatedData originalData = new UserCreatedData("user123", "[email protected]", "John", "Doe");
CloudEvent event = eventFactory.createUserCreatedEvent(originalData);
// When
UserCreatedData extractedData = eventFactory.extractData(event, UserCreatedData.class);
// Then
assertNotNull(extractedData);
assertEquals(originalData.getUserId(), extractedData.getUserId());
assertEquals(originalData.getEmail(), extractedData.getEmail());
assertEquals(originalData.getFirstName(), extractedData.getFirstName());
assertEquals(originalData.getLastName(), extractedData.getLastName());
}
@Test
void testEventValidation() {
// Given
CloudEventValidator validator = new CloudEventValidator();
UserCreatedData userData = new UserCreatedData("user123", "[email protected]", "John", "Doe");
CloudEvent event = eventFactory.createUserCreatedEvent(userData);
// When
ValidationResult result = validator.validate(event);
// Then
assertTrue(result.isValid());
assertTrue(result.getErrors().isEmpty());
}
}
2. Integration Tests
package com.example.cloudevents.controller;
import com.example.cloudevents.core.CloudEventFactory;
import com.example.cloudevents.model.UserCreatedData;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.MediaType;
import org.springframework.test.web.servlet.MockMvc;
import java.net.URI;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
@SpringBootTest
@AutoConfigureMockMvc
class CloudEventControllerIntegrationTest {
@Autowired
private MockMvc mockMvc;
@Autowired
private ObjectMapper objectMapper;
@Autowired
private CloudEventFactory eventFactory;
@Test
void testReceiveValidCloudEvent() throws Exception {
// Given
UserCreatedData userData = new UserCreatedData("test-user", "[email protected]", "Test", "User");
CloudEvent cloudEvent = eventFactory.createUserCreatedEvent(userData);
// When & Then
mockMvc.perform(post("/api/events/receive")
.contentType(MediaType.APPLICATION_JSON)
.content(objectMapper.writeValueAsString(cloudEvent)))
.andExpect(status().isOk())
.andExpect(jsonPath("$.success").value(true))
.andExpect(jsonPath("$.eventId").exists());
}
@Test
void testReceiveInvalidCloudEvent() throws Exception {
// Given - Create an invalid CloudEvent (missing required fields)
CloudEvent invalidEvent = CloudEventBuilder.v1()
.withId("test-id")
.withSource(URI.create("https://test.example.com"))
// Missing type - should cause validation to fail
.build();
// When & Then
mockMvc.perform(post("/api/events/receive")
.contentType(MediaType.APPLICATION_JSON)
.content(objectMapper.writeValueAsString(invalidEvent)))
.andExpect(status().isBadRequest())
.andExpect(jsonPath("$.success").value(false));
}
}

Best Practices

1. Configuration for Production
# application-production.yml
app:
events:
source: "https://user-service.production.example.com"
kafka:
bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS}
topic: "production-cloudevents"
consumer-group: "user-service-production"
dlq-topic: "production-cloudevents-dlq"
retry-topic: "production-cloudevents-retry"
http:
event-endpoint: "https://event-broker.production.example.com/events"
events:
max-retries: 5
retry-backoff-multiplier: 2
spring:
kafka:
bootstrap-servers: ${app.kafka.bootstrap-servers}
properties:
security.protocol: SSL
ssl.truststore.location: ${KAFKA_TRUSTSTORE_LOCATION}
ssl.truststore.password: ${KAFKA_TRUSTSTORE_PASSWORD}
ssl.keystore.location: ${KAFKA_KEYSTORE_LOCATION}
ssl.keystore.password: ${KAFKA_KEYSTORE_PASSWORD}
consumer:
group-id: ${app.kafka.consumer-group}
auto-offset-reset: latest
enable-auto-commit: false
producer:
acks: all
retries: 3
batch-size: 16384
linger-ms: 1
buffer-memory: 33554432
management:
endpoints:
web:
exposure:
include: health,metrics,info
endpoint:
health:
show-details: always
2. Monitoring and Metrics
package com.example.cloudevents.metrics;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
@Component
public class EventMetrics {
private final Counter eventsReceivedCounter;
private final Counter eventsProcessedCounter;
private final Counter eventsFailedCounter;
private final Timer eventProcessingTimer;
public EventMetrics(MeterRegistry meterRegistry) {
this.eventsReceivedCounter = Counter.builder("cloudevents.received")
.description("Number of CloudEvents received")
.tag("transport", "http") // or "kafka"
.register(meterRegistry);
this.eventsProcessedCounter = Counter.builder("cloudevents.processed")
.description("Number of CloudEvents successfully processed")
.register(meterRegistry);
this.eventsFailedCounter = Counter.builder("cloudevents.failed")
.description("Number of CloudEvents that failed processing")
.register(meterRegistry);
this.eventProcessingTimer = Timer.builder("cloudevents.processing.time")
.description("Time taken to process CloudEvents")
.register(meterRegistry);
}
public void recordEventReceived(String transport) {
eventsReceivedCounter.increment();
}
public void recordEventProcessed() {
eventsProcessedCounter.increment();
}
public void recordEventFailed() {
eventsFailedCounter.increment();
}
public void recordProcessingTime(long duration, TimeUnit unit) {
eventProcessingTimer.record(duration, unit);
}
}

Conclusion

Implementing CloudEvents SDK in Java provides:

  • Standardized event format across different systems and platforms
  • Protocol interoperability (HTTP, Kafka, AMQP, etc.)
  • Type-safe event handling with Java models
  • Comprehensive validation and error handling
  • Production-ready features like retry mechanisms and dead-letter queues

Key benefits demonstrated:

  1. Event-driven architecture with CloudEvents specification
  2. Multiple transport support (HTTP, Kafka)
  3. Spring Boot integration for easy development
  4. Schema validation for data consistency
  5. Correlation and tracing for distributed systems
  6. Error handling with retry and DLQ patterns
  7. Monitoring and metrics for observability

This implementation enables building robust, scalable event-driven systems that can communicate seamlessly across different platforms and technologies while maintaining strong typing and comprehensive error handling.

Category : JAVA

Leave a Reply

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


Macro Nepal Helper