# 운영 프로파일에 KafkaProfileValidator 가 거는 두 요구
  KafkaProfileValidator.java:51     if (profile.production() && !profile.tlsEnabled()) {
  KafkaProfileValidator.java:52       throw new IllegalArgumentException(
  KafkaProfileValidator.java:53           "a production Kafka connection requires TLS: " + profile.broker());
  KafkaProfileValidator.java:54     }
  KafkaProfileValidator.java:55     if (profile.production() && !profile.authenticationEnabled()) {
  KafkaProfileValidator.java:56       throw new IllegalArgumentException(
  KafkaProfileValidator.java:57           "a production Kafka connection requires broker authentication: " + profile.broker());
  KafkaProfileValidator.java:58     }

# 그 검증기가 기동에서 도는 경로와, 그 거부를 지키는 테스트
  main · KafkaMessagingAutoConfiguration.java:55       kafkaProfileStartupValidation(
  main · StartupProfileValidation.java:19  * <p>Validation runs as {@code afterPropertiesSet} rather than on an application event, so the
  main · StartupProfileValidation.java:31 final class StartupProfileValidation<T> implements InitializingBean {
  main · StartupProfileValidation.java:43   public void afterPropertiesSet() {
  test · MessagingConfigurationBindingTest.java:235   void aProductionKafkaBrokerWithoutTransportSecurityFailsStartup() {

# 두 조립부를 켜는 프로퍼티
  main · MessagingBridgeRootAutoConfiguration.java:21 @ConditionalOnProperty(prefix = "app.messaging", name = "enabled", havingValue = "true")
  main · MessagingAdminAutoConfiguration.java:27 @ConditionalOnProperty(prefix = "app.messaging.admin", name = "enabled", havingValue = "true")
  main · MessagingPlatformRootAutoConfiguration.java:28 @ConditionalOnProperty(prefix = MessagingSettings.PREFIX, name = "enabled", havingValue = "true")
  main · MessagingProviderSelection.java:48           "kafka", KafkaMessagingAutoConfiguration.class,
  main · KafkaSenderConfig.java:38 @ConditionalOnProperty(name = "app.messaging.broker", havingValue = "kafka")
  출하 기본값 : application.yml:949:    enabled: ${APP_MESSAGING_ENABLED:false}

# 프로덕션에서 KafkaProducer 를 만드는 두 자리 — 설정 맵 원문 그대로
  KafkaMessagingAutoConfiguration.java:128  @Bean(destroyMethod = "close")
  KafkaMessagingAutoConfiguration.java:129  @ConditionalOnMissingBean
  KafkaMessagingAutoConfiguration.java:139     java.util.Map<String, Object> config = new java.util.HashMap<>();
  KafkaMessagingAutoConfiguration.java:140     config.put(
  KafkaMessagingAutoConfiguration.java:141         org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  KafkaMessagingAutoConfiguration.java:142         bootstrapServers);
  KafkaMessagingAutoConfiguration.java:143     config.put(
  KafkaMessagingAutoConfiguration.java:144         org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
  KafkaMessagingAutoConfiguration.java:145         org.apache.kafka.common.serialization.ByteArraySerializer.class);
  KafkaMessagingAutoConfiguration.java:146     config.put(
  KafkaMessagingAutoConfiguration.java:147         org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
  KafkaMessagingAutoConfiguration.java:148         org.apache.kafka.common.serialization.ByteArraySerializer.class);
  KafkaMessagingAutoConfiguration.java:149     // Durability is not a default worth inheriting: acks=1 loses an accepted publish to a leader
  KafkaMessagingAutoConfiguration.java:150     // failover, which is exactly the outcome an outbox exists to prevent.
  KafkaMessagingAutoConfiguration.java:151     config.put(org.apache.kafka.clients.producer.ProducerConfig.ACKS_CONFIG, "all");
  KafkaMessagingAutoConfiguration.java:152     config.put(org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
  KafkaMessagingAutoConfiguration.java:153     return new org.apache.kafka.clients.producer.KafkaProducer<>(config);

  KafkaSenderConfig.java:61  @Bean(name = "kafkaSeamProducer", destroyMethod = "close")
  KafkaSenderConfig.java:62  @ConditionalOnMissingBean(name = "kafkaSeamProducer")
  KafkaSenderConfig.java:64     Map<String, Object> config = new HashMap<>();
  KafkaSenderConfig.java:65     config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, String.join(",", settings.brokers()));
  KafkaSenderConfig.java:66     config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  KafkaSenderConfig.java:67     config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  KafkaSenderConfig.java:68     config.put(ProducerConfig.ACKS_CONFIG, "all");
  KafkaSenderConfig.java:69     config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
  KafkaSenderConfig.java:77     config.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10_000);
  KafkaSenderConfig.java:78     config.put(ProducerConfig.LINGER_MS_CONFIG, 0);
  KafkaSenderConfig.java:79     config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 15_000);
  KafkaSenderConfig.java:80     config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 30_000);
  KafkaSenderConfig.java:81     return new KafkaProducer<>(config);

# security.protocol 을 넣는 프로덕션 코드 — 상수명과 리터럴 양쪽으로
  main · KafkaSecurityConfigurer.java:32   public static final String SECURITY_PROTOCOL = "security.protocol";
  main · KafkaSecurityConfigurer.java:80     properties.put(SECURITY_PROTOCOL, securityProtocol(profile, credential));

# KafkaSecurityConfigurer.configure 가 자격 종류마다 하는 일
  KafkaSecurityConfigurer.java:90       case BrokerCredentialProfile.SaslScram scram -> {
  KafkaSecurityConfigurer.java:92         properties.put(SASL_MECHANISM, "SCRAM-SHA-512");
  KafkaSecurityConfigurer.java:93         properties.put(SASL_JAAS_CONFIG, scramJaas(scram.credentialId(), resolved));
  KafkaSecurityConfigurer.java:95       case BrokerCredentialProfile.OAuth2 oauth -> {
  KafkaSecurityConfigurer.java:99         throw new dev.caskeleton.messaging.api.error.MessagingConfigurationException(
  KafkaSecurityConfigurer.java:106       case BrokerCredentialProfile.MutualTls ignored -> properties.put(SASL_MECHANISM, "NONE");
  KafkaSecurityConfigurer.java:107       case BrokerCredentialProfile.UsernamePassword plain -> {
  KafkaSecurityConfigurer.java:109         properties.put(SASL_MECHANISM, "PLAIN");
  KafkaSecurityConfigurer.java:110         properties.put(SASL_JAAS_CONFIG, plainJaas(plain.credentialId(), resolved));
  KafkaSecurityConfigurer.java:112       case BrokerCredentialProfile.Nkey ignored ->
  KafkaSecurityConfigurer.java:113           throw new IllegalArgumentException(

# 그 클래스가 빈이 되기까지의 조건 사슬
  KafkaMessagingAutoConfiguration.java:105   @Bean
  KafkaMessagingAutoConfiguration.java:106   @ConditionalOnMissingBean
  KafkaMessagingAutoConfiguration.java:107   @org.springframework.boot.autoconfigure.condition.ConditionalOnBean(
  KafkaMessagingAutoConfiguration.java:108       CredentialRuntimeRegistry.class)
  KafkaMessagingAutoConfiguration.java:109   public KafkaSecurityConfigurer kafkaSecurityConfigurer(
  KafkaMessagingAutoConfiguration.java:110       CredentialRuntimeRegistry credentials, BrokerTlsPolicy tlsPolicy) {
  KafkaMessagingAutoConfiguration.java:111     return new KafkaSecurityConfigurer(credentials, tlsPolicy);
  KafkaMessagingAutoConfiguration.java:112   }
  main · MessagingCoreAutoConfiguration.java:315   @ConditionalOnBean(CredentialProvider.class)
  main · MessagingCoreAutoConfiguration.java:317   public CredentialRuntimeRegistry credentialRuntimeRegistry(CredentialProvider provider) {
  CredentialProvider 를 구현하는 main 클래스 : 0 건
  그 인터페이스를 구현하는 자리 전부 :
  test · CredentialRuntimeRegistryTest.java:21   private static final class CountingProvider implements CredentialProvider {

# 그 빈을 파라미터나 필드로 받는 코드 (빈을 만드는 팩토리 자신은 뺀다)
  main 에서 0 건
  getBean · ObjectProvider · 빈 이름 문자열로 가져가는 자리 :
    0 건

# 조립을 확인하는 테스트가 단언하는 것과, 실 브로커 시험이 붙는 컨테이너
  test · MessagingStarterOffContractTest.java:118   void selectingKafkaAssemblesOnlyKafka() {
  test · MessagingStarterOffContractTest.java:172   void aSelectedTransportAssemblesAPublisher() {
  test · MessagingStarterOffContractTest.java:127               assertThat(context).hasSingleBean(KafkaMessagingAutoConfiguration.class);
  test · MessagingStarterOffContractTest.java:130                   .doesNotHaveBean(RabbitMessagingAutoConfiguration.class);
  test · MessagingStarterOffContractTest.java:189               assertThat(context).hasSingleBean(MessagePublisher.class);
  test · MessagingLiveRoundTripQualificationTest.java:66   @Container static final KafkaContainer KAFKA = new KafkaContainer("apache/kafka:4.1.0");
