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>>>() {

jacace

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> streamLeft = builder.stream("TopicIn1");
        KStream<String, String> streamRight = builder.stream("TopicIn2");

        KStream<String, String> joined = streamLeft.join(streamRight,
            (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
            JoinWindows.of(5000),
            Joined.with(
                Serdes.String(),  /* key */
                Serdes.String(),  /* left value */

jacace

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}

jacace

#With Kafka Streaming API:
stream.mapValues(value -> value.toLowerCase());

#With Spark Streaming API:
stream.map(record -> new Tuple2<>(record.key(), record.value()));

jacace

sh kafka-topics.sh --delete --topic demo_topic --bootstrap-server <server-id>.us-east1.gcp.confluent.cloud:9092 --command-config jaas.conf

jacace

kubectl describe secrets/build-robot-secret

jacace

//Sample kafka client in nodejs

'use strict';
require('dotenv').config()

//kafka init starts

const { Kafka } = require('kafkajs')
const kafka = new Kafka({
	clientId: process.env.consumerid, 

jacace

sh kafka-topics.sh --list --bootstrap-server <server-id>.us-east1.gcp.confluent.cloud:9092 --command-config jaas.conf

jacace

sh kafka-console-producer.sh --broker-list <server:id>.us-east1.gcp.confluent.cloud:9092 --producer.config config.properties --topic in_topic

jacace

sh kafka-topics.sh --create --topic out_topic --bootstrap-server <server_id>.us-east1.gcp.confluent.cloud:9092 --command-config jaas.conf