Spring 15: 메시징/Kafka

파티션 구조 시각화

파티션 키는 메시지가 어느 파티션으로 전송될지 결정하는 값입니다. 동일한 키를 가진 메시지는 항상 같은 파티션으로 전송되어 순서가 보장됩니다.

Offset 구조

<!-- Maven -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- Gradle -->
implementation 'org.springframework.kafka:spring-kafka'

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      retries: 3
    consumer:
      group-id: my-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest
      enable-auto-commit: false

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");
        configProps.put(ProducerConfig.RETRIES_CONFIG, 3);
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        configProps.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        configProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

@Bean
public ProducerFactory<String, Object> jsonProducerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    configProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
    
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, Object> jsonKafkaTemplate() {
    return new KafkaTemplate<>(jsonProducerFactory());
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
    
    return new DefaultKafkaConsumerFactory<>(props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(3); // 동시 처리 스레드 수
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    return factory;
}

@Component
public class MessageListener {

    @KafkaListener(topics = "user-events", groupId = "user-service")
    public void listen(String message) {
        log.info("Received message: {}", message);
        // 메시지 처리 로직
    }

    @KafkaListener(topics = "order-events", groupId = "order-service")
    public void listenWithHeaders(
            @Payload String message,
            @Header("eventType") String eventType,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
            @Header(KafkaHeaders.OFFSET) long offset) {
        
        log.info("Received message: {} from topic: {}, partition: {}, offset: {}", 
                message, topic, partition, offset);
    }
}

@Service
public class MessageProducer {
    
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    public MessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    // 단순 전송
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
    
    // 키와 함께 전송
    public void sendMessage(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message);
    }
    
    // 파티션 지정 전송
    public void sendMessage(String topic, int partition, String key, String message) {
        kafkaTemplate.send(topic, partition, key, message);
    }
}

public void sendMessageWithHeaders(String topic, String key, String message) {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, message);
    
    // 헤더 추가
    record.headers().add("eventType", "USER_CREATED".getBytes());
    record.headers().add("version", "1.0".getBytes());
    record.headers().add("timestamp", String.valueOf(System.currentTimeMillis()).getBytes());
    
    kafkaTemplate.send(record);
}

// JSON 객체 전송
public void sendJsonMessage(String topic, String key, Object payload) {
    ProducerRecord<String, Object> record = new ProducerRecord<>(topic, key, payload);
    record.headers().add("contentType", "application/json".getBytes());
    
    jsonKafkaTemplate.send(record);
}

public void sendMessageWithCallback(String topic, String key, String message) {
    ListenableFuture<SendResult<String, String>> future = 
        kafkaTemplate.send(topic, key, message);
    
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            RecordMetadata metadata = result.getRecordMetadata();
            log.info("Message sent successfully to topic: {}, partition: {}, offset: {}", 
                    metadata.topic(), metadata.partition(), metadata.offset());
        }
        
        @Override
        public void onFailure(Throwable ex) {
            log.error("Failed to send message to topic: {}", topic, ex);
            // 실패 처리 로직 (재시도, 알림 등)
        }
    });
}

public CompletableFuture<SendResult<String, String>> sendMessageAsync(
        String topic, String key, String message) {
    
    return kafkaTemplate.send(topic, key, message)
        .completable()
        .whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("Failed to send message", ex);
                // 메트릭 업데이트, 알림 등
                updateFailureMetrics(topic);
            } else {
                log.info("Message sent: {}", result.getRecordMetadata());
                updateSuccessMetrics(topic);
            }
        });
}

// 배치 전송
public List<CompletableFuture<SendResult<String, String>>> sendBatchMessages(
        String topic, List<MessageDto> messages) {
    
    return messages.stream()
        .map(msg -> sendMessageAsync(topic, msg.getKey(), msg.getValue()))
        .collect(Collectors.toList());
}

@Configuration
public class KafkaTransactionConfig {
    
