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

<!-- Для Avro сериализации -->
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.4.0</version>
</dependency>

<!-- Для JSON сериализации -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all  # Ждать подтверждения от всех реплик
      retries: 3
      batch-size: 16384
      linger-ms: 5  # Ждать 5мс для батчинга
      buffer-memory: 33554432
    consumer:
      group-id: my-consumer-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest  # Читать с начала при первом подключении
      enable-auto-commit: false  # Ручное управление offset'ами
      properties:
        spring.json.trusted.packages: "com.example.model"

@Service
public class OrderEventProducer {
    
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    public void sendOrderEvent(OrderEvent orderEvent) {
        kafkaTemplate.send("order-events", orderEvent.getOrderId(), orderEvent)
            .addCallback(
                result -> log.info("Sent message: {}", orderEvent),
                failure -> log.error("Failed to send: {}", orderEvent, failure)
            );
    }
}

@Service
public class NotificationProducer {
    
    private final KafkaTemplate<String, Object> kafkaTemplate;
    
    public CompletableFuture<SendResult<String, Object>> sendNotificationAsync(String userId, NotificationEvent event) {
        return kafkaTemplate.send("notifications", userId, event)
            .thenApply(result -> {
                log.info("Notification sent to user: {} at offset: {}", 
                    userId, result.getRecordMetadata().offset());
                return result;
            })
            .exceptionally(throwable -> {
                log.error("Failed to send notification to user: {}", userId, throwable);
                return null;
            });
    }
}

@Service
public class CriticalEventProducer {
    
    private final KafkaTemplate<String, CriticalEvent> kafkaTemplate;
    
    public void sendCriticalEvent(CriticalEvent event) {
        try {
            SendResult<String, CriticalEvent> result = kafkaTemplate
                .send("critical-events", event.getId(), event)
                .get(10, TimeUnit.SECONDS);  // Таймаут 10 секунд
            
            log.info("Critical event sent successfully: offset {}", 
                result.getRecordMetadata().offset());
        } catch (Exception e) {
            log.error("Failed to send critical event", e);
            throw new EventSendingException("Critical event not sent", e);
        }
    }
}

@Service
@Transactional
public class TransactionalProducer {
    
    private final KafkaTransactionManager transactionManager;
    private final KafkaTemplate<String, Object> kafkaTemplate;
    
    @KafkaTransactional
    public void processOrderWithEvents(Order order) {
        // Обновление в БД
        orderRepository.save(order);
        
        // Отправка событий в рамках транзакции
        kafkaTemplate.send("order-created", order.getId(), new OrderCreatedEvent(order));
        kafkaTemplate.send("inventory-update", order.getProductId(), 
            new InventoryUpdateEvent(order.getProductId(), -order.getQuantity()));
        
        // Если что-то пойдет не так, откатятся и БД, и Kafka сообщения
    }
}

public class CustomPartitioner implements Partitioner {
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, 
                        Object value, byte[] valueBytes, Cluster cluster) {
        
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        
        if (key instanceof String) {
            String stringKey = (String) key;
            // VIP пользователи идут в первую партицию
            if (stringKey.startsWith("VIP_")) {
                return 0;
            }
            // Остальные распределяются по hash
            return Math.abs(stringKey.hashCode()) % numPartitions;
        }
        
        return 0;
    }
}

@Component
public class OrderEventConsumer {
    
    @KafkaListener(topics = "order-events", groupId = "order-processing-group")
    public void handleOrderEvent(OrderEvent orderEvent, 
                                @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                @Header(KafkaHeaders.OFFSET) long offset) {
        
        log.info("Received order event: {} from topic: {}, partition: {}, offset: {}", 
            orderEvent, topic, partition, offset);
        
        try {
            orderService.processOrder(orderEvent);
        } catch (Exception e) {
            log.error("Failed to process order: {}", orderEvent.getOrderId(), e);
            // Здесь можно отправить в DLQ или retry topic
            throw e; // Повторная обработка
        }
    }
}

@Component
public class ManualAckConsumer {
    
    @KafkaListener(
        topics = "payment-events",
        containerFactory = "manualAckContainerFactory"
    )
    public void handlePaymentEvent(PaymentEvent event, Acknowledgment ack) {
        try {
            paymentService.processPayment(event);
            
            // Подтверждаем обработку только после успешного выполнения
            ack.acknowledge();
            log.info("Successfully processed payment: {}", event.getPaymentId());
            
        } catch (RetryableException e) {
            log.warn("Retryable error processing payment: {}", event.getPaymentId(), e);
            // Не подтверждаем - сообщение будет повторно обработано
        } catch (NonRetryableException e) {
            log.error("Non-retryable error: {}", event.getPaymentId(), e);
            // Подтверждаем, чтобы не зациклиться
            ack.acknowledge();
        }
    }
}

@Component
public class BatchConsumer {
    
    @KafkaListener(
        topics = "analytics-events",
        containerFactory = "batchContainerFactory"
    )
    public void handleAnalyticsBatch(List<AnalyticsEvent> events,
                                   List<Acknowledgment> acknowledgments) {
        
        log.info("Processing batch of {} analytics events", events.size());
        
        try {
            // Обработка пакета для улучшения производительности
            analyticsService.processBatch(events);
            
            // Подтверждение всего пакета
            acknowledgments.forEach(Acknowledgment::acknowledge);
            
        } catch (Exception e) {
            log.error("Failed to process analytics batch", e);
            // При ошибке можно подтвердить только часть сообщений
            handlePartialBatchFailure(events, acknowledgments, e);
        }
    }
}

@Component
public class FilteredConsumer {
    
    @KafkaListener(
        topics = "user-events",
        filter = "userEventFilter"
    )
    public void handleUserEvent(UserEvent event) {
        // Обрабатываем только отфильтрованные события
        userService.processEvent(event);
    }
}

@Component("userEventFilter")
public class UserEventFilter implements RecordFilterStrategy<String, UserEvent> {
    
    @Override
    public boolean filter(ConsumerRecord<String, UserEvent> record) {
        UserEvent event = record.value();
        // Фильтруем только события активных пользователеай
        return !event.isActiveUser();
    }
}

@Configuration
public class KafkaConsumerConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> manualAckContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setConcurrency(3); // 3 потока на партицию
        
        return factory;
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> batchContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        
        return factory;
    }
}

@Configuration
public class KafkaRetryConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> retryContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory());
        factory.setRetryTemplate(retryTemplate());
        factory.setRecoveryCallback(context -> {
            ConsumerRecord<String, Object> record = 
                (ConsumerRecord<String, Object>) context.getAttribute("record");
            
            // Отправляем в Dead Letter Queue
            kafkaTemplate.send("dlq-topic", record.key(), record.value());
            log.error("Sent to DLQ: {}", record.value());
            return null;
        });
        
        return factory;
    }
    
    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000); // 1 секунда между попытками
        retryTemplate.setBackOffPolicy(backOffPolicy);
        
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        return retryTemplate;
    }
}

// Модель данных
public class OrderEvent {
    private String orderId;
    private String customerId;
    private BigDecimal amount;
    private LocalDateTime timestamp;
    
    // Конструкторы, геттеры, сеттеры
}

// Producer конфигурация
@Bean
public ProducerFactory<String, Object> 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, JsonSerializer.class);
    
    return new DefaultKafkaProducerFactory<>(configProps);
}

// Avro schema (order-event.avsc)
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.avro",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "timestamp", "type": "long"}
  ]
}

@Bean
public ProducerFactory<String, OrderEvent> avroProducerFactory() {
    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, KafkaAvroSerializer.class);
    props.put("schema.registry.url", "http://localhost:8081");
    
    return new DefaultKafkaProducerFactory<>(props);
}

management:
  endpoints:
    web:
      exposure:
        include: health, metrics, kafka
  endpoint:
    health:
      show-details: always
    kafka:
      enabled: true

@Component
public class KafkaMetrics {
    
    private final Counter messagesProduced;
    private final Counter messagesConsumed;
    private final Timer processingTime;
    
    public KafkaMetrics(MeterRegistry meterRegistry) {
        this.messagesProduced = Counter.builder("kafka.messages.produced")
            .description("Number of messages produced")
            .register(meterRegistry);
            
        this.messagesConsumed = Counter.builder("kafka.messages.consumed")
            .description("Number of messages consumed")
            .register(meterRegistry);
            
        this.processingTime = Timer.builder("kafka.processing.time")
            .description("Message processing time")
            .register(meterRegistry);
    }
    
    @EventListener
    public void handleProducerEvent(ProducerEvent event) {
        messagesProduced.increment(
            Tags.of("topic", event.getTopic(), "status", "success")
        );
    }
    
    public void recordProcessingTime(String topic, long timeMs) {
        processingTime.record(timeMs, TimeUnit.MILLISECONDS, 
            Tags.of("topic", topic));
    }
}

@Component
public class KafkaHealthIndicator implements HealthIndicator {
    
    private final KafkaTemplate<String, Object> kafkaTemplate;
    
    @Override
    public Health health() {
        try {
            // Проверяем доступность Kafka
            kafkaTemplate.partitionsFor("health-check-topic");
            return Health.up()
                .withDetail("kafka", "Available")
                .build();
        } catch (Exception e) {
            return Health.down()
                .withDetail("kafka", "Unavailable")
                .withDetail("error", e.getMessage())
                .build();
        }
    }
}

@Component
public class OrderSagaOrchestrator {
    
    @KafkaListener(topics = "order-started")
    public void handleOrderStarted(OrderStartedEvent event) {
        try {
            // Шаг 1: Резервирование инвентаря
            kafkaTemplate.send("inventory-reserve", event.getOrderId(), 
                new ReserveInventoryCommand(event));
        } catch (Exception e) {
            // Компенсация
            kafkaTemplate.send("order-failed", event.getOrderId(), 
                new OrderFailedEvent(event.getOrderId(), e.getMessage()));
        }
    }
    
