anchor.diff:                              .doubleValue();
   
                       LOG.infov("station: {0}, temperature: {1}", station.name, temperature);
  -                    return KafkaRecord.of(station.id, Instant.now() + ";" + temperature);
  +                    return Record.of(station.id, Instant.now() + ";" + temperature);
                   });
       }
   
  -    @Outgoing("weather-stations")                               // <3>
  -    public Flowable<KafkaRecord<Integer, String>> weatherStations() {
  -        List<KafkaRecord<Integer, String>> stationsAsJson = stations.stream()
  -            .map(s -> KafkaRecord.of(
  +    @Outgoing("weather-stations")                                          // <3>
  +    public Multi<Record<Integer, String>> weatherStations() {
  +        return Multi.createFrom().items(stations.stream()
  +            .map(s -> Record.of(
                       s.id,
                       "{ \"id\" : " + s.id +
                       ", \"name\" : \"" + s.name + "\" }"))
  -            .collect(Collectors.toList());
  -
  -        return Flowable.fromIterable(stationsAsJson);
  -    };
  +        );
  +    }
   
       private static class WeatherStation {
   
  
positive.code:     }
  }
  ----
  <1> Instruct Reactive Messaging to dispatch the items from the returned `Flowable` to `temperature-values`.
  <2> The method returns a RX Java 2 _stream_ (`Flowable`) emitting a random temperature value every 0.5 seconds.
  <3> Instruct Reactive Messaging to dispatch the items from the returned `Flowable` (static list of weather stations) to `weather-stations`.
  
  The two methods each return a _reactive stream_ whose items are sent to the streams named `temperature-values` and `weather-stations`, respectively.
  
  
negative.code: 
  import java.math.BigDecimal;
  import java.math.RoundingMode;
  import java.time.Duration;
  import java.time.Instant;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.Random;
  
  import javax.enterprise.context.ApplicationScoped;
  
  import io.smallrye.mutiny.Multi;
  import io.smallrye.reactive.messaging.kafka.Record;
  import org.eclipse.microprofile.reactive.messaging.Outgoing;
  import org.jboss.logging.Logger;
  
  /**
   * A bean producing random temperature data every second.
  