jacace

    private static void joinDemo(JavaDStream<ConsumerRecord<String, String>> productStream,
                                 JavaDStream<ConsumerRecord<String, String>> salesStream) {
        ObjectMapper jacksonParser = new ObjectMapper();
        JavaPairDStream<Object, Object> s1 = productStream.mapToPair(record -> new Tuple2<Object, Object>(record.key(),
                jacksonParser.readValue(record.value(), Item.class)));
        JavaPairDStream<Object, Object> s2 = salesStream.mapToPair(record -> new Tuple2<Object, Object>(record.key(),
                jacksonParser.readValue(record.value(), DailySales.class)));
        JavaPairDStream<Object, Tuple2<Object, Object>> s3 = s1.join(s2);

        s3.foreachRDD(new VoidFunction<JavaPairRDD<Object, Tuple2<Object, Object>>>() {