    @KafkaListener(topics = "inventory-reserved")
    public void handleInventoryReserved(InventoryReservedEvent event) {
        // Шаг 2: Обработка платежа
        kafkaTemplate.send("payment-process", event.getOrderId(),
            new ProcessPaymentCommand(event));
    }
    
    @KafkaListener(topics = "payment-failed")
    public void handlePaymentFailed(PaymentFailedEvent event) {
        // Компенсация: отмена резервирования
        kafkaTemplate.send("inventory-release", event.getOrderId(),
            new ReleaseInventoryCommand(event));
    }
}

@Component
public class EventStore {
    
    @KafkaListener(topics = "domain-events")
    public void storeEvent(@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                          @Header(KafkaHeaders.OFFSET) long offset,
                          DomainEvent event) {
        
        // Сохранение события в Event Store
        EventRecord record = EventRecord.builder()
            .aggregateId(event.getAggregateId())
            .eventType(event.getClass().getSimpleName())
            .eventData(serializeEvent(event))
            .version(event.getVersion())
            .timestamp(event.getTimestamp())
            .streamOffset(offset)
            .build();
            
        eventRepository.save(record);
    }
    
    public List<DomainEvent> getEventsForAggregate(String aggregateId) {
        return eventRepository.findByAggregateIdOrderByVersion(aggregateId)
            .stream()
            .map(this::deserializeEvent)
            .collect(Collectors.toList());
    }
}

@Component
public class ProductProjectionHandler {
    
    @KafkaListener(topics = "product-events")
    public void handleProductEvent(ProductEvent event) {
        switch (event.getEventType()) {
            case "ProductCreated":
                createProductProjection((ProductCreatedEvent) event);
                break;
            case "ProductUpdated":
                updateProductProjection((ProductUpdatedEvent) event);
                break;
            case "ProductDeleted":
                deleteProductProjection((ProductDeletedEvent) event);
                break;
        }
    }
    
    private void createProductProjection(ProductCreatedEvent event) {
        ProductProjection projection = ProductProjection.builder()
            .id(event.getProductId())
            .name(event.getName())
            .price(event.getPrice())
            .category(event.getCategory())
            .lastUpdated(event.getTimestamp())
            .build();
            
        productProjectionRepository.save(projection);
    }
}

spring:
  kafka:
    ssl:
      trust-store-location: classpath:kafka.client.truststore.jks
      trust-store-password: password
      key-store-location: classpath:kafka.client.keystore.jks
      key-store-password: password
      key-password: password
    properties:
      security.protocol: SSL

spring:
  kafka:
    properties:
      security.protocol: SASL_SSL
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="password";

// Пример настройки ACL через AdminClient
@Bean
public NewTopic createTopicWithAcl() {
    NewTopic topic = TopicBuilder.name("secure-topic")
        .partitions(3)
        .replicas(2)
        .build();
    
    // ACL настраивается на уровне брокера
    return topic;
}

spring:
  kafka:
    producer:
      # Батчинг для увеличения throughput
      batch-size: 65536  # 64KB
      linger-ms: 10      # Ждать 10мс для заполнения батча
      buffer-memory: 67108864  # 64MB буфер
      
      # Компрессия для экономии сети
      compression-type: snappy
      
      # Производительность vs надежность
      acks: 1  # Ждать подтверждения только от лидера
      retries: 0  # Без повторов для максимальной скорости

spring:
  kafka:
    consumer:
      # Размер fetch для увеличения throughput
      fetch-min-size: 1048576  # 1MB
      fetch-max-wait: 500      # Максимум 500мс ожидания
      
      # Обработка в память
      max-poll-records: 1000   # Читать до 1000 записей за раз
      
      # Параллелизм
    listener:
      concurrency: 10  # 10 потоков на слушатель

@Configuration
public class KafkaPerformanceConfig {
    
    @Bean
    public ProducerFactory<String, Object> cachedProducerFactory() {
        Map<String, Object> props = new HashMap<>();
        // ... базовая конфигурация
        
        // Включение кеширования соединений
        props.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, 300000);
        
        return new DefaultKafkaProducerFactory<>(props);
    }
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> optimizedContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory());
        
        // Настройка пула потоков
        factory.getContainerProperties().setConsumerTaskExecutor(
            new SimpleAsyncTaskExecutor("kafka-consumer-")
        );
        
        return factory;
    }
}

// Структурированное именование
public class TopicNames {
    public static final String ORDER_EVENTS = "orders.events.v1";
    public static final String USER_PROFILE_UPDATES = "users.profile.updates.v1";
    public static final String PAYMENT_COMMANDS = "payments.commands.v1";
    
    // DLQ топики
    public static final String ORDER_EVENTS_DLQ = "orders.events.v1.dlq";
}

public abstract class BaseEvent {
    @JsonProperty("eventVersion")
    private String eventVersion = "1.0";
    
    @JsonProperty("eventType")
    private String eventType;
    
    @JsonProperty("timestamp")
    private LocalDateTime timestamp = LocalDateTime.now();
    
    // Базовые поля для всех событий
}

public class OrderCreatedEvent extends BaseEvent {
    public OrderCreatedEvent() {
        setEventType("OrderCreated");
        setEventVersion("1.1"); // Новая версия
    }
    
    // Поля события
}

@Component
public class IdempotentConsumer {
    
    private final RedisTemplate<String, String> redisTemplate;
    
    @KafkaListener(topics = "payment-events")
    public void handlePayment(PaymentEvent event,
                             @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                             @Header(KafkaHeaders.OFFSET) long offset) {
        
        String idempotencyKey = generateKey(topic, offset, event.getPaymentId());
        
        // Проверка, не обработано ли уже
        if (redisTemplate.hasKey(idempotencyKey)) {
            log.info("Payment already processed: {}", event.getPaymentId());
            return;
        }
        
        try {
            paymentService.processPayment(event);
            
            // Сохранение ключа идемпотентности
            redisTemplate.opsForValue().set(idempotencyKey, "processed", 
                Duration.ofHours(24));
                
        } catch (Exception e) {
            log.error("Payment processing failed: {}", event.getPaymentId(), e);
            throw e;
        }
    }
}

@Component
public class KafkaGracefulShutdown {
    
    private final KafkaListenerEndpointRegistry registry;
    
    @PreDestroy
    public void shutdown() {
        registry.getListenerContainers().forEach(container -> {
            container.pause();
            container.stop();
        });
        
        log.info("All Kafka consumers stopped gracefully");
    }
}

/var/kafka-logs/
├── orders-0/
│   ├── 00000000000000000000.log    # Segment file
│   ├── 00000000000000000000.index  # Offset index
│   ├── 00000000000000000000.timeindex # Time index
│   └── leader-epoch-checkpoint     # Leader epoch info
├── orders-1/
└── orders-2/

# Увеличение file descriptor limits
fs.file-max = 2097152
# TCP buffer sizes
net.core.rmem_max = 134217728
net.core.wmem_max = 134217728
# Disable swapping
vm.swappiness = 1

# G1GC рекомендуется для Kafka
-XX:+UseG1GC
-XX:MaxGCPauseMillis=20
-XX:InitiatingHeapOccupancyPercent=35

# Network threads
num.network.threads=8
# I/O threads  
num.io.threads=16
# Socket buffer sizes
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
# Log settings
log.segment.bytes=1073741824
log.retention.hours=168

# Batching for throughput
batch.size=65536
linger.ms=10
compression.type=snappy
# Memory allocation
buffer.memory=67108864

# Fetch settings
fetch.min.bytes=50000
fetch.max.wait.ms=500
# Processing parallelism  
max.poll.records=2000

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<!-- Для JSON сериализации -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 60000
    publisher-confirms: true
    publisher-returns: true
    template:
      mandatory: true
      retry:
        enabled: true
        initial-interval: 1000
        max-attempts: 3
        multiplier: 2
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 10
        retry:
          enabled: true
          max-attempts: 3

@Configuration
public class RabbitConfig {

    // Direct exchange для заказов
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange", true, false);
    }

    // Topic exchange для уведомлений  
    @Bean
    public TopicExchange notificationExchange() {
        return new TopicExchange("notification.exchange", true, false);
    }

    // Fanout exchange для логирования
    @Bean
    public FanoutExchange logExchange() {
        return new FanoutExchange("log.exchange", true, false);
    }

    // Очереди
    @Bean
    public Queue orderProcessingQueue() {
        return QueueBuilder.durable("order.processing")
            .withArgument("x-dead-letter-exchange", "order.dlx")
            .withArgument("x-message-ttl", 300000) // 5 минут TTL
            .build();
    }

    @Bean
    public Queue orderDlqQueue() {
        return QueueBuilder.durable("order.dlq").build();
    }

    // Привязки (Bindings)
    @Bean
    public Binding orderProcessingBinding() {
        return BindingBuilder
            .bind(orderProcessingQueue())
            .to(orderExchange())
            .with("order.process");
    }

    @Bean
    public Binding notificationBinding() {
        return BindingBuilder
            .bind(notificationQueue())
            .to(notificationExchange())
            .with("notification.*.email");
    }
}

@Bean
public DirectExchange deadLetterExchange() {
    return new DirectExchange("order.dlx", true, false);
}

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("order.dlq").build();
}

@Bean
public Binding deadLetterBinding() {
    return BindingBuilder
        .bind(deadLetterQueue())
        .to(deadLetterExchange())
        .with("order.failed");
}

@Service
public class OrderService {

    private final RabbitTemplate rabbitTemplate;

