jacace

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
        username="" \
        password="";

jacace

sh kafka-topics.sh --create --topic out_topic --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 --list --bootstrap-server <server-id>.us-east1.gcp.confluent.cloud:9092 --command-config jaas.conf

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

kubectl describe secrets/build-robot-secret

jacace

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

jacace

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

#With Spark Streaming API:
stream.map(record -> new Tuple2<>(record.key(), record.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

        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 */