anchor.diff:                          (stationId, timestampAndValue) -> stationId,
                           (timestampAndValue, station) -> {
                               String[] parts = timestampAndValue.split(";");
  -                            return new TemperatureMeasurement(station.id, station.name, Instant.parse(parts[0]), Double.valueOf(parts[1]));
  +                            return new TemperatureMeasurement(station.id, station.name,
  +                                    Instant.parse(parts[0]), Double.valueOf(parts[1]));
                           }
                   )
                   .groupByKey()                                                 // <5>
  
positive.code:     @Outgoing("weather-stations")                               // <3>
      public Flowable<KafkaMessage<Integer, String>> weatherStations() {
          List<KafkaMessage<Integer, String>> stationsAsJson = stations.stream()
              .map(s -> KafkaMessage.of(s.id, "{ \"id\" : " + s.id + ", \"name\" : \"" + s.name + "\" }"))
              .collect(Collectors.toList());
  
          return Flowable.fromIterable(stationsAsJson);
  
negative.code: 
      private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsPipeline.class);
  
      @ConfigProperty(
          name="org.acme.quarkus.sample.kafkastreams.bootstrap.servers",
          defaultValue="localhost:9092"
      )
      String bootstrapServers;
  
      @ConfigProperty(name="HOSTNAME")                            // <1>
  