    public void createOrder(Order order) {
        // Простая отправка
        rabbitTemplate.convertAndSend("order.exchange", "order.process", order);
        
        // С дополнительными свойствами
        rabbitTemplate.convertAndSend(
            "order.exchange", 
            "order.process", 
            order,
            message -> {
                message.getMessageProperties().setContentType("application/json");
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                message.getMessageProperties().setExpiration("60000"); // TTL 1 минута
                return message;
            }
        );
    }
}

@Service
public class ReliableMessageService {

    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void setup() {
        // Callback для подтверждений
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("Message delivered successfully: {}", correlationData);
            } else {
                log.error("Message delivery failed: {}, cause: {}", correlationData, cause);
            }
        });

        // Callback для returned сообщений
        rabbitTemplate.setReturnsCallback(returned -> {
            log.error("Message returned: {}, reply: {}", 
                returned.getMessage(), returned.getReplyText());
        });
    }

    public void sendMessage(Object message) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        
        rabbitTemplate.convertAndSend(
            "order.exchange",
            "order.process", 
            message,
            correlationData
        );
    }
}

@Service
@Transactional
public class TransactionalOrderService {

    @RabbitTransactional
    public void processOrderWithNotification(Order order) {
        // Сохранение в БД
        orderRepository.save(order);
        
        // Отправка уведомления (в рамках транзакции)
        rabbitTemplate.convertAndSend("notification.exchange", 
            "notification.order.created", 
            new OrderCreatedEvent(order));
            
        // Если произойдет исключение, откатятся и БД, и сообщение
    }
}

@Component
public class OrderProcessingConsumer {

    @RabbitListener(queues = "order.processing")
    public void handleOrderProcessing(Order order, 
                                    @Header Map<String, Object> headers,
                                    Channel channel,
                                    @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        
        try {
            log.info("Processing order: {}", order.getId());
            orderService.processOrder(order);
            
            // Ручное подтверждение успешной обработки
            channel.basicAck(deliveryTag, false);
            
        } catch (Exception e) {
            log.error("Failed to process order: {}", order.getId(), e);
            
            try {
                // Отклонение с отправкой в DLQ
                channel.basicReject(deliveryTag, false);
            } catch (IOException ioException) {
                log.error("Failed to reject message", ioException);
            }
        }
    }
}

@Component
public class RetryableConsumer {

    @RabbitListener(queues = "payment.processing")
    @Retryable(
        value = {RetryableException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void processPayment(PaymentRequest request) throws Exception {
        try {
            paymentService.processPayment(request);
        } catch (TemporaryServiceException e) {
            throw new RetryableException("Payment service temporarily unavailable", e);
        } catch (InvalidDataException e) {
            // Не retry для permanent ошибок
            log.error("Invalid payment data: {}", request, e);
        }
    }

    @Recover
    public void recoverPayment(RetryableException ex, PaymentRequest request) {
        log.error("Payment processing failed after all retries: {}", request, ex);
        // Отправка в DLQ или альтернативная обработка
    }
}

@Component
public class BatchConsumer {

    @RabbitListener(queues = "analytics.events", 
                   containerFactory = "batchListenerContainerFactory")
    public void processBatch(List<AnalyticsEvent> events) {
        log.info("Processing batch of {} events", events.size());
        
        try {
            analyticsService.processBatch(events);
        } catch (Exception e) {
            log.error("Batch processing failed", e);
            // При ошибке можно обработать события по одному
            events.forEach(this::processSingleEvent);
        }
    }

    private void processSingleEvent(AnalyticsEvent event) {
        try {
            analyticsService.processEvent(event);
        } catch (Exception e) {
            log.error("Failed to process event: {}", event, e);
        }
    }
}

@Bean
public Queue priorityQueue() {
    return QueueBuilder.durable("priority.processing")
        .withArgument("x-max-priority", 10)
        .build();
}

// Отправка с приоритетом
public void sendPriorityMessage(Object message, int priority) {
    rabbitTemplate.convertAndSend("priority.exchange", "priority.key", message, 
        msg -> {
            msg.getMessageProperties().setPriority(priority);
            return msg;
        });
}

@RabbitListener(queues = "priority.processing")
public void handlePriorityMessage(Object message) {
    // Сообщения обрабатываются в порядке приоритета
    processMessage(message);
}

@Configuration
public class RabbitListenerConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        
        factory.setConnectionFactory(connectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setPrefetchCount(10);
        factory.setConcurrentConsumers(2);
        factory.setMaxConcurrentConsumers(5);
        factory.setDefaultRequeueRejected(false);
        
        // Retry policy
        factory.setRetryTemplate(retryTemplate());
        factory.setRecoveryCallback(context -> {
            Message failedMessage = (Message) context.getAttribute("message");
            // Отправка в DLQ
            rabbitTemplate.send("failed.exchange", "", failedMessage);
            return null;
        });
        
        return factory;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory batchListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        
        factory.setConnectionFactory(connectionFactory);
        factory.setBatchListener(true);
        factory.setBatchSize(50);
        factory.setConsumerBatchEnabled(true);
        
        return factory;
    }

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000);
        backOffPolicy.setMultiplier(2.0);
        backOffPolicy.setMaxInterval(10000);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        return retryTemplate;
    }
}

// Отправка с различными routing keys
rabbitTemplate.convertAndSend("notification.exchange", "notification.email.signup", emailNotification);
rabbitTemplate.convertAndSend("notification.exchange", "notification.sms.signup", smsNotification);
rabbitTemplate.convertAndSend("notification.exchange", "notification.push.order", pushNotification);

// Binding patterns
@Bean
public Binding emailNotificationsBinding() {
    return BindingBuilder
        .bind(emailQueue())
        .to(notificationExchange())
        .with("notification.email.*"); // Все email уведомления
}

@Bean
public Binding allNotificationsBinding() {
    return BindingBuilder
        .bind(auditQueue())
        .to(notificationExchange()) 
        .with("notification.#"); // Все уведомления
}

@Bean
public Binding signupNotificationsBinding() {
    return BindingBuilder
        .bind(signupQueue())
        .to(notificationExchange())
        .with("notification.*.signup"); // Все уведомления о регистрации
}

@Bean
public HeadersExchange headersExchange() {
    return new HeadersExchange("headers.exchange");
}

@Bean 
public Binding headersBinding() {
    return BindingBuilder
        .bind(processQueue())
        .to(headersExchange())
        .whereAll(Map.of("type", "order", "priority", "high"))
        .match();
}

// Отправка с заголовками
public void sendWithHeaders(Object message) {
    rabbitTemplate.convertAndSend("headers.exchange", "", message,
        msg -> {
            msg.getMessageProperties().getHeaders().put("type", "order");
            msg.getMessageProperties().getHeaders().put("priority", "high");
            msg.getMessageProperties().getHeaders().put("region", "europe");
            return msg;
        });
}

@Service
public class RabbitManagementService {

    private final RabbitAdmin rabbitAdmin;
    
    public QueueInfo getQueueInfo(String queueName) {
        Properties queueProperties = rabbitAdmin.getQueueProperties(queueName);
        if (queueProperties != null) {
            return QueueInfo.builder()
                .name(queueName)
                .messageCount((Integer) queueProperties.get("QUEUE_MESSAGE_COUNT"))
                .consumerCount((Integer) queueProperties.get("QUEUE_CONSUMER_COUNT"))
                .build();
        }
        return null;
    }

    public void purgeQueue(String queueName) {
        rabbitAdmin.purgeQueue(queueName, false);
    }

    public void createQueueDynamically(String queueName) {
        Queue queue = new Queue(queueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
    }
}

management:
  endpoints:
    web:
      exposure:
        include: health, rabbit, metrics
  endpoint:
    health:
      show-details: always
    rabbit:
      enabled: true

@Component
public class RabbitHealthIndicator implements HealthIndicator {

    private final RabbitTemplate rabbitTemplate;

    @Override
    public Health health() {
        try {
            // Проверка соединения
            rabbitTemplate.execute(channel -> {
                channel.queueDeclarePassive("health.check");
                return null;
            });
            
            return Health.up()
                .withDetail("rabbit", "Available")
                .build();
        } catch (Exception e) {
            return Health.down()
                .withDetail("rabbit", "Unavailable")
                .withDetail("error", e.getMessage())
                .build();
        }
    }
}

@Component
public class RabbitMetrics {

    private final Counter messagesProduced;
    private final Counter messagesConsumed;
    private final Timer processingTime;

    public RabbitMetrics(MeterRegistry meterRegistry) {
        this.messagesProduced = Counter.builder("rabbitmq.messages.produced")
            .description("Number of messages produced")
            .register(meterRegistry);
            
        this.messagesConsumed = Counter.builder("rabbitmq.messages.consumed")
            .description("Number of messages consumed")
            .register(meterRegistry);
            
        this.processingTime = Timer.builder("rabbitmq.processing.time")
            .description("Message processing time")
            .register(meterRegistry);
    }

    @EventListener
    public void handleMessageSent(MessageSentEvent event) {
        messagesProduced.increment(
            Tags.of("exchange", event.getExchange(), "routing-key", event.getRoutingKey())
        );
    }

    public void recordProcessingTime(String queue, long timeMs) {
        processingTime.record(timeMs, TimeUnit.MILLISECONDS, Tags.of("queue", queue));
    }
}

# Настройка политики для HA
rabbitmqctl set_policy ha-orders "^order\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

@Bean
public Queue quorumQueue() {
    return QueueBuilder.durable("order.processing.quorum")
        .withArgument("x-queue-type", "quorum")
        .withArgument("x-max-in-memory-length", 1000)
        .build();
}

# rabbitmq.conf
cluster_formation.peer_discovery_backend = classic_config
cluster_formation.classic_config.nodes.1 = rabbit@node1
cluster_formation.classic_config.nodes.2 = rabbit@node2
cluster_formation.classic_config.nodes.3 = rabbit@node3

# Политика для автоматического создания quorum queues
default_queue_type = quorum

@Configuration
public class RabbitConnectionConfig {

    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setAddresses("rabbit1:5672,rabbit2:5672,rabbit3:5672");
        factory.setUsername("user");
        factory.setPassword("password");
        
