Overview
Java Flight Recorder (JFR) Event Streaming allows real-time monitoring of JVM events by streaming them directly to external systems like Kafka. This enables centralized monitoring, analysis, and alerting for Java applications.
Key Components
- JFR Event Streaming API (JDK 14+)
- Kafka Producer API
- Custom Event Processing
- Configuration Management
Basic Setup
1. Dependencies
<dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.4.0</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.15.2</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.15.2</version> </dependency> </dependencies>
2. Basic JFR to Kafka Streamer
import jdk.jfr.consumer.*;
import org.apache.kafka.clients.producer.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.nio.file.Path;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
public class BasicJfrKafkaStreamer {
private final KafkaProducer<String, String> kafkaProducer;
private final String topic;
private final ObjectMapper objectMapper;
private volatile boolean running = true;
public BasicJfrKafkaStreamer(String bootstrapServers, String topic) {
this.topic = topic;
this.objectMapper = new ObjectMapper();
// Kafka producer configuration
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.ACKS_CONFIG, "1");
this.kafkaProducer = new KafkaProducer<>(props);
}
public void startStreaming() {
try {
// Create a recording that streams events
RecordingStream recordingStream = new RecordingStream();
// Enable common event types
recordingStream.enable("jdk.CPULoad").withPeriod(Duration.ofSeconds(1));
recordingStream.enable("jdk.GarbageCollection").withPeriod(Duration.ofSeconds(5));
recordingStream.enable("jdk.JavaMonitorEnter").withThreshold(Duration.ofMillis(10));
recordingStream.enable("jdk.ThreadSleep").withThreshold(Duration.ofSeconds(1));
// Subscribe to events
recordingStream.onEvent(this::handleJfrEvent);
// Start the stream
recordingStream.start();
} catch (Exception e) {
System.err.println("Failed to start JFR streaming: " + e.getMessage());
}
}
private void handleJfrEvent(RecordedEvent event) {
try {
String eventName = event.getEventType().getName();
Map<String, Object> eventData = parseEventData(event);
// Create Kafka message
String message = objectMapper.writeValueAsString(eventData);
String key = eventName + "-" + System.currentTimeMillis();
// Send to Kafka
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, message);
kafkaProducer.send(record, (metadata, exception) -> {
if (exception != null) {
System.err.println("Failed to send event to Kafka: " + exception.getMessage());
}
});
} catch (Exception e) {
System.err.println("Error processing JFR event: " + e.getMessage());
}
}
private Map<String, Object> parseEventData(RecordedEvent event) {
Map<String, Object> data = new LinkedHashMap<>();
data.put("eventType", event.getEventType().getName());
data.put("startTime", event.getStartTime().toString());
data.put("duration", event.getDuration().toNanos());
// Add event-specific fields
event.getFields().forEach(field -> {
try {
Object value = event.getValue(field.getName());
data.put(field.getName(), value != null ? value.toString() : null);
} catch (Exception e) {
data.put(field.getName(), "ERROR: " + e.getMessage());
}
});
return data;
}
public void stop() {
running = false;
kafkaProducer.close(Duration.ofSeconds(5));
}
}
Advanced Implementation
1. Configurable JFR Kafka Streamer
import jdk.jfr.Configuration;
import jdk.jfr.FlightRecorder;
import jdk.jfr.consumer.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
public class ConfigurableJfrKafkaStreamer {
private final KafkaProducer<String, String> kafkaProducer;
private final JfrStreamerConfig config;
private final ObjectMapper objectMapper;
private final ExecutorService executor;
private volatile boolean running = false;
private RecordingStream recordingStream;
public ConfigurableJfrKafkaStreamer(JfrStreamerConfig config) {
this.config = config;
this.objectMapper = new ObjectMapper();
this.executor = Executors.newSingleThreadExecutor();
Properties props = new Properties();
props.putAll(config.getKafkaProperties());
this.kafkaProducer = new KafkaProducer<>(props);
}
public void start() {
if (running) {
throw new IllegalStateException("Streamer is already running");
}
running = true;
executor.submit(this::streamEvents);
}
private void streamEvents() {
try {
recordingStream = new RecordingStream();
// Configure event settings
configureEvents(recordingStream);
// Set event handlers
recordingStream.onEvent(this::processEvent);
recordingStream.onFlush(this::handleFlush);
recordingStream.onClose(this::handleClose);
// Start streaming
recordingStream.start();
} catch (Exception e) {
System.err.println("JFR streaming failed: " + e.getMessage());
stop();
}
}
private void configureEvents(RecordingStream stream) {
config.getEventConfigs().forEach((eventName, eventConfig) -> {
try {
EventSettings settings = stream.enable(eventName);
if (eventConfig.getPeriod() != null) {
settings.withPeriod(eventConfig.getPeriod());
}
if (eventConfig.getThreshold() != null) {
settings.withThreshold(eventConfig.getThreshold());
}
if (eventConfig.getStackTrace() != null) {
settings.withStackTrace(eventConfig.getStackTrace());
}
} catch (Exception e) {
System.err.println("Failed to configure event: " + eventName + " - " + e.getMessage());
}
});
}
private void processEvent(RecordedEvent event) {
if (!running) return;
try {
JfrEventMessage message = new JfrEventMessage(event);
String jsonMessage = objectMapper.writeValueAsString(message);
String key = generateKey(event);
ProducerRecord<String, String> record =
new ProducerRecord<>(config.getTopic(), key, jsonMessage);
kafkaProducer.send(record, new JfrEventCallback(event.getEventType().getName()));
} catch (Exception e) {
System.err.println("Failed to process JFR event: " + e.getMessage());
}
}
private String generateKey(RecordedEvent event) {
return String.format("%s-%d-%d",
event.getEventType().getName(),
event.getStartTime().toEpochMilli(),
ThreadLocalRandom.current().nextInt(1000));
}
private void handleFlush() {
System.out.println("JFR stream flushed");
kafkaProducer.flush();
}
private void handleClose() {
System.out.println("JFR stream closed");
stop();
}
public void stop() {
running = false;
if (recordingStream != null) {
recordingStream.close();
}
executor.shutdown();
try {
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
kafkaProducer.close(Duration.ofSeconds(5));
}
// Configuration classes
public static class JfrStreamerConfig {
private String topic;
private Map<String, EventConfig> eventConfigs = new HashMap<>();
private Properties kafkaProperties = new Properties();
// Getters and setters
public String getTopic() { return topic; }
public void setTopic(String topic) { this.topic = topic; }
public Map<String, EventConfig> getEventConfigs() { return eventConfigs; }
public void setEventConfigs(Map<String, EventConfig> eventConfigs) {
this.eventConfigs = eventConfigs;
}
public Properties getKafkaProperties() { return kafkaProperties; }
public void setKafkaProperties(Properties kafkaProperties) {
this.kafkaProperties = kafkaProperties;
}
}
public static class EventConfig {
private Duration period;
private Duration threshold;
private Boolean stackTrace;
// Getters and setters
public Duration getPeriod() { return period; }
public void setPeriod(Duration period) { this.period = period; }
public Duration getThreshold() { return threshold; }
public void setThreshold(Duration threshold) { this.threshold = threshold; }
public Boolean getStackTrace() { return stackTrace; }
public void setStackTrace(Boolean stackTrace) { this.stackTrace = stackTrace; }
}
// Event message wrapper
public static class JfrEventMessage {
private final String eventType;
private final long startTime;
private final long duration;
private final Map<String, Object> fields;
public JfrEventMessage(RecordedEvent event) {
this.eventType = event.getEventType().getName();
this.startTime = event.getStartTime().toEpochMilli();
this.duration = event.getDuration().toNanos();
this.fields = extractFields(event);
}
private Map<String, Object> extractFields(RecordedEvent event) {
Map<String, Object> fieldMap = new LinkedHashMap<>();
event.getFields().forEach(field -> {
try {
Object value = event.getValue(field.getName());
fieldMap.put(field.getName(), convertValue(value));
} catch (Exception e) {
fieldMap.put(field.getName(), "ERROR: " + e.getMessage());
}
});
return fieldMap;
}
private Object convertValue(Object value) {
if (value == null) return null;
if (value instanceof RecordedObject) {
return value.toString(); // Simplified handling
}
return value;
}
// Getters for JSON serialization
public String getEventType() { return eventType; }
public long getStartTime() { return startTime; }
public long getDuration() { return duration; }
public Map<String, Object> getFields() { return fields; }
}
// Kafka callback
private class JfrEventCallback implements Callback {
private final String eventType;
public JfrEventCallback(String eventType) {
this.eventType = eventType;
}
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
System.err.println("Failed to send " + eventType + " to Kafka: " + exception.getMessage());
}
}
}
}
2. Event Filtering and Transformation
public class FilteredJfrKafkaStreamer {
private final ConfigurableJfrKafkaStreamer delegate;
private final List<EventFilter> filters;
private final List<EventTransformer> transformers;
public FilteredJfrKafkaStreamer(JfrStreamerConfig config,
List<EventFilter> filters,
List<EventTransformer> transformers) {
this.delegate = new ConfigurableJfrKafkaStreamer(config);
this.filters = filters != null ? filters : new ArrayList<>();
this.transformers = transformers != null ? transformers : new ArrayList<>();
}
public void start() {
delegate.start();
}
public void stop() {
delegate.stop();
}
// Filter interface
public interface EventFilter {
boolean shouldInclude(RecordedEvent event);
}
// Transformer interface
public interface EventTransformer {
Map<String, Object> transform(RecordedEvent event);
}
// Built-in filters
public static class DurationFilter implements EventFilter {
private final long minDurationNanos;
public DurationFilter(long minDurationNanos) {
this.minDurationNanos = minDurationNanos;
}
@Override
public boolean shouldInclude(RecordedEvent event) {
return event.getDuration().toNanos() >= minDurationNanos;
}
}
public static class EventTypeFilter implements EventFilter {
private final Set<String> includedTypes;
public EventTypeFilter(Set<String> includedTypes) {
this.includedTypes = includedTypes;
}
@Override
public boolean shouldInclude(RecordedEvent event) {
return includedTypes.contains(event.getEventType().getName());
}
}
// Built-in transformers
public static class SimplifiedGcTransformer implements EventTransformer {
@Override
public Map<String, Object> transform(RecordedEvent event) {
Map<String, Object> simplified = new HashMap<>();
simplified.put("eventType", event.getEventType().getName());
simplified.put("timestamp", event.getStartTime().toEpochMilli());
if ("jdk.GarbageCollection".equals(event.getEventType().getName())) {
simplified.put("gcId", event.getValue("gcId"));
simplified.put("longestPause", event.getDuration().toMillis());
simplified.put("name", event.getValue("name"));
}
return simplified;
}
}
public static class CpuLoadTransformer implements EventTransformer {
@Override
public Map<String, Object> transform(RecordedEvent event) {
Map<String, Object> transformed = new HashMap<>();
transformed.put("eventType", event.getEventType().getName());
transformed.put("timestamp", event.getStartTime().toEpochMilli());
if ("jdk.CPULoad".equals(event.getEventType().getName())) {
transformed.put("jvmUser", event.getValue("jvmUser"));
transformed.put("jvmSystem", event.getValue("jvmSystem"));
transformed.put("machineTotal", event.getValue("machineTotal"));
}
return transformed;
}
}
}
Practical Examples
Example 1: Production-Ready JFR Monitor
public class ProductionJfrMonitor {
private final ConfigurableJfrKafkaStreamer streamer;
private final MetricsCollector metrics;
private final AlertManager alertManager;
public ProductionJfrMonitor(String bootstrapServers, String topic) {
JfrStreamerConfig config = createConfig(bootstrapServers, topic);
this.streamer = new ConfigurableJfrKafkaStreamer(config);
this.metrics = new MetricsCollector();
this.alertManager = new AlertManager();
}
private JfrStreamerConfig createConfig(String bootstrapServers, String topic) {
JfrStreamerConfig config = new JfrStreamerConfig();
config.setTopic(topic);
// Kafka configuration
Properties kafkaProps = new Properties();
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put(ProducerConfig.ACKS_CONFIG, "all");
kafkaProps.put(ProducerConfig.RETRIES_CONFIG, "3");
kafkaProps.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
kafkaProps.put(ProducerConfig.LINGER_MS_CONFIG, "10");
config.setKafkaProperties(kafkaProps);
// JFR event configuration
Map<String, EventConfig> eventConfigs = new HashMap<>();
// CPU monitoring
EventConfig cpuConfig = new EventConfig();
cpuConfig.setPeriod(Duration.ofSeconds(5));
eventConfigs.put("jdk.CPULoad", cpuConfig);
// Memory monitoring
EventConfig memoryConfig = new EventConfig();
memoryConfig.setPeriod(Duration.ofSeconds(10));
eventConfigs.put("jdk.GCHeapSummary", memoryConfig);
// GC monitoring
EventConfig gcConfig = new EventConfig();
gcConfig.setPeriod(Duration.ofSeconds(30));
eventConfigs.put("jdk.GarbageCollection", gcConfig);
// Thread monitoring
EventConfig threadConfig = new EventConfig();
threadConfig.setThreshold(Duration.ofMillis(100));
eventConfigs.put("jdk.JavaMonitorEnter", threadConfig);
config.setEventConfigs(eventConfigs);
return config;
}
public void startMonitoring() {
System.out.println("Starting JFR to Kafka monitoring...");
streamer.start();
metrics.start();
}
public void stopMonitoring() {
System.out.println("Stopping JFR to Kafka monitoring...");
streamer.stop();
metrics.stop();
}
// Inner classes for supporting functionality
private static class MetricsCollector {
private final Map<String, Long> eventCounts = new ConcurrentHashMap<>();
private ScheduledExecutorService scheduler;
public void start() {
scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(this::reportMetrics, 1, 1, TimeUnit.MINUTES);
}
public void stop() {
if (scheduler != null) {
scheduler.shutdown();
}
}
private void reportMetrics() {
System.out.println("JFR Event Metrics:");
eventCounts.forEach((event, count) -> {
System.out.printf(" %s: %d events/min%n", event, count);
});
eventCounts.clear();
}
public void recordEvent(String eventType) {
eventCounts.merge(eventType, 1L, Long::sum);
}
}
private static class AlertManager {
private static final double CPU_THRESHOLD = 0.9; // 90%
private static final long GC_THRESHOLD_MS = 1000; // 1 second
public void checkForAlerts(RecordedEvent event) {
String eventType = event.getEventType().getName();
switch (eventType) {
case "jdk.CPULoad":
checkCpuAlert(event);
break;
case "jdk.GarbageCollection":
checkGcAlert(event);
break;
case "jdk.JavaMonitorEnter":
checkThreadAlert(event);
break;
}
}
private void checkCpuAlert(RecordedEvent event) {
Double machineTotal = event.getValue("machineTotal");
if (machineTotal != null && machineTotal > CPU_THRESHOLD) {
System.err.printf("HIGH CPU ALERT: %.2f%% utilization%n", machineTotal * 100);
}
}
private void checkGcAlert(RecordedEvent event) {
long durationMs = event.getDuration().toMillis();
if (durationMs > GC_THRESHOLD_MS) {
System.err.printf("LONG GC ALERT: %d ms pause%n", durationMs);
}
}
private void checkThreadAlert(RecordedEvent event) {
Long duration = event.getValue("duration");
if (duration != null && duration > 1000) { // 1 second in milliseconds
String monitorClass = event.getValue("monitorClass");
System.err.printf("LONG MONITOR WAIT: %d ms for %s%n", duration, monitorClass);
}
}
}
}
Example 2: Custom JFR Events to Kafka
// Custom JFR event
import jdk.jfr.*;
@Name("com.example.ApplicationEvent")
@Label("Application Custom Event")
@Category("Application")
@Description("Custom application-level events")
public class ApplicationEvent extends Event {
@Label("Event Type")
@Description("Type of application event")
private String type;
@Label("User ID")
@Description("User associated with the event")
private String userId;
@Label("Duration")
@Description("Event duration in milliseconds")
private long duration;
@Label("Success")
@Description("Whether the operation was successful")
private boolean success;
// Getters and setters
public String getType() { return type; }
public void setType(String type) { this.type = type; }
public String getUserId() { return userId; }
public void setUserId(String userId) { this.userId = userId; }
public long getDuration() { return duration; }
public void setDuration(long duration) { this.duration = duration; }
public boolean isSuccess() { return success; }
public void setSuccess(boolean success) { this.success = success; }
}
// Custom event producer and streamer
public class CustomJfrEventStreamer {
private final ConfigurableJfrKafkaStreamer streamer;
private final ObjectMapper objectMapper;
public CustomJfrEventStreamer(String bootstrapServers, String topic) {
JfrStreamerConfig config = createConfig(bootstrapServers, topic);
this.streamer = new ConfigurableJfrKafkaStreamer(config);
this.objectMapper = new ObjectMapper();
}
private JfrStreamerConfig createConfig(String bootstrapServers, String topic) {
JfrStreamerConfig config = new JfrStreamerConfig();
config.setTopic(topic);
Properties kafkaProps = new Properties();
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
config.setKafkaProperties(kafkaProps);
// Enable custom events
Map<String, EventConfig> eventConfigs = new HashMap<>();
EventConfig customEventConfig = new EventConfig();
customEventConfig.setStackTrace(true);
eventConfigs.put("com.example.ApplicationEvent", customEventConfig);
config.setEventConfigs(eventConfigs);
return config;
}
public void start() {
streamer.start();
}
public void stop() {
streamer.stop();
}
// Method to record custom events
public void recordApplicationEvent(String type, String userId, long duration, boolean success) {
ApplicationEvent event = new ApplicationEvent();
event.setType(type);
event.setUserId(userId);
event.setDuration(duration);
event.setSuccess(success);
event.commit();
}
// Custom event processor
public static class CustomEventProcessor {
private final KafkaProducer<String, String> producer;
private final String topic;
public CustomEventProcessor(String bootstrapServers, String topic) {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
this.producer = new KafkaProducer<>(props);
this.topic = topic;
}
public void processCustomEvent(RecordedEvent event) {
try {
Map<String, Object> eventData = new HashMap<>();
eventData.put("eventType", event.getEventType().getName());
eventData.put("timestamp", event.getStartTime().toEpochMilli());
eventData.put("type", event.getValue("type"));
eventData.put("userId", event.getValue("userId"));
eventData.put("duration", event.getValue("duration"));
eventData.put("success", event.getValue("success"));
String message = new ObjectMapper().writeValueAsString(eventData);
ProducerRecord<String, String> record =
new ProducerRecord<>(topic, event.getString("type"), message);
producer.send(record);
} catch (Exception e) {
System.err.println("Failed to process custom event: " + e.getMessage());
}
}
}
}
Example 3: Kubernetes Deployment with JFR Streaming
# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: jfr-kafka-streamer spec: replicas: 2 selector: matchLabels: app: jfr-kafka-streamer template: metadata: labels: app: jfr-kafka-streamer spec: containers: - name: jfr-streamer image: mycompany/jfr-kafka-streamer:latest env: - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka-cluster:9092" - name: KAFKA_TOPIC value: "jfr-events" - name: JVM_OPTS value: "-XX:+FlightRecorder -XX:StartFlightRecording=disk=true,maxsize=1G,maxage=24h" resources: requests: memory: "512Mi" cpu: "200m" limits: memory: "1Gi" cpu: "500m"
// Kubernetes-aware streamer
public class KubernetesJfrStreamer {
private final String podName;
private final String namespace;
private final ConfigurableJfrKafkaStreamer streamer;
public KubernetesJfrStreamer(String bootstrapServers, String topic) {
this.podName = System.getenv().getOrDefault("HOSTNAME", "unknown");
this.namespace = System.getenv().getOrDefault("NAMESPACE", "default");
JfrStreamerConfig config = createKubernetesConfig(bootstrapServers, topic);
this.streamer = new ConfigurableJfrKafkaStreamer(config);
}
private JfrStreamerConfig createKubernetesConfig(String bootstrapServers, String topic) {
JfrStreamerConfig config = new JfrStreamerConfig();
config.setTopic(topic);
Properties kafkaProps = new Properties();
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put(ProducerConfig.CLIENT_ID_CONFIG, podName);
config.setKafkaProperties(kafkaProps);
// Enhanced event configuration for Kubernetes
Map<String, EventConfig> eventConfigs = new HashMap<>();
// Container resource monitoring
EventConfig containerConfig = new EventConfig();
containerConfig.setPeriod(Duration.ofSeconds(10));
eventConfigs.put("jdk.CPULoad", containerConfig);
eventConfigs.put("jdk.GCHeapSummary", containerConfig);
// Application performance monitoring
EventConfig perfConfig = new EventConfig();
perfConfig.setThreshold(Duration.ofMillis(100));
eventConfigs.put("jdk.JavaMonitorEnter", perfConfig);
eventConfigs.put("jdk.ThreadSleep", perfConfig);
config.setEventConfigs(eventConfigs);
return config;
}
public void start() {
System.out.printf("Starting JFR streaming from pod %s in namespace %s%n",
podName, namespace);
streamer.start();
// Add shutdown hook for graceful shutdown
Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
}
public void stop() {
System.out.println("Shutting down JFR streaming...");
streamer.stop();
}
// Enhanced event processing with Kubernetes metadata
public static class KubernetesEventEnhancer implements FilteredJfrKafkaStreamer.EventTransformer {
private final String podName;
private final String namespace;
public KubernetesEventEnhancer(String podName, String namespace) {
this.podName = podName;
this.namespace = namespace;
}
@Override
public Map<String, Object> transform(RecordedEvent event) {
Map<String, Object> enhanced = new HashMap<>();
enhanced.put("eventType", event.getEventType().getName());
enhanced.put("timestamp", event.getStartTime().toEpochMilli());
enhanced.put("duration", event.getDuration().toNanos());
enhanced.put("kubernetes_pod", podName);
enhanced.put("kubernetes_namespace", namespace);
enhanced.put("jvm_id", ManagementFactory.getRuntimeMXBean().getName());
// Add event-specific fields
event.getFields().forEach(field -> {
try {
Object value = event.getValue(field.getName());
if (value != null) {
enhanced.put(field.getName(), value.toString());
}
} catch (Exception e) {
// Skip problematic fields
}
});
return enhanced;
}
}
}
Best Practices
1. Configuration Management
public class JfrStreamerFactory {
public static ConfigurableJfrKafkaStreamer createFromEnvironment() {
String bootstrapServers = getEnv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092");
String topic = getEnv("KAFKA_TOPIC", "jfr-events");
String configFile = getEnv("JFR_CONFIG", "jfr-config.json");
JfrStreamerConfig config = loadConfig(configFile);
config.getKafkaProperties().putIfAbsent(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
config.setTopic(topic);
return new ConfigurableJfrKafkaStreamer(config);
}
private static JfrStreamerConfig loadConfig(String configFile) {
try {
ObjectMapper mapper = new ObjectMapper();
return mapper.readValue(new File(configFile), JfrStreamerConfig.class);
} catch (Exception e) {
System.err.println("Failed to load config, using defaults: " + e.getMessage());
return createDefaultConfig();
}
}
private static JfrStreamerConfig createDefaultConfig() {
JfrStreamerConfig config = new JfrStreamerConfig();
// Default event configuration
Map<String, EventConfig> events = new HashMap<>();
events.put("jdk.CPULoad", createEventConfig(Duration.ofSeconds(5), null));
events.put("jdk.GarbageCollection", createEventConfig(Duration.ofSeconds(30), null));
events.put("jdk.JavaMonitorEnter", createEventConfig(null, Duration.ofMillis(100)));
config.setEventConfigs(events);
// Default Kafka properties
Properties kafkaProps = new Properties();
kafkaProps.put(ProducerConfig.ACKS_CONFIG, "1");
kafkaProps.put(ProducerConfig.RETRIES_CONFIG, "3");
kafkaProps.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
config.setKafkaProperties(kafkaProps);
return config;
}
private static EventConfig createEventConfig(Duration period, Duration threshold) {
EventConfig config = new EventConfig();
config.setPeriod(period);
config.setThreshold(threshold);
return config;
}
private static String getEnv(String key, String defaultValue) {
String value = System.getenv(key);
return value != null ? value : defaultValue;
}
}
2. Monitoring and Health Checks
public class JfrStreamerHealthCheck {
private final ConfigurableJfrKafkaStreamer streamer;
private final KafkaProducer<String, String> producer;
private long lastEventTime = System.currentTimeMillis();
private final long healthCheckInterval = 60000; // 1 minute
public JfrStreamerHealthCheck(ConfigurableJfrKafkaStreamer streamer,
KafkaProducer<String, String> producer) {
this.streamer = streamer;
this.producer = producer;
}
public boolean isHealthy() {
long currentTime = System.currentTimeMillis();
// Check if we're receiving events regularly
if (currentTime - lastEventTime > healthCheckInterval * 2) {
return false; // No events for too long
}
// Additional health checks can be added here
return true;
}
public void recordEvent() {
lastEventTime = System.currentTimeMillis();
}
public Map<String, Object> getHealthStatus() {
Map<String, Object> status = new HashMap<>();
status.put("healthy", isHealthy());
status.put("lastEventTime", lastEventTime);
status.put("timeSinceLastEvent", System.currentTimeMillis() - lastEventTime);
return status;
}
}
This comprehensive JFR to Kafka streaming implementation provides production-ready monitoring capabilities with proper error handling, configuration management, and Kubernetes support.