    @Bean
    public ProducerFactory<String, String> transactionalProducerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        
        // 트랜잭션 설정
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-producer-1");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
        
        return new DefaultKafkaProducerFactory<>(props);
    }
    
    @Bean
    public KafkaTemplate<String, String> transactionalKafkaTemplate() {
        return new KafkaTemplate<>(transactionalProducerFactory());
    }
}

@Service
@Transactional
public class TransactionalMessageService {
    
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final UserRepository userRepository;
    
    // 데이터베이스와 Kafka 트랜잭션 연동
    @KafkaTransactionManager
    public void createUserWithEvent(UserDto userDto) {
        // 1. 데이터베이스 저장
        User user = userRepository.save(new User(userDto));
        
        // 2. Kafka 메시지 전송 (같은 트랜잭션)
        kafkaTemplate.send("user-events", user.getId(), 
            createUserEvent(user));
        
        // 3. 추가 이벤트 전송
        kafkaTemplate.send("audit-events", user.getId(), 
            createAuditEvent("USER_CREATED", user));
    }
    
    // 수동 트랜잭션 관리
    public void sendMessagesInTransaction(List<MessageDto> messages) {
        kafkaTemplate.executeInTransaction(template -> {
            for (MessageDto msg : messages) {
                template.send(msg.getTopic(), msg.getKey(), msg.getValue());
            }
            return null;
        });
    }
}

public class UserEventSerializer implements Serializer<UserEvent> {
    
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public byte[] serialize(String topic, UserEvent data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (JsonProcessingException e) {
            throw new SerializationException("Error serializing UserEvent", e);
        }
    }
    
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 설정 초기화
    }
    
    @Override
    public void close() {
        // 리소스 정리
    }
}

// Avro 직렬화 예시
public class AvroUserEventSerializer implements Serializer<UserEvent> {
    
    private final SpecificDatumWriter<UserEvent> writer = 
        new SpecificDatumWriter<>(UserEvent.class);
    
    @Override
    public byte[] serialize(String topic, UserEvent data) {
        try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
            BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
            writer.write(data, encoder);
            encoder.flush();
            return out.toByteArray();
        } catch (IOException e) {
            throw new SerializationException("Error serializing Avro UserEvent", e);
        }
    }
}

@Component
public class UserEventConsumer {
    
    private static final Logger log = LoggerFactory.getLogger(UserEventConsumer.class);
    
    @KafkaListener(topics = "user-events", groupId = "user-service")
    public void handleUserEvent(String message) {
        log.info("Received user event: {}", message);
        // 메시지 처리 로직
        processUserEvent(message);
    }
    
