title: Apache kafka - print from java application to console
body: <p>I am running ubuntu 18 and using java (intellij IDE) and writing a basic kafka application. I am trying to use the basic example from <a href="https://docs.confluent.io/current/streams/quickstart.html" rel="nofollow noreferrer">here</a>, and to stream information to the application, and then print something to the screen, I am running the application using intellij "run" command.</p>
  
  <p>When I connent the application to an output stream, it works fine and I manage to output information to the terminal. </p>
  
  <p>I tried adding <code>System.out.println()</code> in the foreach methods, in the apply methods, it's not working, I am adding breakpoints inside them and running debug mode, and it's not getting there, I guess the flow does not get there during the run.</p>
  
  <p>I am streaming information to the application using the proper topic, and prints that I put in the application outside the apply and foreach are working fine.</p>
  
  <p>How can I get the application to print something each time I stream information to it? the main idea is to process something and print the results to the monitor and not to a kafka stream</p>
  
  <p>Here is the code:</p>
  
  <pre><code>import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.common.serialization.Serdes;
  
  import org.apache.kafka.streams.KafkaStreams;
  import org.apache.kafka.streams.StreamsBuilder;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.kstream.*;
  import org.apache.kafka.streams.kstream.Printed;
  
  import java.util.Arrays;
  import java.util.Locale;
  import java.util.Properties;
  import java.util.concurrent.CountDownLatch;
  
  public class main {
      public static void main(String[] args) throws Exception {
          Properties props = new Properties();
          props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
          props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
          props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
          props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
  
          // setting offset reset to earliest so that we can re-run the demo code with the same pre-loaded data
          // Note: To re-run the demo, you need to use the offset reset tool:
          // https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  
          StreamsBuilder builder = new StreamsBuilder();
  
          KStream&lt;String, String&gt; source = builder.stream("streams-plaintext-input");
  
          source.foreach(new ForeachAction&lt;String, String&gt;() {
              @Override
              public void apply(String key, String value) {
                  System.out.println("yeah");
              }
          });
  
          KTable&lt;String, Long&gt; counts = source
                  .flatMapValues(new ValueMapper&lt;String, Iterable&lt;String&gt;&gt;() {
                      @Override
                      public Iterable&lt;String&gt; apply(String value) {
                          return Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
                      }
                  })
                  .groupBy(new KeyValueMapper&lt;String, String, String&gt;() {
                      @Override
                      public String apply(String key, String value) {
                          System.out.println("what");
                          return value;
                      }
                  })
                  .count();
          //System.exit(0);
          // need to override value serde to Long type
          System.out.println("what");
          //counts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));
  
  
  
          final KafkaStreams streams = new KafkaStreams(builder.build(), props);
          final CountDownLatch latch = new CountDownLatch(1);
  
          // attach shutdown handler to catch control-c
          Runtime.getRuntime().addShutdownHook(new Thread("streams-wordcount-shutdown-hook") {
              @Override
              public void run() {
                  streams.close();
                  latch.countDown();
              }
          });
  
          try {
              streams.start();
              latch.await();
          } catch (Throwable e) {
              System.exit(2);
          }
          System.exit(0);
      }
  
  
  }
  </code></pre>