        // Настройки reconnection
        factory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true);
        factory.getRabbitConnectionFactory().setNetworkRecoveryInterval(5000);
        factory.getRabbitConnectionFactory().setTopologyRecoveryEnabled(true);
        
        return factory;
    }
}

@Service
public class RequestReplyService {

    private final RabbitTemplate rabbitTemplate;

    // Синхронный request-reply
    public String syncRequest(String request) {
        return (String) rabbitTemplate.convertSendAndReceive(
            "request.exchange", 
            "request.process", 
            request
        );
    }

    // Асинхронный request-reply
    public void asyncRequest(String request, CorrelationData correlationData) {
        rabbitTemplate.convertAndSend(
            "request.exchange",
            "request.process", 
            request,
            message -> {
                message.getMessageProperties().setReplyTo("response.queue");
                message.getMessageProperties().setCorrelationId(correlationData.getId());
                return message;
            }
        );
    }
}

@Component
public class RequestProcessor {

    @RabbitListener(queues = "request.queue")
    public void processRequest(String request, 
                              @Header(name = "replyTo", required = false) String replyTo,
                              @Header(name = "correlationId", required = false) String correlationId) {
        
        String response = businessService.processRequest(request);
        
        if (replyTo != null) {
            rabbitTemplate.convertAndSend("", replyTo, response, message -> {
                if (correlationId != null) {
                    message.getMessageProperties().setCorrelationId(correlationId);
                }
                return message;
            });
        }
    }
}

@Service 
public class EventPublisher {

    public void publishEvent(DomainEvent event) {
        // Fanout exchange отправляет во все подписанные очереди
        rabbitTemplate.convertAndSend("events.fanout", "", event);
    }
}

@Component
public class EmailSubscriber {

    @RabbitListener(queues = "events.email")
    public void handleEvent(DomainEvent event) {
        if (event instanceof UserRegisteredEvent) {
            emailService.sendWelcomeEmail((UserRegisteredEvent) event);
        }
    }
}

@Component
public class AnalyticsSubscriber {

    @RabbitListener(queues = "events.analytics")
    public void handleEvent(DomainEvent event) {
        analyticsService.trackEvent(event);
    }
}

// Несколько consumer'ов читают из одной очереди
@Component
public class OrderProcessor1 {

    @RabbitListener(queues = "order.processing", id = "processor1")
    public void processOrder(Order order) {
        log.info("Processor 1 handling order: {}", order.getId());
        orderService.processOrder(order);
    }
}

@Component  
public class OrderProcessor2 {

    @RabbitListener(queues = "order.processing", id = "processor2")
    public void processOrder(Order order) {
        log.info("Processor 2 handling order: {}", order.getId());
        orderService.processOrder(order);
    }
}

spring:
  rabbitmq:
    ssl:
      enabled: true
      key-store: classpath:client_key.p12
      key-store-password: password
      key-store-type: PKCS12
      trust-store: classpath:trust_store.jks
      trust-store-password: password
      trust-store-type: JKS
      algorithm: TLSv1.2

# Создание пользователя
rabbitmqctl add_user myuser mypassword

# Назначение прав
rabbitmqctl set_permissions -p "/" myuser "order\..*" "order\..*" "order\..*"

# Роли
rabbitmqctl set_user_tags myuser management

# Формат: configure write read
rabbitmqctl set_permissions -p "/" producer "^$" "order\..*" "^$"        # Только отправка
rabbitmqctl set_permissions -p "/" consumer "^$" "^$" "order\..*"        # Только чтение  
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"                  # Полный доступ

@Bean
public RabbitTemplate optimizedRabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    
    // Batching для увеличения throughput
    template.setUsePublisherConnection(true);
    
    // JSON converter для лучшей производительности
    template.setMessageConverter(new Jackson2JsonMessageConverter());
    
    return template;
}

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 50              # Больше prefetch для высокой нагрузки
        concurrency: 5-10         # Диапазон consumer'ов
        batch-size: 10            # Batch processing
        idle-event-interval: 60s  # Частота idle events

@Bean
public CachingConnectionFactory connectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory("localhost");
    
    // Connection caching
    factory.setConnectionCacheSize(10);
    factory.setChannelCacheSize(50);
    
    // Publisher confirms
    factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
    factory.setPublisherReturns(true);
    
    return factory;
}

# rabbitmq.conf
vm_memory_high_watermark.relative = 0.6  # 60% RAM для RabbitMQ
disk_free_limit.relative = 2.0           # 2GB свободного места
vm_memory_high_watermark_paging_ratio = 0.5  # Paging при 50% от watermark

# Настройки очередей для экономии памяти
default_queue_type = classic
queue_master_locator = min-masters

# Проверка состояния
rabbitmqctl status
rabbitmqctl environment

# Очистка memory alarm (осторожно!)
rabbitmqctl eval 'rabbit_alarm:clear_alarm({resource_limit, memory, node()}).'

# Проверка использования диска
rabbitmqctl status | grep disk
du -sh /var/lib/rabbitmq/mnesia/

# Принудительная очистка (emergency)
rabbitmqctl stop_app
rabbitmqctl reset  # ВНИМАНИЕ: удаляет все данные!
rabbitmqctl start_app

# Мониторинг соединений
rabbitmqctl list_connections
rabbitmqctl list_channels

# Закрытие зависших соединений
rabbitmqctl close_connection "<connection_name>" "Connection cleanup"

# application.yml
logging:
  level:
    org.springframework.amqp: DEBUG
    org.springframework.rabbit: DEBUG
    com.rabbitmq: DEBUG

@Component
public class RabbitDebugListener {

    @RabbitListener(queues = "debug.queue")
    public void debugMessage(Message message, Channel channel) {
        log.debug("Received message: {}", message);
        log.debug("Headers: {}", message.getMessageProperties().getHeaders());
        log.debug("Routing key: {}", message.getMessageProperties().getReceivedRoutingKey());
        log.debug("Exchange: {}", message.getMessageProperties().getReceivedExchange());
    }
}

@Component
public class RabbitPerformanceMonitor {

    @EventListener
    public void onListenerConsumerStarted(ListenerContainerConsumerStartedEvent event) {
        log.info("Consumer started: {}", event.getContainer());
    }

    @EventListener  
    public void onListenerConsumerStopped(ListenerContainerConsumerStoppedEvent event) {
        log.warn("Consumer stopped: {}", event.getContainer());
    }

    @EventListener
    public void onListenerConsumerFailed(ListenerContainerConsumerFailedEvent event) {
        log.error("Consumer failed: {}", event.getContainer(), event.getReason());
    }
}

public class NamingConventions {
    // Exchanges
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String NOTIFICATION_EXCHANGE = "notification.exchange";
    
    // Queues (включают назначение)
    public static final String ORDER_PROCESSING_QUEUE = "order.processing";
    public static final String ORDER_EMAIL_QUEUE = "order.email";
    public static final String ORDER_DLQ = "order.dlq";
    
    // Routing keys (иерархические)
    public static final String ORDER_CREATED = "order.created";
    public static final String ORDER_CANCELLED = "order.cancelled";
    public static final String NOTIFICATION_EMAIL_SIGNUP = "notification.email.signup";
}

@Component
public class IdempotentConsumer {

    private final RedisTemplate<String, String> redisTemplate;

    @RabbitListener(queues = "payment.processing")
    public void processPayment(PaymentRequest request, 
                              @Header(AmqpHeaders.MESSAGE_ID) String messageId) {
        
        String idempotencyKey = "payment:" + messageId;
        
        // Проверка на дублирование
        if (redisTemplate.hasKey(idempotencyKey)) {
            log.info("Payment already processed: {}", request.getPaymentId());
            return;
        }
        
        try {
            paymentService.processPayment(request);
            
            // Сохранение ключа идемпотентности
            redisTemplate.opsForValue().set(idempotencyKey, "processed", 
                Duration.ofHours(24));
                
        } catch (Exception e) {
            log.error("Payment processing failed: {}", request.getPaymentId(), e);
            throw e;
        }
    }
}

@Component
public class CircuitBreakerConsumer {

    private final CircuitBreaker circuitBreaker;

    @RabbitListener(queues = "external.api.calls")
    public void handleApiCall(ApiCallRequest request) {
        
        Supplier<ApiResponse> protectedCall = CircuitBreaker
            .decorateSupplier(circuitBreaker, () -> externalApiService.call(request));
            
        Try<ApiResponse> result = Try.ofSupplier(protectedCall);
        
        if (result.isSuccess()) {
            // Успешная обработка
            log.info("API call successful: {}", request.getId());
        } else {
            // Fallback или отправка в retry queue
            log.error("API call failed: {}", request.getId(), result.getCause());
            rabbitTemplate.convertAndSend("external.api.retry", request);
        }
    }
}

@Component
public class GracefulShutdown {

    private final SimpleMessageListenerContainer listenerContainer;

    @PreDestroy
    public void shutdown() {
        log.info("Shutting down RabbitMQ listeners...");
        
        // Остановка приема новых сообщений
        listenerContainer.shutdown();
        
        // Ожидание завершения обработки текущих сообщений
        try {
            if (!listenerContainer.isRunning()) {
                log.info("All messages processed, shutdown complete");
            }
        } catch (Exception e) {
            log.warn("Error during graceful shutdown", e);
        }
    }
}

public abstract class BaseMessage {
    @JsonProperty("messageVersion")
    private String messageVersion = "1.0";
    
