Overview

KEDA (Kubernetes-based Event Driven Autoscaling) allows you to automatically scale your Java applications based on events from various sources like Kafka, RabbitMQ, Redis, HTTP, and more. This enables true event-driven autoscaling for your microservices.

Core KEDA Concepts

1. KEDA ScaledObject Configuration Generator

import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; import io.kubernetes.client.openapi.models.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; import java.util.stream.Collectors; public class KedaConfigGenerator { private static final Logger logger = LoggerFactory.getLogger(KedaConfigGenerator.class); private final ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory()); public String generateScaledObject(ScaledObjectConfig config) { try { Map<String, Object> scaledObject = createScaledObject(config); return yamlMapper.writeValueAsString(scaledObject); } catch (Exception e) { throw new RuntimeException("Failed to generate ScaledObject YAML", e); } } private Map<String, Object> createScaledObject(ScaledObjectConfig config) { Map<String, Object> scaledObject = new LinkedHashMap<>(); // API Version scaledObject.put("apiVersion", "keda.sh/v1alpha1"); scaledObject.put("kind", "ScaledObject"); // Metadata Map<String, Object> metadata = new LinkedHashMap<>(); metadata.put("name", config.getName()); metadata.put("namespace", config.getNamespace()); if (config.getLabels() != null) { metadata.put("labels", config.getLabels()); } scaledObject.put("metadata", metadata); // Spec Map<String, Object> spec = new LinkedHashMap<>(); // Scale Target Map<String, Object> scaleTargetRef = new LinkedHashMap<>(); scaleTargetRef.put("apiVersion", config.getTargetApiVersion()); scaleTargetRef.put("kind", config.getTargetKind()); scaleTargetRef.put("name", config.getTargetName()); spec.put("scaleTargetRef", scaleTargetRef); // Polling Interval if (config.getPollingInterval() != null) { spec.put("pollingInterval", config.getPollingInterval()); } // Cooldown Period if (config.getCooldownPeriod() != null) { spec.put("cooldownPeriod", config.getCooldownPeriod()); } // Min/Max Replica Count spec.put("minReplicaCount", config.getMinReplicas()); if (config.getMaxReplicas() != null) { spec.put("maxReplicaCount", config.getMaxReplicas()); } // Advanced Config if (config.getAdvanced() != null) { spec.put("advanced", config.getAdvanced()); } // Triggers spec.put("triggers", config.getTriggers().stream() .map(this::createTrigger) .collect(Collectors.toList())); scaledObject.put("spec", spec); return scaledObject; } private Map<String, Object> createTrigger(TriggerConfig trigger) { Map<String, Object> triggerMap = new LinkedHashMap<>(); triggerMap.put("type", trigger.getType()); if (trigger.getMetadata() != null && !trigger.getMetadata().isEmpty()) { triggerMap.put("metadata", trigger.getMetadata()); } if (trigger.getAuthenticationRef() != null) { Map<String, Object> authRef = new LinkedHashMap<>(); authRef.put("name", trigger.getAuthenticationRef()); triggerMap.put("authenticationRef", authRef); } return triggerMap; } public static class ScaledObjectConfig { private String name; private String namespace = "default"; private Map<String, String> labels; private String targetApiVersion = "apps/v1"; private String targetKind = "Deployment"; private String targetName; private Integer pollingInterval = 30; private Integer cooldownPeriod = 300; private Integer minReplicas = 0; private Integer maxReplicas = 100; private Map<String, Object> advanced; private List<TriggerConfig> triggers = new ArrayList<>(); // Builder methods public ScaledObjectConfig withName(String name) { this.name = name; return this; } public ScaledObjectConfig withNamespace(String namespace) { this.namespace = namespace; return this; } public ScaledObjectConfig withTarget(String targetName) { this.targetName = targetName; return this; } public ScaledObjectConfig withTarget(String targetKind, String targetName) { this.targetKind = targetKind; this.targetName = targetName; return this; } public ScaledObjectConfig withReplicaBounds(int min, int max) { this.minReplicas = min; this.maxReplicas = max; return this; } public ScaledObjectConfig withPollingInterval(int seconds) { this.pollingInterval = seconds; return this; } public ScaledObjectConfig withCooldownPeriod(int seconds) { this.cooldownPeriod = seconds; return this; } public ScaledObjectConfig withTrigger(TriggerConfig trigger) { this.triggers.add(trigger); return this; } public ScaledObjectConfig withAdvanced(Map<String, Object> advanced) { this.advanced = advanced; return this; } // Getters public String getName() { return name; } public String getNamespace() { return namespace; } public Map<String, String> getLabels() { return labels; } public String getTargetApiVersion() { return targetApiVersion; } public String getTargetKind() { return targetKind; } public String getTargetName() { return targetName; } public Integer getPollingInterval() { return pollingInterval; } public Integer getCooldownPeriod() { return cooldownPeriod; } public Integer getMinReplicas() { return minReplicas; } public Integer getMaxReplicas() { return maxReplicas; } public Map<String, Object> getAdvanced() { return advanced; } public List<TriggerConfig> getTriggers() { return triggers; } } public static class TriggerConfig { private String type; private Map<String, String> metadata; private String authenticationRef; public static TriggerConfig kafka(String bootstrapServers, String topic, String consumerGroup, int lagThreshold) { Map<String, String> metadata = new HashMap<>(); metadata.put("bootstrapServers", bootstrapServers); metadata.put("topic", topic); metadata.put("consumerGroup", consumerGroup); metadata.put("lagThreshold", String.valueOf(lagThreshold)); metadata.put("offsetResetPolicy", "latest"); return new TriggerConfig() .withType("kafka") .withMetadata(metadata); } public static TriggerConfig rabbitMQ(String host, String queueName, String vhost, int queueLength) { Map<String, String> metadata = new HashMap<>(); metadata.put("host", host); metadata.put("queueName", queueName); metadata.put("vhost", vhost); metadata.put("queueLength", String.valueOf(queueLength)); metadata.put("protocol", "amqp"); return new TriggerConfig() .withType("rabbitmq") .withMetadata(metadata); } public static TriggerConfig redis(String address, String listName, int listLength, String listType) { Map<String, String> metadata = new HashMap<>(); metadata.put("address", address); metadata.put("listName", listName); metadata.put("listLength", String.valueOf(listLength)); metadata.put("listType", listType); // "list" or "set" return new TriggerConfig() .withType("redis") .withMetadata(metadata); } public static TriggerConfig http(String url, String valueLocation, int threshold) { Map<String, String> metadata = new HashMap<>(); metadata.put("url", url); metadata.put("valueLocation", valueLocation); // e.g., "message.count" metadata.put("threshold", String.valueOf(threshold)); return new TriggerConfig() .withType("http") .withMetadata(metadata); } public static TriggerConfig cron(String desiredReplicas, String timezone) { Map<String, String> metadata = new HashMap<>(); metadata.put("desiredReplicas", desiredReplicas); metadata.put("timezone", timezone); metadata.put("start", "0 9 * * 1-5"); // 9 AM on weekdays metadata.put("end", "0 17 * * 1-5"); // 5 PM on weekdays return new TriggerConfig() .withType("cron") .withMetadata(metadata); } // Builder methods public TriggerConfig withType(String type) { this.type = type; return this; } public TriggerConfig withMetadata(Map<String, String> metadata) { this.metadata = metadata; return this; } public TriggerConfig withAuthenticationRef(String authRef) { this.authenticationRef = authRef; return this; } // Getters public String getType() { return type; } public Map<String, String> getMetadata() { return metadata; } public String getAuthenticationRef() { return authenticationRef; } } }

2. KEDA Deployment Service

import io.kubernetes.client.openapi.ApiClient; import io.kubernetes.client.openapi.ApiException; import io.kubernetes.client.openapi.apis.CustomObjectsApi; import io.kubernetes.client.openapi.models.V1DeleteOptions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; import java.util.Map; @Service public class KedaDeploymentService { private static final Logger logger = LoggerFactory.getLogger(KedaDeploymentService.class); private final CustomObjectsApi customObjectsApi; private final KedaConfigGenerator configGenerator; public KedaDeploymentService(ApiClient apiClient, KedaConfigGenerator configGenerator) { this.customObjectsApi = new CustomObjectsApi(apiClient); this.configGenerator = configGenerator; } public void deployScaledObject(KedaConfigGenerator.ScaledObjectConfig config) throws ApiException { String scaledObjectYaml = configGenerator.generateScaledObject(config); logger.info("Deploying ScaledObject: {}", config.getName()); Map<String, Object> scaledObject = configGenerator.generateScaledObjectMap(config); try { // Create ScaledObject Object result = customObjectsApi.createNamespacedCustomObject( "keda.sh", // group "v1alpha1", // version config.getNamespace(), // namespace "scaledobjects", // plural scaledObject, // body null, // pretty null, // dryRun null, // fieldManager null // fieldValidation ); logger.info("Successfully deployed ScaledObject: {}", config.getName()); } catch (ApiException e) { logger.error("Failed to deploy ScaledObject: {}", e.getResponseBody(), e); throw e; } } public void updateScaledObject(KedaConfigGenerator.ScaledObjectConfig config) throws ApiException { Map<String, Object> scaledObject = configGenerator.generateScaledObjectMap(config); try { Object result = customObjectsApi.replaceNamespacedCustomObject( "keda.sh", "v1alpha1", config.getNamespace(), "scaledobjects", config.getName(), scaledObject, null, null, null, null ); logger.info("Successfully updated ScaledObject: {}", config.getName()); } catch (ApiException e) { logger.error("Failed to update ScaledObject: {}", e.getResponseBody(), e); throw e; } } public void deleteScaledObject(String name, String namespace) throws ApiException { try { Object result = customObjectsApi.deleteNamespacedCustomObject( "keda.sh", "v1alpha1", namespace, "scaledobjects", name, null, // gracePeriodSeconds null, // orphanDependents null, // propagationPolicy new V1DeleteOptions(), // body null, null, null ); logger.info("Successfully deleted ScaledObject: {}", name); } catch (ApiException e) { logger.error("Failed to delete ScaledObject: {}", e.getResponseBody(), e); throw e; } } public Object getScaledObject(String name, String namespace) throws ApiException { return customObjectsApi.getNamespacedCustomObject( "keda.sh", "v1alpha1", namespace, "scaledobjects", name ); } public Object listScaledObjects(String namespace) throws ApiException { return customObjectsApi.listNamespacedCustomObject( "keda.sh", "v1alpha1", namespace, "scaledobjects", null, // pretty null, // allowWatchBookmarks null, // continue null, // fieldSelector null, // labelSelector null, // limit null, // resourceVersion null, // resourceVersionMatch null, // timeoutSeconds null // watch ); } }

3. Event-Driven Application with KEDA Scaling

import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @Service public class EventDrivenProcessor { private static final Logger logger = LoggerFactory.getLogger(EventDrivenProcessor.class); private final MetricsService metricsService; public EventDrivenProcessor(MetricsService metricsService) { this.metricsService = metricsService; } @KafkaListener(topics = "${kafka.topic.orders}") public void processOrderEvent(String orderEvent) { logger.info("Processing order event: {}", orderEvent); metricsService.incrementProcessedOrders(); try { // Simulate processing time Thread.sleep(100); // Process the order processOrder(Order.fromJson(orderEvent)); metricsService.incrementSuccessfulOrders(); } catch (Exception e) { logger.error("Failed to process order event", e); metricsService.incrementFailedOrders(); } } @KafkaListener(topics = "${kafka.topic.payments}") public void processPaymentEvent(String paymentEvent) { logger.info("Processing payment event: {}", paymentEvent); metricsService.incrementProcessedPayments(); try { // Process payment processPayment(Payment.fromJson(paymentEvent)); metricsService.incrementSuccessfulPayments(); } catch (Exception e) { logger.error("Failed to process payment event", e); metricsService.incrementFailedPayments(); } } private void processOrder(Order order) { // Order processing logic logger.info("Processing order: {}", order.getId()); // Implement order processing } private void processPayment(Payment payment) { // Payment processing logic logger.info("Processing payment: {}", payment.getId()); // Implement payment processing } // Data classes public static class Order { private String id; private String customerId; private double amount; public static Order fromJson(String json) { // Simple JSON parsing - use Jackson in real implementation return new Order(); } // Getters and setters public String getId() { return id; } public void setId(String id) { this.id = id; } public String getCustomerId() { return customerId; } public void setCustomerId(String customerId) { this.customerId = customerId; } public double getAmount() { return amount; } public void setAmount(double amount) { this.amount = amount; } } public static class Payment { private String id; private String orderId; private double amount; private String status; public static Payment fromJson(String json) { // Simple JSON parsing - use Jackson in real implementation return new Payment(); } // Getters and setters public String getId() { return id; } public void setId(String id) { this.id = id; } public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId = orderId; } public double getAmount() { return amount; } public void setAmount(double amount) { this.amount = amount; } public String getStatus() { return status; } public void setStatus(String status) { this.status = status; } } }

4. Metrics Service for Scaling Decisions

import io.micrometer.core.instrument.Counter; import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; import org.springframework.stereotype.Service; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @Service public class MetricsService { private final MeterRegistry meterRegistry; // Counters private final Counter processedOrders; private final Counter successfulOrders; private final Counter failedOrders; private final Counter processedPayments; private final Counter successfulPayments; private final Counter failedPayments; // Gauges for current load private final AtomicInteger currentQueueSize = new AtomicInteger(0); private final AtomicLong processingTime = new AtomicLong(0); private final AtomicInteger activeConnections = new AtomicInteger(0); public MetricsService(MeterRegistry meterRegistry) { this.meterRegistry = meterRegistry; // Initialize counters this.processedOrders = Counter.builder("orders.processed.total") .description("Total number of processed orders") .register(meterRegistry); this.successfulOrders = Counter.builder("orders.successful.total") .description("Total number of successfully processed orders") .register(meterRegistry); this.failedOrders = Counter.builder("orders.failed.total") .description("Total number of failed orders") .register(meterRegistry); this.processedPayments = Counter.builder("payments.processed.total") .description("Total number of processed payments") .register(meterRegistry); this.successfulPayments = Counter.builder("payments.successful.total") .description("Total number of successfully processed payments") .register(meterRegistry); this.failedPayments = Counter.builder("payments.failed.total") .description("Total number of failed payments") .register(meterRegistry); // Initialize gauges Gauge.builder("queue.current.size") .description("Current queue size") .register(meterRegistry, currentQueueSize, AtomicInteger::get); Gauge.builder("processing.time.avg") .description("Average processing time in milliseconds") .register(meterRegistry, processingTime, AtomicLong::get); Gauge.builder("connections.active") .description("Number of active connections") .register(meterRegistry, activeConnections, AtomicInteger::get); } // Order metrics public void incrementProcessedOrders() { processedOrders.increment(); } public void incrementSuccessfulOrders() { successfulOrders.increment(); } public void incrementFailedOrders() { failedOrders.increment(); } // Payment metrics public void incrementProcessedPayments() { processedPayments.increment(); } public void incrementSuccessfulPayments() { successfulPayments.increment(); } public void incrementFailedPayments() { failedPayments.increment(); } // Queue metrics public void setCurrentQueueSize(int size) { currentQueueSize.set(size); } public void incrementQueueSize() { currentQueueSize.incrementAndGet(); } public void decrementQueueSize() { currentQueueSize.decrementAndGet(); } // Processing time metrics public void recordProcessingTime(long milliseconds) { processingTime.set(milliseconds); } // Connection metrics public void incrementActiveConnections() { activeConnections.incrementAndGet(); } public void decrementActiveConnections() { activeConnections.decrementAndGet(); } // Get metrics for scaling decisions public int getCurrentQueueSize() { return currentQueueSize.get(); } public long getAverageProcessingTime() { return processingTime.get(); } public int getActiveConnections() { return activeConnections.get(); } }

5. KEDA Scaling Manager

import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @Service public class KedaScalingManager { private static final Logger logger = LoggerFactory.getLogger(KedaScalingManager.class); private final KedaDeploymentService kedaDeploymentService; private final MetricsService metricsService; private final Map<String, KedaConfigGenerator.ScaledObjectConfig> scaledObjects; public KedaScalingManager(KedaDeploymentService kedaDeploymentService, MetricsService metricsService) { this.kedaDeploymentService = kedaDeploymentService; this.metricsService = metricsService; this.scaledObjects = new ConcurrentHashMap<>(); } public void registerScaledObject(String name, KedaConfigGenerator.ScaledObjectConfig config) { scaledObjects.put(name, config); logger.info("Registered ScaledObject: {}", name); } public void deployScaledObject(String name) { KedaConfigGenerator.ScaledObjectConfig config = scaledObjects.get(name); if (config != null) { try { kedaDeploymentService.deployScaledObject(config); logger.info("Deployed ScaledObject: {}", name); } catch (Exception e) { logger.error("Failed to deploy ScaledObject: {}", name, e); } } } public void updateScaledObjectThreshold(String scaledObjectName, String triggerType, String thresholdKey, int newThreshold) { KedaConfigGenerator.ScaledObjectConfig config = scaledObjects.get(scaledObjectName); if (config != null) { config.getTriggers().stream() .filter(trigger -> triggerType.equals(trigger.getType())) .findFirst() .ifPresent(trigger -> { trigger.getMetadata().put(thresholdKey, String.valueOf(newThreshold)); try { kedaDeploymentService.updateScaledObject(config); logger.info("Updated threshold for {}: {} = {}", scaledObjectName, thresholdKey, newThreshold); } catch (Exception e) { logger.error("Failed to update ScaledObject threshold", e); } }); } } // Dynamic scaling based on metrics @Scheduled(fixedRate = 30000) // Every 30 seconds public void adjustScalingDynamically() { int currentQueueSize = metricsService.getCurrentQueueSize(); long avgProcessingTime = metricsService.getAverageProcessingTime(); // Adjust Kafka lag threshold based on processing performance if (avgProcessingTime > 500) { // If processing is slow int newThreshold = Math.max(10, currentQueueSize / 10); updateScaledObjectThreshold("order-processor", "kafka", "lagThreshold", newThreshold); logger.info("Adjusted Kafka lag threshold to {} due to slow processing", newThreshold); } else if (avgProcessingTime < 100) { // If processing is fast int newThreshold = Math.min(1000, currentQueueSize * 2); updateScaledObjectThreshold("order-processor", "kafka", "lagThreshold", newThreshold); logger.info("Adjusted Kafka lag threshold to {} due to fast processing", newThreshold); } } // Scale to zero during off-hours @Scheduled(cron = "0 0 22 * * ?") // 10 PM daily public void scaleToZeroForNight() { try { updateScaledObjectThreshold("order-processor", "kafka", "lagThreshold", 0); logger.info("Scaled to zero for night hours"); } catch (Exception e) { logger.error("Failed to scale to zero", e); } } // Scale up for morning peak @Scheduled(cron = "0 0 6 * * ?") // 6 AM daily public void scaleUpForMorning() { try { updateScaledObjectThreshold("order-processor", "kafka", "lagThreshold", 100); logger.info("Scaled up for morning peak hours"); } catch (Exception e) { logger.error("Failed to scale up for morning", e); } } }

6. HTTP Scaler for Custom Metrics

import org.springframework.web.bind.annotation.*; import org.springframework.http.ResponseEntity; import org.springframework.boot.actuate.endpoint.web.annotation.RestControllerEndpoint; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; @RestController @RequestMapping("/api/scaling") public class ScalingMetricsController { private final MetricsService metricsService; private final AtomicInteger customWorkload = new AtomicInteger(0); public ScalingMetricsController(MetricsService metricsService) { this.metricsService = metricsService; } @GetMapping("/metrics") public ResponseEntity<Map<String, Object>> getScalingMetrics() { Map<String, Object> metrics = Map.of( "queueSize", metricsService.getCurrentQueueSize(), "processingTime", metricsService.getAverageProcessingTime(), "activeConnections", metricsService.getActiveConnections(), "customWorkload", customWorkload.get(), "timestamp", System.currentTimeMillis() ); return ResponseEntity.ok(metrics); } @PostMapping("/workload") public ResponseEntity<String> updateWorkload(@RequestBody WorkloadRequest request) { customWorkload.set(request.getWorkload()); metricsService.setCurrentQueueSize(request.getWorkload()); return ResponseEntity.ok("Workload updated to: " + request.getWorkload()); } @GetMapping("/health") public ResponseEntity<Map<String, String>> health() { return ResponseEntity.ok(Map.of("status", "healthy", "service", "scaling-metrics")); } public static class WorkloadRequest { private int workload; public int getWorkload() { return workload; } public void setWorkload(int workload) { this.workload = workload; } } }

7. KEDA Configuration Factory

import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import java.util.HashMap; import java.util.Map; @Component public class KedaConfigFactory { @Value("${kafka.bootstrap.servers:localhost:9092}") private String kafkaBootstrapServers; @Value("${kafka.consumer.group:order-processor-group}") private String kafkaConsumerGroup; @Value("${rabbitmq.host:localhost}") private String rabbitMQHost; @Value("${redis.address:localhost:6379}") private String redisAddress; public KedaConfigGenerator.ScaledObjectConfig createOrderProcessorConfig() { return new KedaConfigGenerator.ScaledObjectConfig() .withName("order-processor-scaledobject") .withNamespace("default") .withTarget("Deployment", "order-processor") .withReplicaBounds(0, 10) .withPollingInterval(15) .withCooldownPeriod(60) .withTrigger(KedaConfigGenerator.TriggerConfig.kafka( kafkaBootstrapServers, "orders", kafkaConsumerGroup, 50 // Scale when lag reaches 50 messages )) .withAdvanced(createAdvancedConfig()); } public KedaConfigGenerator.ScaledObjectConfig createPaymentProcessorConfig() { return new KedaConfigGenerator.ScaledObjectConfig() .withName("payment-processor-scaledobject") .withNamespace("default") .withTarget("Deployment", "payment-processor") .withReplicaBounds(1, 5) // Always keep at least 1 replica .withPollingInterval(30) .withTrigger(KedaConfigGenerator.TriggerConfig.rabbitMQ( rabbitMQHost, "payment-queue", "/", 20 // Scale when queue has 20 messages )); } public KedaConfigGenerator.ScaledObjectConfig createNotificationServiceConfig() { return new KedaConfigGenerator.ScaledObjectConfig() .withName("notification-service-scaledobject") .withNamespace("default") .withTarget("Deployment", "notification-service") .withReplicaBounds(0, 3) .withTrigger(KedaConfigGenerator.TriggerConfig.redis( redisAddress, "notification-queue", 100, // Scale when list has 100 items "list" )); } public KedaConfigGenerator.ScaledObjectConfig createCronScaledConfig() { return new KedaConfigGenerator.ScaledObjectConfig() .withName("batch-job-scaledobject") .withNamespace("default") .withTarget("Job", "batch-processing-job") .withReplicaBounds(0, 1) .withTrigger(KedaConfigGenerator.TriggerConfig.cron("1", "UTC")); } public KedaConfigGenerator.ScaledObjectConfig createHttpScaledConfig(String url) { return new KedaConfigGenerator.ScaledObjectConfig() .withName("http-scaledobject") .withNamespace("default") .withTarget("Deployment", "http-service") .withReplicaBounds(0, 5) .withTrigger(KedaConfigGenerator.TriggerConfig.http( url, "queueSize", 10 // Scale when queueSize > 10 )); } private Map<String, Object> createAdvancedConfig() { Map<String, Object> advanced = new HashMap<>(); Map<String, Object> horizontalPodAutoscalerConfig = new HashMap<>(); Map<String, Object> behavior = new HashMap<>(); Map<String, Object> scaleDown = new HashMap<>(); scaleDown.put("stabilizationWindowSeconds", 300); scaleDown.put("policies", Arrays.asList( Map.of("type", "Pods", "value", 1, "periodSeconds", 60) )); behavior.put("scaleDown", scaleDown); horizontalPodAutoscalerConfig.put("behavior", behavior); advanced.put("horizontalPodAutoscalerConfig", horizontalPodAutoscalerConfig); return advanced; } }

8. Spring Boot Application Configuration

import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableScheduling public class KedaJavaApplication { public static void main(String[] args) { SpringApplication.run(KedaJavaApplication.class, args); } @Bean public CommandLineRunner setupKeda(KedaScalingManager scalingManager, KedaConfigFactory configFactory) { return args -> { // Register scaled objects scalingManager.registerScaledObject("order-processor", configFactory.createOrderProcessorConfig()); scalingManager.registerScaledObject("payment-processor", configFactory.createPaymentProcessorConfig()); scalingManager.registerScaledObject("notification-service", configFactory.createNotificationServiceConfig()); // Deploy scaled objects scalingManager.deployScaledObject("order-processor"); scalingManager.deployScaledObject("payment-processor"); scalingManager.deployScaledObject("notification-service"); }; } @Bean public KedaConfigGenerator kedaConfigGenerator() { return new KedaConfigGenerator(); } }

9. Application Properties

# application.properties # Kafka Configuration kafka.bootstrap.servers=localhost:9092 kafka.topic.orders=orders kafka.topic.payments=payments kafka.consumer.group=order-processor-group # RabbitMQ Configuration rabbitmq.host=localhost rabbitmq.queue.payments=payment-queue # Redis Configuration redis.address=localhost:6379 redis.queue.notifications=notification-queue # KEDA Configuration keda.polling.interval=30 keda.cooldown.period=300 # Application Configuration server.port=8080 management.endpoints.web.exposure.include=health,metrics,info # Logging logging.level.com.example.keda=DEBUG

10. Kubernetes Deployment with KEDA

# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: order-processor namespace: default spec: replicas: 0 # Start with 0, let KEDA scale selector: matchLabels: app: order-processor template: metadata: labels: app: order-processor spec: containers: - name: order-processor image: my-registry/order-processor:latest ports: - containerPort: 8080 env: - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka-broker:9092" - name: SPRING_PROFILES_ACTIVE value: "kubernetes" resources: requests: memory: "128Mi" cpu: "100m" limits: memory: "512Mi" cpu: "500m" --- # Service for metrics apiVersion: v1 kind: Service metadata: name: order-processor-service namespace: default spec: selector: app: order-processor ports: - port: 8080 targetPort: 8080

11. Usage Example

import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RestController; @RestController public class OrderController { private final MetricsService metricsService; private final EventDrivenProcessor eventProcessor; public OrderController(MetricsService metricsService, EventDrivenProcessor eventProcessor) { this.metricsService = metricsService; this.eventProcessor = eventProcessor; } @PostMapping("/api/orders") public String createOrder(@RequestBody OrderRequest request) { metricsService.incrementQueueSize(); try { // Process order (in real scenario, this would send to Kafka) eventProcessor.processOrderEvent(request.toJson()); return "Order accepted"; } finally { metricsService.decrementQueueSize(); } } @PostMapping("/api/orders/batch") public String createOrdersBatch(@RequestBody List<OrderRequest> requests) { metricsService.setCurrentQueueSize(requests.size()); // In real scenario, these would be sent to Kafka // KEDA would automatically scale based on queue size return "Batch processing started"; } public static class OrderRequest { private String customerId; private List<OrderItem> items; public String toJson() { // Convert to JSON return "{}"; } // Getters and setters public String getCustomerId() { return customerId; } public void setCustomerId(String customerId) { this.customerId = customerId; } public List<OrderItem> getItems() { return items; } public void setItems(List<OrderItem> items) { this.items = items; } } public static class OrderItem { private String productId; private int quantity; private double price; // Getters and setters public String getProductId() { return productId; } public void setProductId(String productId) { this.productId = productId; } public int getQuantity() { return quantity; } public void setQuantity(int quantity) { this.quantity = quantity; } public double getPrice() { return price; } public void setPrice(double price) { this.price = price; } } }

This comprehensive KEDA