Mehmet Cem Yücel mehmetcemyucel

	@Component
	public class BasicStream {

	private static final Serde<String> STRING_SERDE = Serdes.String();
	private static final String INPUT_TOPIC = "basic-stream-input-topic";
	private static final String OUTPUT_TOPIC = "basic-stream-output-topic";

	@Autowired
	void buildPipeline(StreamsBuilder streamsBuilder) {
	KStream<String, String> messageStream = streamsBuilder.stream(INPUT_TOPIC, Consumed.with(STRING_SERDE, STRING_SERDE));

	@Configuration
	@EnableKafka
	@EnableKafkaStreams
	public class KafkaConfig {

	@Value(value = "${spring.kafka.bootstrap-servers}")
	private String bootstrapAddress;


	@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)