    @JsonProperty("messageType") 
    private String messageType;
    
    @JsonProperty("timestamp")
    private LocalDateTime timestamp = LocalDateTime.now();
    
    // Геттеры и сеттеры
}

@JsonTypeName("OrderCreated")
public class OrderCreatedMessage extends BaseMessage {
    
    public OrderCreatedMessage() {
        setMessageType("OrderCreated");
        setMessageVersion("1.1"); // Обновленная версия
    }
    
    // Поля сообщения
}

@Component
public class VersionedMessageHandler {

    @RabbitListener(queues = "order.events")
    public void handleOrderEvent(BaseMessage message) {
        
        switch (message.getMessageVersion()) {
            case "1.0":
                handleV1Message(message);
                break;
            case "1.1":
                handleV2Message(message);
                break;
            default:
                log.warn("Unknown message version: {}", message.getMessageVersion());
        }
    }
}

@Component
public class DlqProcessor {

    @RabbitListener(queues = "order.dlq")
    public void processDlqMessage(Message message, Channel channel,
                                 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        
        // Анализ причины попадания в DLQ
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        
        @SuppressWarnings("unchecked")
        List<Map<String, Object>> xDeathHeader = 
            (List<Map<String, Object>>) headers.get("x-death");
            
        if (xDeathHeader != null && !xDeathHeader.isEmpty()) {
            Map<String, Object> death = xDeathHeader.get(0);
            String reason = (String) death.get("reason");
            Long count = (Long) death.get("count");
            
            log.error("Message in DLQ. Reason: {}, Count: {}, Message: {}", 
                reason, count, message);
            
            // Логика обработки в зависимости от причины
            if ("rejected".equals(reason) && count < 3) {
                // Повторная отправка в основную очередь
                reprocessMessage(message);
            } else {
                // Отправка в архив или уведомление администратора
                archiveMessage(message);
            }
        }
        
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("Failed to ack DLQ message", e);
        }
    }
}

# Prometheus метрики
management:
  metrics:
    export:
      prometheus:
        enabled: true
  endpoints:
    web:
      exposure:
        include: prometheus, health, rabbit

@Component
public class RabbitMetricsCollector {

    @Scheduled(fixedRate = 30000) // Каждые 30 секунд
    public void collectMetrics() {
        try {
            // Получение статистики очередей через Management API
            Map<String, Object> queueStats = rabbitAdmin.getQueueProperties("order.processing");
            
            if (queueStats != null) {
                int messageCount = (Integer) queueStats.get("QUEUE_MESSAGE_COUNT");
                int consumerCount = (Integer) queueStats.get("QUEUE_CONSUMER_COUNT");
                
                Metrics.gauge("rabbitmq.queue.messages", messageCount, "queue", "order.processing");
                Metrics.gauge("rabbitmq.queue.consumers", consumerCount, "queue", "order.processing");
                
                // Алерт при большой очереди
                if (messageCount > 1000) {
                    log.warn("High message count in queue: {}", messageCount);
                    alertService.sendAlert("RabbitMQ queue backlog", 
                        "Queue order.processing has " + messageCount + " messages");
                }
            }
        } catch (Exception e) {
            log.error("Failed to collect RabbitMQ metrics", e);
        }
    }
}

# Docker Compose для разработки
version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:3.11-management
    container_name: rabbitmq
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin
      RABBITMQ_DEFAULT_VHOST: /
    ports:

      - "5672:5672"   # AMQP порт
      - "15672:15672" # Management UI
    volumes:

      - rabbitmq_data:/var/lib/rabbitmq
    networks:

      - app-network

volumes:
  rabbitmq_data:

networks:
  app-network:

# Kubernetes deployment
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: rabbitmq
spec:
  serviceName: rabbitmq
  replicas: 3
  selector:
    matchLabels:
      app: rabbitmq
  template:
    metadata:
      labels:
        app: rabbitmq
    spec:
      containers:

      - name: rabbitmq
        image: rabbitmq:3.11-management
        ports:

        - containerPort: 5672
        - containerPort: 15672
        env:

        - name: RABBITMQ_ERLANG_COOKIE
          value: "mycookie"
        - name: RABBITMQ_DEFAULT_USER
          valueFrom:
            secretKeyRef:
              name: rabbitmq-secret
              key: username
        volumeMounts:

        - name: rabbitmq-data
          mountPath: /var/lib/rabbitmq
  volumeClaimTemplates:

  - metadata:
      name: rabbitmq-data
    spec:
      accessModes: ["ReadWriteOnce"]
      resources:
        requests:
          storage: 10Gi

#!/bin/bash
# Backup script

# Экспорт конфигурации
rabbitmqctl export_definitions /backup/definitions-$(date +%Y%m%d).json

# Экспорт сообщений (если нужно)
rabbitmqctl list_queues name messages | while read queue messages; do
    if [ "$messages" -gt 0 ]; then
        echo "Backing up queue: $queue with $messages messages"
        # Используйте shovel plugin для миграции сообщений
    fi
done

# Backup Mnesia database (при остановленном RabbitMQ)
# systemctl stop rabbitmq-server
# tar -czf /backup/mnesia-$(date +%Y%m%d).tar.gz /var/lib/rabbitmq/mnesia/
# systemctl start rabbitmq-server

@Component
public class CapacityMonitor {

    @Scheduled(fixedRate = 300000) // Каждые 5 минут
    public void monitorCapacity() {
        // Проверка throughput
        long currentTime = System.currentTimeMillis();
        long messagesPerMinute = getMessagesPerMinute();
        
        // Прогнозирование нагрузки
        if (messagesPerMinute > THRESHOLD_HIGH) {
            log.warn("High message throughput detected: {} messages/min", messagesPerMinute);
            
            // Автоматическое масштабирование consumer'ов
            scaleConsumers("order.processing", messagesPerMinute);
        }
        
        // Проверка размера очередей
        checkQueueSizes();
        
        // Проверка memory usage
        checkMemoryUsage();
    }
    
    private void scaleConsumers(String queueName, long throughput) {
        int currentConsumers = getCurrentConsumerCount(queueName);
        int optimalConsumers = calculateOptimalConsumers(throughput);
        
        if (optimalConsumers > currentConsumers) {
            log.info("Scaling consumers for {}: {} -> {}", 
                queueName, currentConsumers, optimalConsumers);
            // Логика масштабирования
        }
    }
}

Topic: user-events
Partition 0: [msg1][msg2][msg3][msg4] → (append-only log)
Partition 1: [msg5][msg6][msg7][msg8] →
Partition 2: [msg9][msg10][msg11]    →

Queue: order-processing
[msg1] → [msg2] → [msg3] → Consumer (msg удаляется после ack)

# Типичные показатели Kafka
Producer: 800,000 msgs/sec (100 byte messages)
Consumer: 900,000 msgs/sec
Latency: 2-5ms (end-to-end)

# Типичные показатели RabbitMQ
Producer: 20,000-50,000 msgs/sec
Consumer: 30,000-70,000 msgs/sec  
Latency: 0.1-1ms (для simple routing)

# Конфигурация для exactly-once
spring.kafka.producer:
  acks: all
  enable-idempotence: true
  transactional-id: my-transactional-id
spring.kafka.consumer:
  isolation-level: read_committed

# Конфигурация для надежности
spring.rabbitmq:
  publisher-confirms: true
  publisher-returns: true
  listener.simple.acknowledge-mode: manual

// Kafka Streams example
@Component
public class OrderAnalytics {
    
    @Autowired
    private StreamsBuilder streamsBuilder;
    
    @Bean
    public KStream<String, Order> processOrders() {
        return streamsBuilder
            .stream("orders")
            .filter((key, order) -> order.getAmount() > 100)
            .groupByKey()
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
            .aggregate(OrderStats::new, this::updateStats)
            .toStream()
            .to("order-analytics");
    }
}

// RabbitMQ routing example
@RabbitListener(queues = "order.processing.high-priority")
public void processHighPriorityOrder(Order order) {
    // Обработка приоритетных заказов
}

@RabbitListener(queues = "order.processing.normal")  
public void processNormalOrder(Order order) {
    // Обработка обычных заказов
}

✅ Обработка миллионов событий в секунду
✅ Analytics и ETL pipelines
✅ Activity tracking (clicks, page views)
✅ IoT data ingestion

✅ Нужно replay events с любого момента времени
✅ Audit trails и compliance требования
✅ Stream processing и real-time analytics
✅ CQRS implementations

✅ Data pipeline между системами
✅ CDC (Change Data Capture)
✅ Log aggregation от множества services
✅ Backup и disaster recovery

✅ Request-reply communication
✅ Task queues с приоритетами
✅ RPC calls между микросервисами
✅ Workflow orchestration

✅ Routing по multiple criteria
✅ Fan-out broadcasts
✅ Content-based routing
✅ Protocol translation (AMQP/STOMP/MQTT)

✅ Real-time notifications
✅ Chat applications
✅ Gaming backends
✅ Financial trading systems

Service A → [Event Store] ← Service B
                ↓
    [Event-driven processing]
                ↓
         [Projections/Views]

Service A → [Queue] → Service B
     ↑                   ↓
[Response Queue] ←── [ACK/NACK]

Sources → Kafka Connect → Kafka Streams → Sinks
  (DB)      (Ingestion)   (Processing)    (DWH)

Scheduler → [Job Queue] → Workers → [Result Queue] → Aggregator

External APIs → RabbitMQ → Service → Kafka → Analytics
    (Fast)      (Routing)  (Process)  (Store)   (Insights)

Infrastructure: Средняя (нужны диски, память)
Operational:    Высокая (требует expertise)
Licensing:      Бесплатная (OSS) / Expensive (Confluent)
Monitoring:     Комплексная (много метрик)