    @KafkaListener(topics = "order-events", groupId = "order-service")
    public void handleOrderEvent(
            @Payload String message,
            @Header("eventType") String eventType,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {
        
        try {
            log.info("Processing {} event from {}:{} at offset {}", 
                    eventType, topic, partition, offset);
            
            processOrderEvent(message, eventType);
            
            // 수동 커밋
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("Failed to process order event", e);
            // 에러 처리 로직
        }
    }
}

@KafkaListener(topics = "batch-events", 
               groupId = "batch-processor",
               containerFactory = "batchKafkaListenerContainerFactory")
public void handleBatchEvents(List<ConsumerRecord<String, String>> records,
                             Acknowledgment ack) {
    
    log.info("Received batch of {} messages", records.size());
    
    try {
        List<EventDto> events = records.stream()
            .map(record -> parseEvent(record.value()))
            .collect(Collectors.toList());
        
        // 배치 처리
        processBatchEvents(events);
        
        // 배치 커밋
        ack.acknowledge();
        
    } catch (Exception e) {
        log.error("Failed to process batch events", e);
        // 배치 실패 처리
    }
}

// JSON 역직렬화
@KafkaListener(topics = "json-events", groupId = "json-processor")
public void handleJsonEvent(@Payload UserEvent event,
                           ConsumerRecord<String, UserEvent> record) {
    
    log.info("Received user event: {} for user: {}", 
            event.getEventType(), event.getUserId());
    
    // 타입 안전한 처리
    switch (event.getEventType()) {
        case USER_CREATED:
            handleUserCreated(event);
            break;
        case USER_UPDATED:
            handleUserUpdated(event);
            break;
        case USER_DELETED:
            handleUserDeleted(event);
            break;
    }
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> 
    kafkaListenerContainerFactory() {
    
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // 에러 핸들러 설정
    factory.setCommonErrorHandler(new DefaultErrorHandler(
        new FixedBackOff(1000L, 3L))); // 1초 간격으로 3번 재시도
    
    return factory;
}

// 커스텀 에러 핸들러
@Component
public class KafkaErrorHandler implements CommonErrorHandler {
    
    @Override
    public void handleOtherException(Exception thrownException, 
                                   Consumer<?, ?> consumer, 
                                   MessageListenerContainer container, 
                                   boolean batchListener) {
        
        log.error("Kafka consumer error", thrownException);
        
        // 메트릭 업데이트
        updateErrorMetrics(thrownException);
        
        // 알림 발송
        sendErrorNotification(thrownException);
    }
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> 
    retryKafkaListenerContainerFactory() {
    
    ConcurrentKafkaListenerContainerFactory<String, String> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // 지수 백오프 재시도
    ExponentialBackOff backOff = new ExponentialBackOff();
    backOff.setInitialInterval(1000L);      // 1초
    backOff.setMultiplier(2.0);             // 2배씩 증가
    backOff.setMaxInterval(10000L);         // 최대 10초
    backOff.setMaxElapsedTime(60000L);      // 최대 1분
    
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
        (record, exception) -> {
            // 최종 실패 시 처리
            log.error("Final retry failed for record: {}", record, exception);
            sendToDeadLetterQueue(record, exception);
        }, backOff);
    
    // 재시도하지 않을 예외 설정
    errorHandler.addNotRetryableExceptions(ValidationException.class);
    
    factory.setCommonErrorHandler(errorHandler);
    return factory;
}

// 조건부 재시도
@RetryableTopic(
    attempts = "3",
    backoff = @Backoff(delay = 1000, multiplier = 2.0),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    include = {TransientException.class}
)
@KafkaListener(topics = "retry-events", groupId = "retry-processor")
public void handleRetryableEvent(String message) {
    log.info("Processing retryable event: {}", message);
    
    if (shouldFail()) {
        throw new TransientException("Temporary failure");
    }
    
    processEvent(message);
}

@Component
public class RetryableMessageProcessor {
    
    private final RetryTemplate retryTemplate;
    
    public RetryableMessageProcessor() {
        this.retryTemplate = RetryTemplate.builder()
            .maxAttempts(3)
            .exponentialBackoff(1000, 2, 10000)
            .retryOn(TransientException.class)
            .build();
    }
    
    @KafkaListener(topics = "custom-retry-events", groupId = "custom-retry")
    public void handleMessage(String message, Acknowledgment ack) {
        try {
            retryTemplate.execute(context -> {
                log.info("Attempt {} for message: {}", 
                        context.getRetryCount() + 1, message);
                
                processMessage(message);
                return null;
            });
            
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("All retry attempts failed for message: {}", message, e);
            handleFinalFailure(message, e);
            ack.acknowledge();
        }
    }
    
    private void handleFinalFailure(String message, Exception e) {
        // DLQ 전송
        kafkaTemplate.send("failed-events-dlq", message);
        
        // 알림 발송
        notificationService.sendFailureAlert(message, e);
        
        // 메트릭 업데이트
        meterRegistry.counter("kafka.consumer.final_failure").increment();
    }
}

// 도메인 이벤트 정의
public abstract class DomainEvent {
    private final String eventId;
    private final LocalDateTime occurredAt;
    private final String aggregateId;
    private final String eventType;
    
    protected DomainEvent(String aggregateId, String eventType) {
        this.eventId = UUID.randomUUID().toString();
        this.occurredAt = LocalDateTime.now();
        this.aggregateId = aggregateId;
        this.eventType = eventType;
    }
}

// 구체적인 이벤트
public class UserCreatedEvent extends DomainEvent {
    private final String userId;
    private final String email;
    private final String name;
    
    public UserCreatedEvent(String userId, String email, String name) {
        super(userId, "USER_CREATED");
        this.userId = userId;
        this.email = email;
        this.name = name;
    }
}

public class OrderPlacedEvent extends DomainEvent {
    private final String orderId;
    private final String customerId;
    private final List<OrderItem> items;
    private final BigDecimal totalAmount;
    
    public OrderPlacedEvent(String orderId, String customerId, 
                           List<OrderItem> items, BigDecimal totalAmount) {
        super(orderId, "ORDER_PLACED");
        this.orderId = orderId;
        this.customerId = customerId;
        this.items = items;
        this.totalAmount = totalAmount;
    }
}

@Component
public class DomainEventPublisher {
    
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final ObjectMapper objectMapper;
    
    public void publishEvent(DomainEvent event) {
        try {
            String topic = getTopicForEvent(event.getEventType());
            String key = event.getAggregateId();
            
            ProducerRecord<String, Object> record = 
                new ProducerRecord<>(topic, key, event);
            
            // 이벤트 메타데이터 추가
            record.headers().add("eventId", event.getEventId().getBytes());
            record.headers().add("eventType", event.getEventType().getBytes());
            record.headers().add("occurredAt", 
                               event.getOccurredAt().toString().getBytes());
            record.headers().add("aggregateId", event.getAggregateId().getBytes());
            
            kafkaTemplate.send(record)
                .addCallback(
                    result -> log.info("Event published: {}", event.getEventId()),
                    failure -> log.error("Failed to publish event: {}", 
                                       event.getEventId(), failure)
                );
                
        } catch (Exception e) {
            log.error("Error publishing event: {}", event, e);
            throw new EventPublishingException("Failed to publish event", e);
        }
    }
    
    private String getTopicForEvent(String eventType) {
        return switch (eventType) {
            case "USER_CREATED", "USER_UPDATED", "USER_DELETED" -> "user-events";
            case "ORDER_PLACED", "ORDER_CANCELLED", "ORDER_SHIPPED" -> "order-events";
            case "PAYMENT_PROCESSED", "PAYMENT_FAILED" -> "payment-events";
            default -> "domain-events";
        };
    }
}

@Service
@Transactional
public class OrderService {
    
    private final OrderRepository orderRepository;
    private final DomainEventPublisher eventPublisher;
    
    public Order createOrder(CreateOrderRequest request) {
        // 1. 주문 생성
        Order order = Order.builder()
            .customerId(request.getCustomerId())
            .items(request.getItems())
            .totalAmount(calculateTotal(request.getItems()))
            .status(OrderStatus.PENDING)
            .build();
        
        Order savedOrder = orderRepository.save(order);
        
        // 2. 도메인 이벤트 발행
        OrderPlacedEvent event = new OrderPlacedEvent(
            savedOrder.getId(),
            savedOrder.getCustomerId(),
            savedOrder.getItems(),
            savedOrder.getTotalAmount()
        );
        
        eventPublisher.publishEvent(event);
        
        return savedOrder;
    }
    
    public void cancelOrder(String orderId, String reason) {
        Order order = orderRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));
        
        order.cancel(reason);
        orderRepository.save(order);
        
        // 주문 취소 이벤트 발행
        OrderCancelledEvent event = new OrderCancelledEvent(
            orderId, reason, LocalDateTime.now()
        );
        
        eventPublisher.publishEvent(event);
    }
}

