MACRO NEPAL
Event-Driven Java: Building Knative Eventing Sources and Sinks
awiskar acharya
November 19, 2025
11:13 am

Article

Knative Eventing is a Kubernetes-based platform that provides tools for building event-driven applications. Eventing Sources are the entry points that generate or capture events from external systems and deliver them to the Knative eventing mesh. For Java developers, creating custom event sources enables integration with various systems while leveraging Knative's powerful event routing capabilities.

In this guide, we'll explore how to build, deploy, and manage Knative Eventing Sources using Java, creating robust event-driven architectures on Kubernetes.

Why Knative Eventing Sources for Java?

  • Cloud Events Standard: Native support for CNCF CloudEvents specification
  • Multi-Protocol Support: HTTP, gRPC, WebSocket, and custom protocols
  • Scalability: Automatic scaling based on event load
  • Portability: Run on any Kubernetes cluster with Knative
  • Rich Ecosystem: Integrate with existing event sources and brokers
  • Developer Productivity: Focus on business logic rather than infrastructure

Part 1: Knative Eventing Fundamentals

1.1 Core Concepts

  • Event Sources: Producers that generate events (Kafka, GitHub, Cron, Custom)
  • Brokers: Event mesh backbone for routing and delivery
  • Triggers: Subscription rules that route events from brokers to sinks
  • Sinks: Event consumers (Services, Channels, Brokers)
  • CloudEvents: Standard event format for interoperability

1.2 Project Structure

knative-java-eventing/
├── src/
│   ├── main/
│   │   ├── java/com/example/events/
│   │   │   ├── source/
│   │   │   ├── sink/
│   │   │   └── model/
│   │   └── resources/
│   │       └── k8s/
│   └── test/
│       └── java/com/example/events/
├── Dockerfile
├── knative/
│   ├── service.yaml
│   ├── trigger.yaml
│   └── source.yaml
└── pom.xml

Part 2: Dependencies and Setup

2.1 Maven Dependencies