Infrastructure: Низкая (меньше ресурсов)
Operational:    Средняя (проще в управлении)  
Licensing:      Бесплатная (OSS) / Moderate (VMware)
Monitoring:     Простая (веб UI)

Phase 1: Dual write (RabbitMQ + Kafka)
Phase 2: Migrate consumers to Kafka  
Phase 3: Migrate producers to Kafka
Phase 4: Decommission RabbitMQ

Phase 1: Implement RabbitMQ alongside
Phase 2: Route new flows через RabbitMQ
Phase 3: Migrate existing consumers
Phase 4: Retire Kafka components

Нужен high throughput (>100k msg/sec)?
├─ YES → Kafka
└─ NO → Нужна сложная маршрутизация?
    ├─ YES → RabbitMQ  
    └─ NO → Нужно долговременное хранение?
        ├─ YES → Kafka
        └─ NO → Нужна низкая latency?
            ├─ YES → RabbitMQ
            └─ NO → Есть expertise для Kafka?
                ├─ YES → Kafka
                └─ NO → RabbitMQ

// Создание топологии
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");

KStream<String, String> stream = builder.stream("events-topic");
stream.filter((key, value) -> value.contains("error"))
      .to("error-topic");

KTable<String, Long> table = builder.table("users-topic");
// Каждая запись перезаписывает предыдущую с тем же ключом

GlobalKTable<String, String> globalTable = builder.globalTable("reference-data");

stream.filter((key, value) -> value.length() > 5);

stream.map((key, value) -> KeyValue.pair(key.toUpperCase(), value.toLowerCase()));

stream.flatMapValues(value -> Arrays.asList(value.split(" ")));

KStream<String, String>[] branches = stream.branch(
    (key, value) -> value.startsWith("A"),
    (key, value) -> value.startsWith("B"),
    (key, value) -> true  // остальные
);

KTable<String, Long> counts = stream
    .groupByKey()
    .aggregate(
        () -> 0L,  // initializer
        (key, value, aggregate) -> aggregate + 1,  // aggregator
        Materialized.as("counts-store")  // state store
    );

// Stream-Stream Join (windowed)
KStream<String, String> joined = leftStream.join(
    rightStream,
    (leftValue, rightValue) -> leftValue + ":" + rightValue,
    JoinWindows.of(Duration.ofMinutes(5))
);

// Stream-Table Join
KStream<String, String> enriched = stream.join(
    table,
    (streamValue, tableValue) -> streamValue + ":" + tableValue
);

stream.groupByKey()
      .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
      .count();

// Окно 10 минут, сдвиг каждые 2 минуты
stream.groupByKey()
      .windowedBy(TimeWindows.of(Duration.ofMinutes(10))
                              .advanceBy(Duration.ofMinutes(2)))
      .count();

stream.groupByKey()
      .windowedBy(SessionWindows.with(Duration.ofMinutes(30)))
      .count();

// Извлечение времени из payload
stream.transformValues(() -> new ValueTransformer<String, String>() {
    @Override
    public String transform(String value) {
        // Логика извлечения времени
        return value;
    }
});

StoreBuilder<KeyValueStore<String, Long>> storeBuilder = 
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("my-store"),
        Serdes.String(),
        Serdes.Long()
    );

StoreBuilder<WindowStore<String, Long>> windowStoreBuilder = 
    Stores.windowStoreBuilder(
        Stores.persistentWindowStore("window-store", 
                                   Duration.ofDays(1), 
                                   Duration.ofMinutes(5), 
                                   false),
        Serdes.String(),
        Serdes.Long()
    );

StoreBuilder<SessionStore<String, Long>> sessionStoreBuilder = 
    Stores.sessionStoreBuilder(
        Stores.persistentSessionStore("session-store", Duration.ofDays(1)),
        Serdes.String(),
        Serdes.Long()
    );

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

// Exactly-once processing
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
          StreamsConfig.EXACTLY_ONCE_V2);

// Commit interval
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);

// State directory
props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");

StreamsBuilder builder = new StreamsBuilder();
// Построение топологии...

KafkaStreams streams = new KafkaStreams(builder.build(), props);

// Graceful shutdown
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

streams.start();

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
          LogAndContinueExceptionHandler.class);

props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
          DefaultProductionExceptionHandler.class);

streams.setUncaughtExceptionHandler((thread, exception) -> {
    logger.error("Uncaught exception in thread " + thread.getName(), exception);
    return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
});

Serdes.String()    // для String
Serdes.Long()      // для Long
Serdes.Integer()   // для Integer
Serdes.Bytes()     // для byte[]

public class JsonSerde<T> implements Serde<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final Class<T> clazz;
    
    public JsonSerde(Class<T> clazz) {
        this.clazz = clazz;
    }
    
    @Override
    public Serializer<T> serializer() {
        return (topic, data) -> {
            try {
                return objectMapper.writeValueAsBytes(data);
            } catch (Exception e) {
                throw new SerializationException("Error serializing", e);
            }
        };
    }
    
    @Override
    public Deserializer<T> deserializer() {
        return (topic, data) -> {
            try {
                return objectMapper.readValue(data, clazz);
            } catch (Exception e) {
                throw new SerializationException("Error deserializing", e);
            }
        };
    }
}

streams.metrics().forEach((metricName, metric) -> {
    System.out.println(metricName.name() + ": " + metric.metricValue());
});

// Восстановление состояния из событий
KTable<String, Account> accounts = builder
    .stream("account-events")
    .groupByKey()
    .aggregate(
        Account::new,
        (key, event, account) -> account.apply(event),
        Materialized.as("accounts-store")
    );

// Разделение команд и запросов
KStream<String, Command> commands = builder.stream("commands");
KStream<String, Event> events = commands.mapValues(Command::execute);
events.to("events");

// Создание read-модели
KTable<String, ReadModel> readModel = builder
    .stream("events")
    .groupByKey()
    .aggregate(
        ReadModel::new,
        (key, event, model) -> model.update(event),
        Materialized.as("read-model-store")
    );

props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
          StreamsConfig.EXACTLY_ONCE_V2);

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// Увеличение batch size
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 1024 * 1024);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30000);

// Увеличение количества stream threads
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);

// Настройка RocksDB для state stores
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, 
          CustomRocksDBConfigSetter.class);

props.put("acks", "all"); // Ждать подтверждения от всех реплик
props.put("retries", Integer.MAX_VALUE); // Максимальное количество повторов
props.put("enable.idempotence", false); // Отключаем идемпотентность
props.put("max.in.flight.requests.per.connection", 5);

props.put("enable.auto.commit", false); // Ручной коммит офсетов
// Коммитим офсет только после успешной обработки
consumer.commitSync();

props.put("acks", "0"); // Не ждать подтверждения
props.put("retries", 0); // Не повторять отправку
props.put("batch.size", 16384); // Отправка батчами для производительности

props.put("enable.auto.commit", true); // Автоматический коммит
props.put("auto.commit.interval.ms", 1000); // Интервал коммита
// Коммит происходит до обработки сообщения

props.put("enable.idempotence", true); // Включаем идемпотентность
props.put("acks", "all"); // Обязательно для идемпотентности
props.put("retries", Integer.MAX_VALUE);
props.put("max.in.flight.requests.per.connection", 5); // ≤5 для идемпотентности

props.put("transactional.id", "my-transaction-id"); // Уникальный ID транзакции

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // Инициализация транзакций

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic", "key", "value"));
    producer.commitTransaction(); // Атомарный коммит
} catch (Exception e) {
    producer.abortTransaction(); // Откат транзакции
}

props.put("isolation.level", "read_committed"); // Читать только закоммиченные сообщения
props.put("enable.auto.commit", false);

// Обработка в рамках транзакции
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> record : records) {
            // Обработка сообщения
            processMessage(record);
            // Отправка результата в другой топик
            producer.send(new ProducerRecord<>("output-topic", record.value()));
        }
        
        // Коммит офсетов в рамках транзакции
        producer.sendOffsetsToTransaction(getOffsets(records), consumer.groupMetadata());
        producer.commitTransaction();
        
    } catch (Exception e) {
        producer.abortTransaction();
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("enable.idempotence", true); // Включаем идемпотентность
props.put("acks", "all"); // Обязательно для идемпотентности
props.put("retries", Integer.MAX_VALUE); // Автоматические повторы
props.put("max.in.flight.requests.per.connection", 5); // ≤5 для порядка сообщений

props.put("enable.idempotence", true);
props.put("transactional.id", "unique-app-id-123"); // Стабильный ID

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // Регистрация транзакционного ID

producer.beginTransaction();
try {
    producer.send(new ProducerRecord<>("topic", "key", "value"));
    producer.commitTransaction(); // Атомарное подтверждение
} catch (Exception e) {
    producer.abortTransaction(); // Откат при ошибке
}

// Генерация уникального ключа
String messageId = UUID.randomUUID().toString();
ProducerRecord<String, String> record = new ProducerRecord<>("topic", messageId, payload);

// Consumer: проверка дубликатов
Set<String> processedIds = new HashSet<>(); // В реальности — база данных

if (!processedIds.contains(record.key())) {
    processMessage(record.value());
    processedIds.add(record.key());
}

// Структура сообщения с версией
public class VersionedMessage {
    private String id;
    private long version;
    private String data;
    private long timestamp;
}

// Обработка с проверкой версии
if (incomingMessage.getVersion() > currentVersion) {
    updateRecord(incomingMessage);
} else {
    // Игнорируем устаревшее сообщение
    log.info("Ignoring outdated message version");
}

@Entity
public class ProcessedMessage {
    @Id
    private String messageId;
    private LocalDateTime processedAt;
    private String topic;
    private int partition;
    private long offset;
}

// Проверка при обработке
@Transactional
public void processMessage(ConsumerRecord<String, String> record) {
    String messageId = extractMessageId(record.value());
    
    if (processedMessageRepository.existsById(messageId)) {
        log.info("Message {} already processed, skipping", messageId);
        return;
    }
    
    // Атомарная обработка + сохранение факта обработки
    businessLogic.process(record.value());
    processedMessageRepository.save(new ProcessedMessage(messageId, LocalDateTime.now()));
}

props.put("enable.auto.commit", false); // Отключаем автокоммит

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        try {
            processMessageIdempotent(record); // Идемпотентная обработка
            consumer.commitSync(); // Коммит после успешной обработки
        } catch (Exception e) {
            log.error("Processing failed for offset {}", record.offset());
            // Не коммитим офсет, сообщение будет обработано повторно
        }
    }
}