@Component
public class InventoryEventHandler {
    
    private final InventoryService inventoryService;
    private final DomainEventPublisher eventPublisher;
    
    @KafkaListener(topics = "order-events", groupId = "inventory-service")
    public void handleOrderEvent(OrderPlacedEvent event) {
        try {
            log.info("Processing inventory for order: {}", event.getOrderId());
            
            // 재고 확인 및 차감
            List<InventoryReservation> reservations = 
                inventoryService.reserveItems(event.getItems());
            
            if (reservations.stream().allMatch(InventoryReservation::isSuccessful)) {
                // 재고 확보 성공
                InventoryReservedEvent reservedEvent = new InventoryReservedEvent(
                    event.getOrderId(), reservations
                );
                eventPublisher.publishEvent(reservedEvent);
                
            } else {
                // 재고 부족
                InventoryReservationFailedEvent failedEvent = 
                    new InventoryReservationFailedEvent(
                        event.getOrderId(), 
                        getFailedItems(reservations)
                    );
                eventPublisher.publishEvent(failedEvent);
            }
            
        } catch (Exception e) {
            log.error("Failed to process inventory for order: {}", 
                     event.getOrderId(), e);
            
            InventoryProcessingFailedEvent errorEvent = 
                new InventoryProcessingFailedEvent(
                    event.getOrderId(), e.getMessage()
                );
            eventPublisher.publishEvent(errorEvent);
        }
    }
    
