dependencies {
  implementation: "spring-kafka"
}

spring:
  kafka:
    bootstrap-servers: localhost:9092

@Configuration
public class KafkaTopicConfig {

  @Bean
  public NewTopic amgifoscodeTopic() { // NewTopic : org.apache.kafka.clients.admin.NewTopic
    return TopicBuilder.name("amigoscode").build();
  }
}

@Configuration
public class KafkaConsumerConfig {

    @Value("localhost:9092")
    private String bootstrapServers;

    public Map<String, Object> consumerConfig() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return properties;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfig());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>> factory(ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

    // ConcurrentKafkaListenerContainerFactory<String, Object> ~~~
}

@SpringBootApplication
public class KafkaexampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaexampleApplication.class, args);
    }

    @Bean
    CommandLineRunner commandLineRunner(KafkaTemplate<String, String> kafkaTemplate) {
        return args -> {
            kafkaTemplate.send("amigoscode", "hello kafka"); // topic, data(message)
        };
    }
}

@Componenet
public class KafkaListeners {
  @KafkaListener(topics = "amigoscode", groupId = "groupId")
  void listener(String data) {
    System.out.println("Listener received: " + data );
  }
}

// record
public record MessageRequest(String message) {
}

=

// class
// Records provide a public constructor (with all arguments), read methods for each field (equivalent to getters) and the implementation of hashCode, equals and toString methods.
public class MessageRequest {
    private String message;

    public MessageRequest() {
    }

    public MessageRequest(String message) {
    }

    public String getMessage() {
        return message;
    }

    public boolean equals(Object o) {
        return true;
    }

    public int hashCode() {
        return 0;
    }

    public String toString() {
        return "";
    }
}

@RestController
@RequestMapping("api/vi/messages")
public class MessageController {
  private KafkaTemplate<String, String> kafkaTemplate;

  public MessageController(KafkaTemplate<String, String> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }

  @PostMapping
  public void publish(@RequestBody MessageReqeust request) {
    kafkaTemplate.send("amigoscode", request.getMessage());
  }
}

POST http://localhost:8080/api/vi/messages
Content-Type: application/json

{
  "message": "Api With Kafka"
}

dependencies {
  implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka'
}

public interface Sink {
    String INPUT = "process-input"; // INPUT : consumer 입장에서 subscribe 받을 TOPIC명

    @Input(INPUT)
    SubscribableChannel input();
}

public interface Source {
    String OUTPUT = "process-output"; // OUTPUT : producer 입장에서 publish할 TOPIC명

    @Output(OUTPUT)
    MessageChannel output();
}

public interface Processor extends Source, Sink {
}

public interface ProcessMessage {
    String SEND_MESSAGE = "send-message";
    String RECEIVE_MESSAGE = "receive-message";

    @Output(SEND_MESSAGE)
    MessageChannel sendMessage();

    @Input(RECEIVE_MESSAGE)
    SubscribableChannel getMessage();
}

@ConditionalOnProperty(name = "spring.cloud.stream.enabled", havingValue = "true", matchIfMissing = true)
@EnableBinding(EventSource.class)
public class MessageConsumer {

  @StreamListener(target = ProcessMessage.RECEIVE_MESSAGE, condition = "headers['eventType'] == 'MessageEvent'")
    public void pushMessage(MessageEvent event) throws JsonProcessingException {
      ...
    }
}

public class MessageEvent {
  // props
}

spring:
  cloud:
    stream:
      enable:
      kafka:
        binder:
          brokers: localhost // broker ip
          # replication-factor: 2 // minimum = 2
          # auto-create-topics: false
        bindings: // input/output
          receive-message: // channel name
            destination: "${spring.cloud.stream.topic:}process-input" // topic name
            group: "${spring.cloud.stream.consumer.group:}" // consumer group id
          send-message: // output
            destination: "${spring.cloud.stream.topic:}process-output"

@EnableBinding(EventSource.class)

@KafkaListener(topics = "amigoscode", groupId = "groupId")

@ConditionalOnProperty(name = "spring.cloud.stream.enabled", havingValue = "true", matchIfMissing = true)