// Consumer с изоляцией транзакций
consumerProps.put("isolation.level", "read_committed");

// Producer-Consumer в одной транзакции
producer.beginTransaction();
try {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        String result = processMessage(record.value());
        producer.send(new ProducerRecord<>("output-topic", result));
    }
    
    // Коммит офсетов в рамках транзакции
    Map<TopicPartition, OffsetAndMetadata> offsets = buildOffsets(records);
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// 1. Включаем идемпотентность Producer
props.put("enable.idempotence", true);

// 2. Используем уникальные ключи
String businessKey = order.getId() + "-" + order.getVersion();
ProducerRecord<String, Order> record = new ProducerRecord<>("orders", businessKey, order);

// 3. Идемпотентная обработка в Consumer
@Transactional
public void handleOrder(Order order) {
    if (orderRepository.existsByIdAndVersion(order.getId(), order.getVersion())) {
        return; // Уже обработан
    }
    
    processOrder(order);
    markAsProcessed(order.getId(), order.getVersion());
}

// Логирование дубликатов для анализа
if (isDuplicate(messageId)) {
    duplicateCounter.increment();
    log.warn("Duplicate message detected: {} in topic: {}", messageId, topic);
}

// Настройка основной очереди с DLX
@Bean
public Queue mainQueue() {
    return QueueBuilder.durable("main.queue")
        .withArgument("x-dead-letter-exchange", "dlx.exchange") // DLX exchange
        .withArgument("x-dead-letter-routing-key", "failed") // routing key для DLQ
        .withArgument("x-message-ttl", 60000) // TTL сообщений (60 сек)
        .build();
}

@Bean
public DirectExchange deadLetterExchange() {
    return new DirectExchange("dlx.exchange");
}

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("dead.letter.queue").build();
}

@Bean
public Binding deadLetterBinding() {
    return BindingBuilder.bind(deadLetterQueue())
        .to(deadLetterExchange())
        .with("failed");
}

@RabbitListener(queues = "main.queue")
public void processMessage(String message, Channel channel, 
                          @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
    try {
        // Бизнес-логика обработки
        businessLogic.process(message);
        channel.basicAck(deliveryTag, false); // Подтверждение успешной обработки
        
    } catch (BusinessException e) {
        // Отклоняем с requeue=false для отправки в DLX
        channel.basicNack(deliveryTag, false, false);
        log.error("Message rejected and sent to DLX: {}", message);
        
    } catch (RetryableException e) {
        // Отклоняем с requeue=true для повторной попытки
        channel.basicNack(deliveryTag, false, true);
    }
}

@RabbitListener(queues = "main.queue")
public void processWithRetry(String message, Channel channel,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                           @Header(value = "x-retry-count", required = false) Integer retryCount) {
    
    int currentRetry = (retryCount != null) ? retryCount : 0;
    
    try {
        businessLogic.process(message);
        channel.basicAck(deliveryTag, false);
        
    } catch (Exception e) {
        if (currentRetry >= MAX_RETRIES) {
            // Превышено количество попыток — в DLX
            channel.basicNack(deliveryTag, false, false);
            log.error("Max retries exceeded, sending to DLX: {}", message);
        } else {
            // Повторная отправка с увеличенным счетчиком
            republishWithDelay(message, currentRetry + 1);
            channel.basicAck(deliveryTag, false);
        }
    }
}

private void republishWithDelay(String message, int retryCount) {
    rabbitTemplate.convertAndSend("retry.exchange", "retry.key", message, msg -> {
        msg.getMessageProperties().setHeader("x-retry-count", retryCount);
        msg.getMessageProperties().setExpiration(String.valueOf(retryCount * 1000)); // Экспоненциальная задержка
        return msg;
    });
}

@Component
public class KafkaMessageProcessor {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private static final String DLQ_TOPIC = "orders.dlq";
    private static final int MAX_RETRIES = 3;
    
    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void processOrder(ConsumerRecord<String, String> record) {
        
        try {
            businessLogic.processOrder(record.value());
            
        } catch (Exception e) {
            handleFailedMessage(record, e);
        }
    }
    
    private void handleFailedMessage(ConsumerRecord<String, String> record, Exception error) {
        // Извлекаем счетчик попыток из headers
        Integer retryCount = extractRetryCount(record.headers());
        
        if (retryCount >= MAX_RETRIES) {
            // Отправляем в DLQ с метаданными об ошибке
            sendToDLQ(record, error, retryCount);
        } else {
            // Повторная отправка в тот же топик с увеличенным счетчиком
            retryMessage(record, retryCount + 1);
        }
    }
    
    private void sendToDLQ(ConsumerRecord<String, String> record, Exception error, int retryCount) {
        ProducerRecord<String, String> dlqRecord = new ProducerRecord<>(DLQ_TOPIC, record.key(), record.value());
        
        // Добавляем метаданные об ошибке
        dlqRecord.headers().add("original.topic", record.topic().getBytes());
        dlqRecord.headers().add("original.partition", String.valueOf(record.partition()).getBytes());
        dlqRecord.headers().add("original.offset", String.valueOf(record.offset()).getBytes());
        dlqRecord.headers().add("error.message", error.getMessage().getBytes());
        dlqRecord.headers().add("retry.count", String.valueOf(retryCount).getBytes());
        dlqRecord.headers().add("failed.timestamp", String.valueOf(System.currentTimeMillis()).getBytes());
        
        kafkaTemplate.send(dlqRecord);
        log.error("Message sent to DLQ after {} retries: {}", retryCount, record.value());
    }
}

{
  "name": "jdbc-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://localhost:5432/orders",
    
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "orders.dlq",
    "errors.deadletterqueue.topic.replication.factor": "3",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

@Component
public class OrderProcessingStream {
    
    @Bean
    public KStream<String, String> processOrders(StreamsBuilder builder) {
        
        KStream<String, String> ordersStream = builder.stream("orders");
        
        // Разделяем поток на успешные и неуспешные сообщения
        KStream<String, String>[] branches = ordersStream.branch(
            (key, value) -> {
                try {
                    validateOrder(value);
                    return true; // Валидные сообщения
                } catch (Exception e) {
                    return false; // Невалидные — в DLQ
                }
            }
        );
        
        // Обработка валидных сообщений
        branches[0]
            .mapValues(this::processValidOrder)
            .to("processed.orders");
            
        // Отправка невалидных в DLQ с метаданными
        branches[1]
            .map((key, value) -> {
                // Создаем запись с метаданными об ошибке
                DLQRecord dlqRecord = new DLQRecord(value, System.currentTimeMillis(), "Validation failed");
                return KeyValue.pair(key, dlqRecord.toJson());
            })
            .to("orders.dlq");
            
        return ordersStream;
    }
}

// RabbitMQ метрики
@Component
public class DLQMonitoring {
    
    @Autowired
    private RabbitAdmin rabbitAdmin;
    
    @Scheduled(fixedRate = 60000) // Каждую минуту
    public void monitorDLQ() {
        Properties queueInfo = rabbitAdmin.getQueueProperties("dead.letter.queue");
        Integer messageCount = (Integer) queueInfo.get("QUEUE_MESSAGE_COUNT");
        
        if (messageCount > THRESHOLD) {
            alertingService.sendAlert("DLQ overflow: " + messageCount + " messages");
        }
    }
}

// Kafka метрики через JMX или custom metrics
@Component
public class KafkaDLQMonitoring {
    
    @EventListener
    public void handleDLQMessage(DLQMessageEvent event) {
        Metrics.counter("dlq.messages.total", 
            "topic", event.getOriginalTopic(),
            "error.type", event.getErrorType()
        ).increment();
    }
}

docker run -it --rm \
  -p 8161:8161 \
  -p 61616:61616 \
  apache/activemq-artemis:latest

<configuration>
   <core>
      <name>artemis-broker</name>
      
      <!-- Настройки журнала -->
      <journal-type>ASYNCIO</journal-type>
      <journal-file-size>10M</journal-file-size>
      <journal-min-files>2</journal-min-files>
      
      <!-- Лимиты памяти -->
      <global-max-size>1GB</global-max-size>
      <max-disk-usage>90</max-disk-usage>
      
      <!-- Сетевые настройки -->
      <acceptors>
         <acceptor name="artemis">tcp://0.0.0.0:61616</acceptor>
      </acceptors>
      
      <!-- Адреса и очереди -->
      <addresses>
         <address name="orders">
            <anycast>
               <queue name="order-processing"/>
            </anycast>
         </address>
      </addresses>
   </core>
</configuration>

<addresses>
   <!-- Point-to-point -->
   <address name="orders.processing">
      <anycast>
         <queue name="high-priority"/>
         <queue name="low-priority"/>
      </anycast>
   </address>
   
   <!-- Pub-sub -->
   <address name="notifications">
      <multicast/>
   </address>
   
   <!-- Смешанный режим -->
   <address name="mixed.address">
      <anycast>
         <queue name="queue1"/>
      </anycast>
      <multicast/>
   </address>
</addresses>

@Configuration
public class ArtemisConfig {
    
    @Bean
    public ConnectionFactory connectionFactory() {
        return new ActiveMQConnectionFactory(
            "tcp://localhost:61616"
        );
    }
    
    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory factory) {
        JmsTemplate template = new JmsTemplate(factory);
        template.setDeliveryPersistent(true);
        template.setTimeToLive(300000); // 5 минут
        return template;
    }
}