    @KafkaListener(topics = "order-events", groupId = "inventory-service")
    public void handleOrderCancellation(OrderCancelledEvent event) {
        log.info("Releasing inventory for cancelled order: {}", event.getOrderId());
        
        inventoryService.releaseReservation(event.getOrderId());
        
        InventoryReleasedEvent releasedEvent = new InventoryReleasedEvent(
            event.getOrderId(), LocalDateTime.now()
        );
        eventPublisher.publishEvent(releasedEvent);
    }
}

@Component
public class KafkaMetricsCollector {
    
    private final MeterRegistry meterRegistry;
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    @EventListener
    public void handleProducerMetrics(ProducerMetricsEvent event) {
        // Producer 메트릭 수집
        meterRegistry.counter("kafka.producer.records.sent",
                             "topic", event.getTopic())
                     .increment();
        
        meterRegistry.timer("kafka.producer.send.duration",
                           "topic", event.getTopic())
                     .record(event.getDuration(), TimeUnit.MILLISECONDS);
    }
    
    @EventListener
    public void handleConsumerMetrics(ConsumerMetricsEvent event) {
        // Consumer 메트릭 수집
        meterRegistry.counter("kafka.consumer.records.consumed",
                             "topic", event.getTopic(),
                             "group", event.getGroupId())
                     .increment();
        
        meterRegistry.gauge("kafka.consumer.lag",
                           Tags.of("topic", event.getTopic(),
                                  "partition", String.valueOf(event.getPartition())),
                           event.getLag());
    }
    
    @Scheduled(fixedRate = 30000) // 30초마다
    public void collectBrokerMetrics() {
        // JMX를 통한 브로커 메트릭 수집
        try {
            MBeanServer server = ManagementFactory.getPlatformMBeanServer();
            
            // Under-replicated partitions
            ObjectName objectName = new ObjectName(
                "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions");
            Integer underReplicated = (Integer) server.getAttribute(objectName, "Value");
            
            meterRegistry.gauge("kafka.broker.under_replicated_partitions", underReplicated);
            
            // Request rate
            ObjectName requestName = new ObjectName(
                "kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce");
            Double requestRate = (Double) server.getAttribute(requestName, "OneMinuteRate");
            
            meterRegistry.gauge("kafka.broker.produce_requests_per_sec", requestRate);
            
        } catch (Exception e) {
            log.error("Failed to collect broker metrics", e);
        }
    }
}