<!-- pom.xml -->
<properties>
<knative.version>1.11.0</knative.version>
<cloudevents.version>3.0.0</cloudevents.version>
<spring-boot.version>3.1.0</spring-boot.version>
<kubernetes-client.version>6.7.2</kubernetes-client.version>
</properties>
<dependencies>
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>${spring-boot.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<version>${spring-boot.version}</version>
</dependency>
<!-- CloudEvents SDK -->
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-core</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-http-basic</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-json-jackson</artifactId>
<version>${cloudevents.version}</version>
</dependency>
<!-- Knative -->
<dependency>
<groupId>io.kubernetes</groupId>
<artifactId>client-java</artifactId>
<version>${kubernetes-client.version}</version>
</dependency>
<dependency>
<groupId>dev.knative</groupId>
<artifactId>eventing</artifactId>
<version>${knative.version}</version>
</dependency>
<!-- Utilities -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.15.2</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.4.0</version>
</dependency>
<!-- Testing -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>${spring-boot.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.cloudevents</groupId>
<artifactId>cloudevents-test</artifactId>
<version>${cloudevents.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

2.2 CloudEvents Model

// File: src/main/java/com/example/events/model/CloudEventData.java
package com.example.events.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import java.time.Instant;
import java.util.Map;
@JsonInclude(JsonInclude.Include.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown = true)
public class CloudEventData {
private String eventId;
private String eventType;
private String eventSource;
private Instant eventTime;
private String subject;
private Map<String, Object> data;
private Map<String, String> extensions;
// Constructors
public CloudEventData() {}
public CloudEventData(String eventId, String eventType, String eventSource, 
Instant eventTime, String subject, Map<String, Object> data) {
this.eventId = eventId;
this.eventType = eventType;
this.eventSource = eventSource;
this.eventTime = eventTime;
this.subject = subject;
this.data = data;
}
// Getters and Setters
public String getEventId() { return eventId; }
public void setEventId(String eventId) { this.eventId = eventId; }
public String getEventType() { return eventType; }
public void setEventType(String eventType) { this.eventType = eventType; }
public String getEventSource() { return eventSource; }
public void setEventSource(String eventSource) { this.eventSource = eventSource; }
public Instant getEventTime() { return eventTime; }
public void setEventTime(Instant eventTime) { this.eventTime = eventTime; }
public String getSubject() { return subject; }
public void setSubject(String subject) { this.subject = subject; }
public Map<String, Object> getData() { return data; }
public void setData(Map<String, Object> data) { this.data = data; }
public Map<String, String> getExtensions() { return extensions; }
public void setExtensions(Map<String, String> extensions) { this.extensions = extensions; }
}

Part 3: Custom Event Source Implementation

3.1 HTTP-Based Event Source

// File: src/main/java/com/example/events/source/HttpEventSourceController.java
package com.example.events.source;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.provider.EventFormatProvider;
import io.cloudevents.http.HttpMessageFactory;
import io.cloudevents.jackson.JsonFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.net.URI;
import java.time.OffsetDateTime;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@RestController
@RequestMapping("/api/v1/events")
public class HttpEventSourceController {
private static final Logger logger = LoggerFactory.getLogger(HttpEventSourceController.class);
private final EventSenderService eventSenderService;
private final CopyOnWriteArrayList<SseEmitter> sseEmitters = new CopyOnWriteArrayList<>();
private final ConcurrentHashMap<String, CloudEvent> pendingEvents = new ConcurrentHashMap<>();
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
public HttpEventSourceController(EventSenderService eventSenderService) {
this.eventSenderService = eventSenderService;
startEventGenerator();
}
@PostMapping("/ingest")
public ResponseEntity<String> ingestEvent(
@RequestBody String eventData,
@RequestHeader HttpHeaders headers,
HttpServletRequest request,
HttpServletResponse response) {
try {
// Parse CloudEvent from HTTP request
CloudEvent cloudEvent = HttpMessageFactory
.createReaderFromRequest(request, eventData.getBytes())
.toEvent();
logger.info("Received CloudEvent: id={}, type={}, source={}", 
cloudEvent.getId(), cloudEvent.getType(), cloudEvent.getSource());
// Process and forward the event
processIncomingEvent(cloudEvent);
return ResponseEntity.accepted().body("Event processed successfully");
} catch (Exception e) {
logger.error("Failed to process incoming event", e);
return ResponseEntity.badRequest().body("Invalid CloudEvent format");
}
}
@GetMapping("/stream")
public SseEmitter streamEvents() {
SseEmitter emitter = new SseEmitter(30_000L); // 30 seconds timeout
sseEmitters.add(emitter);
emitter.onCompletion(() -> {
sseEmitters.remove(emitter);
logger.info("SSE connection completed");
});
emitter.onTimeout(() -> {
sseEmitters.remove(emitter);
logger.info("SSE connection timed out");
});
emitter.onError((ex) -> {
sseEmitters.remove(emitter);
logger.error("SSE connection error", ex);
});
// Send initial event
sendSseEvent(emitter, createHeartbeatEvent());
return emitter;
}
@PostMapping("/custom")
public ResponseEntity<CloudEvent> createCustomEvent(@RequestBody CustomEventRequest request) {
try {
CloudEvent cloudEvent = CloudEventBuilder.v1()
.withId(UUID.randomUUID().toString())
.withType(request.getEventType())
.withSource(URI.create(request.getEventSource()))
.withTime(OffsetDateTime.now())
.withDataContentType("application/json")
.withData(request.getData().getBytes())
.withExtension("customtype", request.getCustomType())
.build();
// Send to Knative broker
eventSenderService.sendToBroker(cloudEvent);
logger.info("Created and sent custom event: {}", cloudEvent.getId());
return ResponseEntity.accepted().body(cloudEvent);
} catch (Exception e) {
logger.error("Failed to create custom event", e);
return ResponseEntity.badRequest().build();
}
}
private void processIncomingEvent(CloudEvent cloudEvent) {
// Store event temporarily
pendingEvents.put(cloudEvent.getId(), cloudEvent);
// Send to Knative broker
eventSenderService.sendToBroker(cloudEvent);
// Broadcast to SSE clients
broadcastToSseClients(cloudEvent);
// Remove from pending after processing
scheduler.schedule(() -> {
pendingEvents.remove(cloudEvent.getId());
}, 30, TimeUnit.SECONDS);
}
private void broadcastToSseClients(CloudEvent cloudEvent) {
for (SseEmitter emitter : sseEmitters) {
try {
sendSseEvent(emitter, cloudEvent);
} catch (Exception e) {
logger.warn("Failed to send event to SSE client", e);
sseEmitters.remove(emitter);
}
}
}
private void sendSseEvent(SseEmitter emitter, CloudEvent event) {
try {
String eventJson = new String(EventFormatProvider.getInstance()
.resolveFormat(JsonFormat.CONTENT_TYPE)
.serialize(event));
emitter.send(SseEmitter.event()
.id(event.getId())
.name(event.getType())
.data(eventJson));
} catch (Exception e) {
logger.error("Failed to send SSE event", e);
}
}
private CloudEvent createHeartbeatEvent() {
return CloudEventBuilder.v1()
.withId(UUID.randomUUID().toString())
.withType("com.example.heartbeat")
.withSource(URI.create("https://events.example.com/source"))
.withTime(OffsetDateTime.now())
.withDataContentType("application/json")
.withData("{\"status\": \"alive\", \"timestamp\": \"" + OffsetDateTime.now() + "\"}")
.build();
}
private void startEventGenerator() {
// Generate periodic events for demonstration
scheduler.scheduleAtFixedRate(() -> {
try {
CloudEvent heartbeat = createHeartbeatEvent();
eventSenderService.sendToBroker(heartbeat);
broadcastToSseClients(heartbeat);
} catch (Exception e) {
logger.error("Error generating heartbeat event", e);
}
}, 60, 60, TimeUnit.SECONDS);
}
}
// File: src/main/java/com/example/events/source/CustomEventRequest.java
package com.example.events.source;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class CustomEventRequest {
private String eventType;
private String eventSource;
private String customType;
private String data;
// Getters and Setters
public String getEventType() { return eventType; }
public void setEventType(String eventType) { this.eventType = eventType; }
public String getEventSource() { return eventSource; }
public void setEventSource(String eventSource) { this.eventSource = eventSource; }
public String getCustomType() { return customType; }
public void setCustomType(String customType) { this.customType = customType; }
public String getData() { return data; }
public void setData(String data) { this.data = data; }
}

3.2 Event Sender Service

// File: src/main/java/com/example/events/source/EventSenderService.java
package com.example.events.source;
import io.cloudevents.CloudEvent;
import io.cloudevents.http.HttpMessageFactory;
import io.cloudevents.jackson.JsonFormat;
import io.cloudevents.core.provider.EventFormatProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import java.net.URI;
import java.util.Collections;
@Service
public class EventSenderService {
private static final Logger logger = LoggerFactory.getLogger(EventSenderService.class);
private final RestTemplate restTemplate;
@Value("${knative.broker.url:http://broker-knative-broker.default.svc.cluster.local}")
private String brokerUrl;
@Value("${knative.broker.enabled:true}")
private boolean brokerEnabled;
public EventSenderService(RestTemplate restTemplate) {
this.restTemplate = restTemplate;
}
public boolean sendToBroker(CloudEvent cloudEvent) {
if (!brokerEnabled) {
logger.debug("Broker disabled, skipping event send: {}", cloudEvent.getId());
return true;
}
try {
// Convert CloudEvent to HTTP request
HttpHeaders headers = new HttpHeaders();
headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON);
headers.setAccept(Collections.singletonList(org.springframework.http.MediaType.APPLICATION_JSON));
byte[] eventBytes = EventFormatProvider.getInstance()
.resolveFormat(JsonFormat.CONTENT_TYPE)
.serialize(cloudEvent);
HttpEntity<byte[]> requestEntity = new HttpEntity<>(eventBytes, headers);
ResponseEntity<String> response = restTemplate.exchange(
brokerUrl,
HttpMethod.POST,
requestEntity,
String.class
);
if (response.getStatusCode().is2xxSuccessful()) {
logger.debug("Successfully sent event to broker: {}", cloudEvent.getId());
return true;
} else {
logger.error("Failed to send event to broker. Status: {}, Response: {}", 
response.getStatusCode(), response.getBody());
return false;
}
} catch (Exception e) {
logger.error("Error sending event to broker: {}", cloudEvent.getId(), e);
return false;
}
}
public boolean sendToSink(CloudEvent cloudEvent, String sinkUrl) {
try {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON);
byte[] eventBytes = EventFormatProvider.getInstance()
.resolveFormat(JsonFormat.CONTENT_TYPE)
.serialize(cloudEvent);
HttpEntity<byte[]> requestEntity = new HttpEntity<>(eventBytes, headers);
ResponseEntity<String> response = restTemplate.exchange(
sinkUrl,
HttpMethod.POST,
requestEntity,
String.class
);
if (response.getStatusCode().is2xxSuccessful()) {
logger.debug("Successfully sent event to sink: {}", sinkUrl);
return true;
} else {
logger.warn("Failed to send event to sink. Status: {}, Response: {}", 
response.getStatusCode(), response.getBody());
return false;
}
} catch (Exception e) {
logger.error("Error sending event to sink: {}", sinkUrl, e);
return false;
}
}
}

Part 4: Kafka Event Source

4.1 Kafka to Knative Bridge

// File: src/main/java/com/example/events/source/KafkaEventSource.java
package com.example.events.source;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.URI;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.Collections;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
@Component
public class KafkaEventSource {
private static final Logger logger = LoggerFactory.getLogger(KafkaEventSource.class);
private final EventSenderService eventSenderService;
private final AtomicBoolean running = new AtomicBoolean(false);
private ExecutorService executorService;
private KafkaConsumer<String, String> kafkaConsumer;
@Value("${kafka.bootstrap.servers:localhost:9092}")
private String bootstrapServers;
@Value("${kafka.topic:knative-events}")
private String topic;
@Value("${kafka.group.id:knative-event-source}")
private String groupId;
@Value("${kafka.auto.offset.reset:earliest}")
private String autoOffsetReset;
public KafkaEventSource(EventSenderService eventSenderService) {
this.eventSenderService = eventSenderService;
}
@PostConstruct
public void initialize() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
kafkaConsumer = new KafkaConsumer<>(props);
kafkaConsumer.subscribe(Collections.singletonList(topic));
startConsumer();
}
@PreDestroy
public void shutdown() {
running.set(false);
if (executorService != null) {
executorService.shutdown();
}
if (kafkaConsumer != null) {
kafkaConsumer.close();
}
}
private void startConsumer() {
running.set(true);
executorService = Executors.newSingleThreadExecutor();
executorService.submit(() -> {
logger.info("Starting Kafka consumer for topic: {}", topic);
while (running.get()) {
try {
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
processKafkaRecord(record);
}
} catch (Exception e) {
logger.error("Error in Kafka consumer", e);
// Add backoff and retry logic
try {
Thread.sleep(5000);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
logger.info("Kafka consumer stopped");
});
}
private void processKafkaRecord(ConsumerRecord<String, String> record) {
try {
CloudEvent cloudEvent = convertToCloudEvent(record);
boolean sent = eventSenderService.sendToBroker(cloudEvent);
if (sent) {
logger.debug("Successfully processed Kafka record: topic={}, partition={}, offset={}", 
record.topic(), record.partition(), record.offset());
} else {
logger.warn("Failed to send event from Kafka record: topic={}, offset={}", 
record.topic(), record.offset());
}
} catch (Exception e) {
logger.error("Failed to process Kafka record: topic={}, offset={}", 
record.topic(), record.offset(), e);
}
}
private CloudEvent convertToCloudEvent(ConsumerRecord<String, String> record) {
return CloudEventBuilder.v1()
.withId(UUID.randomUUID().toString())
.withType("com.example.kafka." + record.topic())
.withSource(URI.create("https://kafka.example.com/" + record.topic()))
.withTime(OffsetDateTime.now())
.withDataContentType("application/json")
.withData(record.value().getBytes())
.withExtension("kafkatopic", record.topic())
.withExtension("kafkapartition", record.partition())
.withExtension("kafkaoffset", record.offset())
.withExtension("kafkakey", record.key() != null ? record.key() : "")
.build();
}
}

Part 5: Event Sink Implementation

5.1 CloudEvent Receiver

// File: src/main/java/com/example/events/sink/EventSinkController.java
package com.example.events.sink;
import io.cloudevents.CloudEvent;
import io.cloudevents.http.HttpMessageFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletRequest;
import java.util.concurrent.atomic.AtomicLong;
@RestController
@RequestMapping("/api/v1/sink")
public class EventSinkController {
private static final Logger logger = LoggerFactory.getLogger(EventSinkController.class);
private final AtomicLong eventCounter = new AtomicLong(0);
private final EventProcessorService eventProcessorService;
public EventSinkController(EventProcessorService eventProcessorService) {
this.eventProcessorService = eventProcessorService;
}
@PostMapping("/events")
public ResponseEntity<String> receiveEvent(
HttpServletRequest request,
@RequestBody(required = false) String body) {
long eventNumber = eventCounter.incrementAndGet();
try {
// Parse CloudEvent from HTTP request
CloudEvent cloudEvent = HttpMessageFactory
.createReaderFromRequest(request, body != null ? body.getBytes() : new byte[0])
.toEvent();
logger.info("Received CloudEvent #{}: id={}, type={}, source={}", 
eventNumber, cloudEvent.getId(), cloudEvent.getType(), cloudEvent.getSource());
// Process the event
eventProcessorService.processEvent(cloudEvent);
return ResponseEntity.accepted().body("Event processed successfully");
} catch (Exception e) {
logger.error("Failed to process CloudEvent #{}", eventNumber, e);
return ResponseEntity.badRequest().body("Invalid CloudEvent format");
}
}
@GetMapping("/metrics")
public EventMetrics getMetrics() {
return new EventMetrics(eventCounter.get());
}
}
// File: src/main/java/com/example/events/sink/EventProcessorService.java
package com.example.events.sink;
import io.cloudevents.CloudEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
@Service
public class EventProcessorService {
private static final Logger logger = LoggerFactory.getLogger(EventProcessorService.class);
private final ConcurrentHashMap<String, EventHandler> eventHandlers = new ConcurrentHashMap<>();
private final AtomicLong processedCount = new AtomicLong(0);
public EventProcessorService() {
// Register default event handlers
registerHandler("com.example.user.created", new UserEventHandler());
registerHandler("com.example.order.placed", new OrderEventHandler());
registerHandler("com.example.system.alert", new SystemEventHandler());
}
public void processEvent(CloudEvent cloudEvent) {
try {
String eventType = cloudEvent.getType();
EventHandler handler = eventHandlers.get(eventType);
if (handler != null) {
handler.handle(cloudEvent);
} else {
handleUnknownEvent(cloudEvent);
}
processedCount.incrementAndGet();
logger.debug("Processed event: {}", cloudEvent.getId());
} catch (Exception e) {
logger.error("Error processing event: {}", cloudEvent.getId(), e);
// Implement dead letter queue or retry logic
}
}
public void registerHandler(String eventType, EventHandler handler) {
eventHandlers.put(eventType, handler);
logger.info("Registered event handler for type: {}", eventType);
}
private void handleUnknownEvent(CloudEvent cloudEvent) {
logger.warn("No handler registered for event type: {}", cloudEvent.getType());
// Could store in dead letter queue or send to monitoring system
}
public long getProcessedCount() {
return processedCount.get();
}
}
// File: src/main/java/com/example/events/sink/EventHandler.java
package com.example.events.sink;
import io.cloudevents.CloudEvent;
public interface EventHandler {
void handle(CloudEvent cloudEvent);
}
// File: src/main/java/com/example/events/sink/UserEventHandler.java
package com.example.events.sink;
import io.cloudevents.CloudEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class UserEventHandler implements EventHandler {
private static final Logger logger = LoggerFactory.getLogger(UserEventHandler.class);
@Override
public void handle(CloudEvent cloudEvent) {
logger.info("Processing user event: {}", cloudEvent.getId());
// Extract and process user data
if (cloudEvent.getData() != null) {
String data = new String(cloudEvent.getData().toBytes());
logger.info("User event data: {}", data);
// Business logic for user events
// e.g., update user database, send welcome email, etc.
}
// Process extensions
cloudEvent.getExtensionNames().forEach(ext -> {
logger.debug("Extension {}: {}", ext, cloudEvent.getExtension(ext));
});
}
}
// File: src/main/java/com/example/events/sink/EventMetrics.java
package com.example.events.sink;
public class EventMetrics {
private final long totalEventsProcessed;
private final long timestamp;
public EventMetrics(long totalEventsProcessed) {
this.totalEventsProcessed = totalEventsProcessed;
this.timestamp = System.currentTimeMillis();
}
// Getters
public long getTotalEventsProcessed() { return totalEventsProcessed; }
public long getTimestamp() { return timestamp; }
}

Part 6: Knative Configuration

6.1 Knative Service Definition

# knative/service.yaml
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
name: java-event-source
namespace: default
spec:
template:
metadata:
annotations:
autoscaling.knative.dev/minScale: "1"
autoscaling.knative.dev/maxScale: "10"
autoscaling.knative.dev/target: "100"
spec:
containers:
- image: my-registry/java-event-source:latest
env:
- name: KAFKA_BOOTSTRAP_SERVERS
value: "kafka-broker:9092"
- name: KAFKA_TOPIC
value: "knative-events"
- name: KNAITIVE_BROKER_URL
value: "http://broker-knative-broker.default.svc.cluster.local"
- name: JAVA_OPTS
value: "-Xmx512m -Xms256m"
resources:
requests:
memory: "512Mi"
cpu: "250m"
limits:
memory: "1Gi"
cpu: "500m"
livenessProbe:
httpGet:
path: /actuator/health
port: 8080
initialDelaySeconds: 20
periodSeconds: 10
readinessProbe:
httpGet:
path: /actuator/health
port: 8080
initialDelaySeconds: 5
periodSeconds: 5

6.2 Knative Trigger Configuration

# knative/trigger.yaml
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
name: java-user-events-trigger
namespace: default
spec:
broker: default
filter:
attributes:
type: "com.example.user.created"
subscriber:
ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: user-service
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
name: java-order-events-trigger
namespace: default
spec:
broker: default
filter:
attributes:
type: "com.example.order.placed"
subscriber:
ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: order-service
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
name: java-all-events-trigger
namespace: default
spec:
broker: default
filter:
attributes:
type: "com.example.system.alert"
subscriber:
uri: http://monitoring-service.default.svc.cluster.local/api/alerts

6.3 Kafka Source Configuration

# knative/kafka-source.yaml
apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
name: kafka-java-events-source
namespace: default
spec:
consumerGroup: knative-java-consumer
bootstrapServers:
- kafka-broker:9092
topics:
- knative-events
- user-events
- order-events
sink:
ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: java-event-processor

Part 7: Spring Boot Application

7.1 Main Application Class

// File: src/main/java/com/example/EventSourceApplication.java
package com.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;
@SpringBootApplication
public class EventSourceApplication {
public static void main(String[] args) {
SpringApplication.run(EventSourceApplication.class, args);
}
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
@Bean
public io.cloudevents.CloudEventFormat cloudEventFormat() {
return io.cloudevents.core.provider.EventFormatProvider.getInstance()
.resolveFormat(io.cloudevents.jackson.JsonFormat.CONTENT_TYPE);
}
}
// File: src/main/resources/application.yaml
server:
port: 8080
spring:
application:
name: java-event-source
jackson:
date-format: com.fasterxml.jackson.databind.util.StdDateFormat
default-property-inclusion: NON_NULL
management:
endpoints:
web:
exposure:
include: health,info,metrics,configreload
endpoint:
health:
show-details: always
knative:
broker:
url: http://broker-knative-broker.default.svc.cluster.local
enabled: true
kafka:
bootstrap:
servers: localhost:9092
topic: knative-events
group:
id: knative-event-source-java
logging:
level:
com.example.events: DEBUG
org.apache.kafka: WARN

Part 8: Testing

8.1 Unit Tests

// File: src/test/java/com/example/events/source/EventSenderServiceTest.java
package com.example.events.source;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;
import java.net.URI;
import java.time.OffsetDateTime;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class EventSenderServiceTest {
@Mock
private RestTemplate restTemplate;
private EventSenderService eventSenderService;
private CloudEvent testEvent;
@BeforeEach
void setUp() {
eventSenderService = new EventSenderService(restTemplate);
testEvent = CloudEventBuilder.v1()
.withId("test-id")
.withType("com.example.test")
.withSource(URI.create("https://test.example.com"))
.withTime(OffsetDateTime.now())
.withDataContentType("application/json")
.withData("{\"test\": \"data\"}".getBytes())
.build();
}
@Test
void shouldSendEventToBrokerSuccessfully() {
// Given
ResponseEntity<String> response = ResponseEntity.accepted().body("OK");
when(restTemplate.exchange(
anyString(),
eq(HttpMethod.POST),
any(HttpEntity.class),
eq(String.class)
)).thenReturn(response);
// When
boolean result = eventSenderService.sendToBroker(testEvent);
// Then
assertTrue(result);
verify(restTemplate, times(1)).exchange(
anyString(),
eq(HttpMethod.POST),
any(HttpEntity.class),
eq(String.class)
);
}
@Test
void shouldHandleBrokerError() {
// Given
ResponseEntity<String> response = ResponseEntity.status(500).body("Error");
when(restTemplate.exchange(
anyString(),
eq(HttpMethod.POST),
any(HttpEntity.class),
eq(String.class)
)).thenReturn(response);
// When
boolean result = eventSenderService.sendToBroker(testEvent);
// Then
assertFalse(result);
}
}

8.2 Integration Tests

// File: src/test/java/com/example/events/sink/EventSinkControllerIntegrationTest.java
package com.example.events.sink;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.jackson.JsonFormat;
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 java.time.OffsetDateTime;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
@SpringBootTest
@AutoConfigureMockMvc
class EventSinkControllerIntegrationTest {
@Autowired
private MockMvc mockMvc;
@Test
void shouldAcceptValidCloudEvent() throws Exception {
CloudEvent cloudEvent = CloudEventBuilder.v1()
.withId("test-integration-id")
.withType("com.example.user.created")
.withSource(URI.create("https://test.example.com"))
.withTime(OffsetDateTime.now())
.withDataContentType("application/json")
.withData("{\"userId\": \"123\", \"name\": \"Test User\"}".getBytes())
.build();
byte[] eventBytes = new JsonFormat().serialize(cloudEvent);
mockMvc.perform(post("/api/v1/sink/events")
.contentType(MediaType.APPLICATION_JSON)
.header("ce-id", cloudEvent.getId())
.header("ce-type", cloudEvent.getType())
.header("ce-source", cloudEvent.getSource().toString())
.header("ce-specversion", "1.0")
.content(eventBytes))
.andExpect(status().isAccepted())
.andExpect(content().string("Event processed successfully"));
}
}

Best Practices for Knative Eventing in Java

  1. Idempotency: Design event handlers to be idempotent to handle duplicate events
  2. Error Handling: Implement proper error handling and dead letter queues
  3. Observability: Add comprehensive logging, metrics, and tracing
  4. Resource Management: Configure proper resource limits and scaling parameters
  5. Security: Implement authentication and authorization for event endpoints
  6. Testing: Test event flows end-to-end including failure scenarios
  7. Monitoring: Set up alerts for event processing failures and performance issues
  8. Documentation: Document event schemas and processing expectations

Conclusion

Building Knative Eventing Sources in Java enables you to create robust, scalable event-driven architectures on Kubernetes. By leveraging the CloudEvents specification and Knative's powerful event routing capabilities, Java applications can:

  • Process events from multiple sources (HTTP, Kafka, custom)
  • Route events intelligently using Knative brokers and triggers
  • Scale automatically based on event load
  • Maintain interoperability with other CloudEvents-compliant systems
  • Implement complex event processing with type-safe Java code

The patterns and examples in this guide provide a solid foundation for building production-ready Knative event sources and sinks in Java, enabling you to create sophisticated event-driven microservices architectures.

Pyroscope Profiling in Java
Explains how to use Pyroscope for continuous profiling in Java applications, helping developers analyze CPU and memory usage patterns to improve performance and identify bottlenecks.
https://macronepal.com/blog/pyroscope-profiling-in-java/

OpenTelemetry Metrics in Java: Comprehensive Guide
Provides a complete guide to collecting and exporting metrics in Java using OpenTelemetry, including counters, histograms, gauges, and integration with monitoring tools. (MACRO NEPAL)
https://macronepal.com/blog/opentelemetry-metrics-in-java-comprehensive-guide/

OTLP Exporter in Java: Complete Guide for OpenTelemetry
Explains how to configure OTLP exporters in Java to send telemetry data such as traces, metrics, and logs to monitoring systems using HTTP or gRPC protocols. (MACRO NEPAL)
https://macronepal.com/blog/otlp-exporter-in-java-complete-guide-for-opentelemetry/

Thanos Integration in Java: Global View of Metrics
Explains how to integrate Thanos with Java monitoring systems to create a scalable global metrics view across multiple Prometheus instances.

https://macronepal.com/blog/thanos-integration-in-java-global-view-of-metrics

Time Series with InfluxDB in Java: Complete Guide (Version 2)
Explains how to manage time-series data using InfluxDB in Java applications, including storing, querying, and analyzing metrics data.

https://macronepal.com/blog/time-series-with-influxdb-in-java-complete-guide-2

Time Series with InfluxDB in Java: Complete Guide
Provides an overview of integrating InfluxDB with Java for time-series data handling, including monitoring applications and managing performance metrics.

https://macronepal.com/blog/time-series-with-influxdb-in-java-complete-guide

Implementing Prometheus Remote Write in Java (Version 2)
Explains how to configure Java applications to send metrics data to Prometheus-compatible systems using the remote write feature for scalable monitoring.

https://macronepal.com/blog/implementing-prometheus-remote-write-in-java-a-complete-guide-2

Implementing Prometheus Remote Write in Java: Complete Guide
Provides instructions for sending metrics from Java services to Prometheus servers, enabling centralized monitoring and real-time analytics.

https://macronepal.com/blog/implementing-prometheus-remote-write-in-java-a-complete-guide

Building a TileServer GL in Java: Vector and Raster Tile Server
Explains how to build a TileServer GL in Java for serving vector and raster map tiles, useful for geographic visualization and mapping applications.

https://macronepal.com/blog/building-a-tileserver-gl-in-java-vector-and-raster-tile-server

Indoor Mapping in Java
Explains how to create indoor mapping systems in Java, including navigation inside buildings, spatial data handling, and visualization techniques.

Category : JAVA

Leave a Reply

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


Macro Nepal Helper