@Service
public class OrderMessageProducer {
    
    @Autowired
    private JmsTemplate jmsTemplate;
    
    // Отправка в очередь (anycast)
    public void sendOrderToQueue(Order order) {
        jmsTemplate.convertAndSend("order-processing", order, message -> {
            message.setJMSPriority(9); // Высокий приоритет
            message.setStringProperty("orderType", order.getType());
            return message;
        });
    }
    
    // Отправка в топик (multicast) 
    public void publishOrderEvent(OrderEvent event) {
        jmsTemplate.setPubSubDomain(true);
        jmsTemplate.convertAndSend("order.events", event);
    }
}

@Component
public class OrderMessageConsumer {
    
    // Listener для очереди
    @JmsListener(destination = "order-processing")
    public void processOrder(Order order, Message message) {
        try {
            String orderType = message.getStringProperty("orderType");
            processOrderByType(order, orderType);
            
        } catch (Exception e) {
            // Сообщение автоматически пойдет в DLQ
            throw new RuntimeException("Processing failed", e);
        }
    }
    
    // Listener с селектором
    @JmsListener(destination = "order-processing", 
                 selector = "orderType = 'URGENT'")
    public void processUrgentOrder(Order order) {
        // Обработка только срочных заказов
    }
}

@Service
@Transactional
public class TransactionalOrderService {
    
    @JmsListener(destination = "orders")
    @Transactional(rollbackFor = Exception.class)
    public void processOrderTransactionally(Order order) {
        // 1. Сохранение в БД
        orderRepository.save(order);
        
        // 2. Отправка уведомления
        jmsTemplate.convertAndSend("notifications", 
            new OrderProcessedEvent(order.getId()));
        
        // При исключении - откат всей транзакции
        if (order.getAmount() < 0) {
            throw new IllegalArgumentException("Invalid amount");
        }
    }
}

@Configuration
public class XAConfig {
    
    @Bean
    public XAConnectionFactory xaConnectionFactory() {
        ActiveMQXAConnectionFactory factory = 
            new ActiveMQXAConnectionFactory("tcp://localhost:61616");
        return factory;
    }
    
    @Bean
    public JtaTransactionManager transactionManager() {
        return new JtaTransactionManager();
    }
}

<address-settings>
   <address-setting match="orders.#">
      <!-- Максимум попыток доставки -->
      <max-delivery-attempts>3</max-delivery-attempts>
      
      <!-- Задержка между попытками -->
      <redelivery-delay>5000</redelivery-delay>
      <redelivery-delay-multiplier>2</redelivery-delay-multiplier>
      <max-redelivery-delay>30000</max-redelivery-delay>
      
      <!-- DLQ настройки -->
      <dead-letter-address>DLQ</dead-letter-address>
      <auto-create-dead-letter-resources>true</auto-create-dead-letter-resources>
   </address-setting>
</address-settings>

<address-setting match="orders.#">
   <expiry-address>ExpiryQueue</expiry-address>
   <auto-create-expiry-resources>true</auto-create-expiry-resources>
</address-setting>

@Component
public class DLQHandler {
    
    @JmsListener(destination = "DLQ")
    public void handleFailedMessages(Message message) {
        try {
            String originalDest = message.getStringProperty("_AMQ_ORIG_ADDRESS");
            int deliveryCount = message.getIntProperty("_AMQ_DELIVERY_COUNT");
            
            logger.error("Message failed {} times from {}", 
                deliveryCount, originalDest);
                
            // Логирование, алертинг, ручная обработка
            alertService.sendAlert("DLQ message received", message);
            
        } catch (JMSException e) {
            logger.error("Error processing DLQ message", e);
        }
    }
}

<cluster-connections>
   <cluster-connection name="artemis-cluster">
      <connector-ref>artemis</connector-ref>
      <retry-interval>500</retry-interval>
      <use-duplicate-detection>true</use-duplicate-detection>
      <message-load-balancing>STRICT</message-load-balancing>
      <max-hops>1</max-hops>
      <discovery-group-ref>dg-group1</discovery-group-ref>
   </cluster-connection>
</cluster-connections>

<discovery-groups>
   <discovery-group name="dg-group1">
      <jgroups-file>jgroups-ping.xml</jgroups-file>
      <jgroups-channel>artemis_broadcast</jgroups-channel>
      <refresh-timeout>10000</refresh-timeout>
   </discovery-group>
</discovery-groups>

<!-- Master broker -->
<ha-policy>
   <replication>
      <master>
         <check-for-live-server>true</check-for-live-server>
      </master>
   </replication>
</ha-policy>

<!-- Slave broker -->
<ha-policy>
   <replication>
      <slave>
         <allow-failback>true</allow-failback>
      </slave>
   </replication>
</ha-policy>

@Component
public class ArtemisMonitor {
    
    @Autowired
    private MBeanServer mBeanServer;
    
    public QueueMetrics getQueueMetrics(String queueName) {
        try {
            ObjectName objectName = new ObjectName(
                "org.apache.activemq.artemis:broker=\"artemis\",component=addresses,address=\"" 
                + queueName + "\",subcomponent=queues,routing-type=\"anycast\",queue=\"" 
                + queueName + "\""
            );
            
            Long messageCount = (Long) mBeanServer.getAttribute(objectName, "MessageCount");
            Long consumerCount = (Long) mBeanServer.getAttribute(objectName, "ConsumerCount");
            
            return new QueueMetrics(messageCount, consumerCount);
            
        } catch (Exception e) {
            throw new RuntimeException("Failed to get metrics", e);
        }
    }
}

<!-- logging.properties -->
logger.org.apache.activemq.artemis.level=INFO
logger.org.apache.activemq.artemis.core.server.level=DEBUG

# Аудит сообщений
logger.org.apache.activemq.audit.level=INFO
logger.org.apache.activemq.audit.handlers=AUDIT_FILE

<configuration>
   <core>
      <!-- Batch настройки -->
      <journal-buffer-size>490KiB</journal-buffer-size>
      <journal-max-io>4096</journal-max-io>
      <journal-sync-transactional>false</journal-sync-transactional>
      
      <!-- Memory optimization -->
      <page-size-bytes>10M</page-size-bytes>
      <page-max-cache-size>5</page-max-cache-size>
      
      <!-- Network tuning -->
      <netty-threads>8</netty-threads>
      <scheduled-thread-pool-max-size>5</scheduled-thread-pool-max-size>
   </core>
</configuration>

<address-settings>
   <address-setting match="high-throughput.#">
      <!-- Увеличение размера страниц -->
      <page-size-bytes>10M</page-size-bytes>
      <max-size-bytes>100M</max-size-bytes>
      
      <!-- Отключение персистентности для временных данных -->
      <default-non-destructive>false</default-non-destructive>
      
      <!-- Batch обработка -->
      <redistribution-delay>0</redistribution-delay>
   </address-setting>
</address-settings>

<security-settings>
   <security-setting match="#">
      <permission type="createNonDurableQueue" roles="user"/>
      <permission type="deleteNonDurableQueue" roles="user"/>
      <permission type="createDurableQueue" roles="admin"/>
      <permission type="deleteDurableQueue" roles="admin"/>
      <permission type="createAddress" roles="admin"/>
      <permission type="deleteAddress" roles="admin"/>
      <permission type="consume" roles="user"/>
      <permission type="browse" roles="user"/>
      <permission type="send" roles="user"/>
      <permission type="manage" roles="admin"/>
   </security-setting>
</security-settings>

<security-manager>
   <ldap-security-setting>
      <initial-context-factory>com.sun.jndi.ldap.LdapCtxFactory</initial-context-factory>
      <connection-url>ldap://localhost:389</connection-url>
      <connection-username>uid=admin,ou=system</connection-username>
      <connection-password>secret</connection-password>
      <user-base>ou=people,dc=example,dc=com</user-base>
      <user-search-filter>uid={0}</user-search-filter>
      <role-base>ou=groups,dc=example,dc=com</role-base>
      <role-search-filter>member={0}</role-search-filter>
   </ldap-security-setting>
</security-manager>

# Создание брокера
artemis create mybroker --user admin --password admin

# Запуск/остановка
artemis run --home mybroker
artemis stop --home mybroker

# Управление очередями
artemis queue create --name orders --anycast --home mybroker
artemis queue stat --name orders --home mybroker

# Отправка тестового сообщения
artemis producer --message-count 100 --url tcp://localhost:61616 --destination orders

# Мониторинг
artemis browser --destination orders --home mybroker

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-artemis</artifactId>
</dependency>

spring:
  artemis:
    mode: native
    broker-url: tcp://localhost:61616
    user: admin
    password: admin
    pool:
      enabled: true
      max-connections: 10
    embedded:
      enabled: false
      
  jms:
    template:
      default-destination: orders
      delivery-mode: persistent
      time-to-live: 300000

@Component
public class ArtemisHealthIndicator implements HealthIndicator {
    
    @Autowired
    private ConnectionFactory connectionFactory;
    
    @Override
    public Health health() {
        try (Connection connection = connectionFactory.createConnection()) {
            connection.start();
            return Health.up()
                .withDetail("broker", "artemis")
                .withDetail("status", "connected")
                .build();
        } catch (Exception e) {
            return Health.down()
                .withDetail("error", e.getMessage())
                .build();
        }
    }
}