Apache Kafka (WIP)

version: '3.5'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 22181:2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

> docker-compose up -d
> docker ps
CONTAINER ID   IMAGE                              COMMAND                  CREATED          STATUS          PORTS                                         NAMES
ab09b5c8bc7b   confluentinc/cp-kafka:latest       "/etc/confluent/dock…"   11 minutes ago   Up 11 minutes   9092/tcp, 0.0.0.0:29092->29092/tcp            kafka-cluster_kafka_1
8e7725b7874b   confluentinc/cp-zookeeper:latest   "/etc/confluent/dock…"   11 minutes ago   Up 11 minutes   2888/tcp, 3888/tcp, 0.0.0.0:22181->2181/tcp   kafka-cluster_zookeeper_1

> $ netstat -aon | grep 22181
  TCP    0.0.0.0:22181          0.0.0.0:0              LISTENING       23792
  TCP    [::]:22181             [::]:0                 LISTENING       23792
  TCP    [::1]:22181            [::]:0                 LISTENING       27612

> netstat -aon | grep 29092
  TCP    0.0.0.0:29092          0.0.0.0:0              LISTENING       23792
  TCP    [::]:29092             [::]:0                 LISTENING       23792
  TCP    [::1]:29092            [::]:0                 LISTENING       27612

> docker-compose logs kafka | grep -i started
kafka_1      | [2022-05-31 14:49:34,438] DEBUG [ReplicaStateMachine controllerId=1] Started replica state machine with initial state -> HashMap() (kafka.controller.ZkReplicaStateMachine)
kafka_1      | [2022-05-31 14:49:34,441] DEBUG [PartitionStateMachine controllerId=1] Started partition state machine with initial state -> HashMap() (kafka.controller.ZkPartitionStateMachine)
kafka_1      | [2022-05-31 14:49:34,445] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
kafka_1      | [2022-05-31 14:49:34,450] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT_HOST) (kafka.network.SocketServer)
kafka_1      | [2022-05-31 14:49:34,450] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Started socket server acceptors and processors (kafka.network.SocketServer)
kafka_1      | [2022-05-31 14:49:34,459] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
kafka_1      | [2022-05-31 18:51:50,791] DEBUG [ReplicaStateMachine controllerId=1] Started replica state machine with initial state -> HashMap() (kafka.controller.ZkReplicaStateMachine)
kafka_1      | [2022-05-31 18:51:50,791] DEBUG [PartitionStateMachine controllerId=1] Started partition state machine with initial state -> HashMap() (kafka.controller.ZkPartitionStateMachine)
kafka_1      | [2022-05-31 19:52:52,922] DEBUG [ReplicaStateMachine controllerId=1] Started replica state machine with initial state -> HashMap() (kafka.controller.ZkReplicaStateMachine)
kafka_1      | [2022-05-31 19:52:52,923] DEBUG [PartitionStateMachine controllerId=1] Started partition state machine with initial state -> HashMap() (kafka.controller.ZkPartitionStateMachine)
kafka_1      | [2022-05-31 21:02:51,738] DEBUG [ReplicaStateMachine controllerId=1] Started replica state machine with initial state -> HashMap() (kafka.controller.ZkReplicaStateMachine)
kafka_1      | [2022-05-31 21:02:51,739] DEBUG [PartitionStateMachine controllerId=1] Started partition state machine with initial state -> HashMap() (kafka.controller.ZkPartitionStateMachine)
kafka_1      | [2022-05-31 23:24:34,226] DEBUG [ReplicaStateMachine controllerId=1] Started replica state machine with initial state -> HashMap() (kafka.controller.ZkReplicaStateMachine)
kafka_1      | [2022-05-31 23:24:34,226] DEBUG [PartitionStateMachine controllerId=1] Started partition state machine with initial state -> HashMap() (kafka.controller.ZkPartitionStateMachine)

> docker-compose ps
          Name                       Command            State                      Ports
-----------------------------------------------------------------------------------------------------------
kafka-cluster_kafka_1       /etc/confluent/docker/run   Up      0.0.0.0:29092->29092/tcp, 9092/tcp
kafka-cluster_zookeeper_1   /etc/confluent/docker/run   Up      0.0.0.0:22181->2181/tcp, 2888/tcp, 3888/tcp

> docker exec -t kafka-cluster_kafka_1 kafka-topics
Create, delete, describe, or change a topic.
Option                                   Description
------                                   -----------
--alter                                  Alter the number of partitions,
                                           replica assignment, and/or
                                           configuration for the topic.
.
.
.

> docker exec -it kafka-cluster_kafka_1 sh
sh-4.4$

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --create --topic first_topic --partitions 3 --replication-factor 1

WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic first_topic.

sh-4.4$

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --list

first_topic

sh-4.4$

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --describe --topic first_topic

Topic: first_topic      TopicId: vQ6o8fX1Sx-qNQuUQ5vbAg PartitionCount: 3       ReplicationFactor: 1    Configs:
        Topic: first_topic      Partition: 0    Leader: 1       Replicas: 1     Isr: 1
        Topic: first_topic      Partition: 1    Leader: 1       Replicas: 1     Isr: 1
        Topic: first_topic      Partition: 2    Leader: 1       Replicas: 1     Isr: 1

sh-4.4$

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --list

first-topic
first_topic

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --delete --topic first-topic

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --list

first_topic

sh-4.4$

sh-4.4$ kafka-console-producer --broker-list localhost:9092 --topic first_topic

>message1
>message2
>message3
>message4
>^C

sh-4.4$

sh-4.4$ kafka-console-producer --broker-list localhost:9092 --topic new_topic

>This is a message to a new topic
[2022-06-02 18:56:53,185] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 3 : {new_topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
>another message

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --list

first_topic
new_topic

sh-4.4$ kafka-topics --bootstrap-server localhost:9092 --topic new_topic --describe

Topic: new_topic        TopicId: FuC1JLRETNCrgUWaEPZCOg PartitionCount: 1       ReplicationFactor: 1    Configs:
        Topic: new_topic        Partition: 0    Leader: 1       Replicas: 1     Isr: 1

sh-4.4$

sh-4.4$ kafka-console-producer --broker-list localhost:9092 --topic first_topic --producer-property acks=all

>acks message1
>^C

sh-4.4$

sh-4.4$ kafka-console-consumer --bootstrap-server localhost:9092 --topic first_topic --from-beginning

message2
acks message1
message3
message1
message4

sh-4.4$ kafka-console-producer --broker-list localhost:9092 --topic first_topic

>patate
>carotte
>pomme
>orange
>

sh-4.4$ kafka-console-consumer --bootstrap-server localhost:9092 --topic first_topic --group my-group

carotte
orange

sh-4.4$ kafka-console-consumer --bootstrap-server localhost:9092 --topic first_topic --group my-group

patate
pomme

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --list

my-group

sh-4.4$

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group my-group

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
my-group        first_topic     0          3               3               0               console-consumer-13be95b9-8704-4471-8211-2660c5ab59b1 /172.19.0.3     console-consumer
my-group        first_topic     1          2               2               0               console-consumer-13be95b9-8704-4471-8211-2660c5ab59b1 /172.19.0.3     console-consumer
my-group        first_topic     2          4               4               0               console-consumer-13be95b9-8704-4471-8211-2660c5ab59b1 /172.19.0.3     console-consumer

sh-4.4$

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group my-group

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
my-group        first_topic     0          3               3               0               console-consumer-13be95b9-8704-4471-8211-2660c5ab59b1 /172.19.0.3     console-consumer
my-group        first_topic     1          2               2               0               console-consumer-13be95b9-8704-4471-8211-2660c5ab59b1 /172.19.0.3     console-consumer
my-group        first_topic     2          4               4               0               console-consumer-13be95b9-8704-4471-8211-2660c5ab59b1 /172.19.0.3     console-consumer

sh-4.4$

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --group my-group --reset-offsets --to-earliest --execute --topic first_topic

GROUP                          TOPIC                          PARTITION  NEW-OFFSET
my-group                       first_topic                    0          0
my-group                       first_topic                    1          0
my-group                       first_topic                    2          0

sh-4.4$

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group my-group

Consumer group 'my-group' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
my-group        first_topic     0          0               3               3               -               -               -
my-group        first_topic     1          0               2               2               -               -               -
my-group        first_topic     2          0               4               4               -               -               -

sh-4.4$

kafka-consumer-groups --bootstrap-server localhost:9092 --group my-group --reset-offsets --shift-by 1 --
execute --topic first_topic

GROUP                          TOPIC                          PARTITION  NEW-OFFSET
my-group                       first_topic                    0          1
my-group                       first_topic                    1          1
my-group                       first_topic                    2          1

sh-4.4$

--to-datetime
--by-period
--to-earliest
--to-latest
--shift-by
--from-file
--to-current

public class KafkaProducerDemo {

    public static void main(String[] args) {
        Properties properties = new Properties();

        // Set Producer Properties
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:29092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // Create Kafka Producer
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        ProducerRecord<String, String> record = new ProducerRecord<>("first_topic", "hello world from Java!");

        // Send data
        producer.send(record); // asynchronous, need to flush data
        producer.flush();

        // Tear down Producer
        producer.close();
    }
}

public class KafkaProducerDemoWithCallback {

    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(KafkaProducerDemoWithCallback.class);

        Properties properties = new Properties();

        // Set Producer Properties
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:29092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // Create Kafka Producer
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        ProducerRecord<String, String> record = new ProducerRecord<>("first_topic", "hello world from Java!");

        // Send data
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                // execute everytime a message is sent OR an exception is thrown.
                if (exception == null) {
                    // successfully sent message
                    logger.info("Received metadata - Topic: "
                            + metadata.topic() + " Partition: "
                            + metadata.partition() + " Offset: "
                            + metadata.offset() + " Timestamp: "
                            + metadata.timestamp());
                } else {
                    logger.error("Error while producer sent a message", exception);
                }
            }
        }); // asynchronous, need to flush data
        producer.flush();

        // Tear down Producer
        producer.close();
    }
}

public class KafkaConsumerDemo {
    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(KafkaConsumerDemo.class.getName());

        // Configure the consumer
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // Possible values : none, earliest, latest

        // Create consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // Subscribe the consumer to the Topic or Topics
        consumer.subscribe(Arrays.asList("first_topic"));

        // Poll for new data
        while(true) { // Avoid in production - demo purpose only.
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

            records.forEach(r -> logger.info("Key: " + r.key()
                        + " Value: " + r.value()
                        + " Partition: " + r.partition()
                        + " Offset: " + r.offset()
                        + " Timestamp: " + r.timestamp()));
        }
    }
}

public class KafkaConsumerWithThreadsDemo {
    Logger logger = LoggerFactory.getLogger(KafkaConsumerWithThreadsDemo.class.getName());

    public KafkaConsumerWithThreadsDemo() { }

    public void run() {
        String bootstrapServer = "localhost:29092";
        String groupId = "my_group";
        String topic = "first_topic";

        // Latch for dealing with multiple threads;
        CountDownLatch latch = new CountDownLatch(1);

        // Create Consumer Runnable;
        Runnable consumerRunnable = new ConsumerRunnable(bootstrapServer, groupId, topic, latch);

        // Starting Consumer Runnable Thread;
        Thread thread = new Thread(consumerRunnable);
        thread.start();

        // Add a shutdown hook;
        Runtime.getRuntime().addShutdownHook(new Thread( () -> {
            logger.info("Received a shutdown hook...");
            ((ConsumerRunnable) consumerRunnable).shutdown();

            try {
                latch.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            logger.info("Consumer application has exited...");
        }));

        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.error("Consumer application got interrupted", e);
        } finally {
            logger.info("Consumer application is closing...");
        }

    }

    public class ConsumerRunnable implements Runnable {
        private CountDownLatch latch;
        KafkaConsumer<String, String> consumer;

        public ConsumerRunnable(String bootstrapServer, String groupId, String topic, CountDownLatch latch) {
            this.latch = latch;

            // Configure the consumer
            Properties properties = new Properties();
            properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
            properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // Possible values : none, earliest, latest

            // Create consumer
            consumer = new KafkaConsumer<String, String>(properties);

            // Subscribe the consumer to the Topic or Topics
            consumer.subscribe(Arrays.asList(topic));
        }

        @Override
        public void run() {
            // Poll for new data
            try {
                while (true) { // Avoid in production - demo purpose only.
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                    records.forEach(r -> logger.info("Key: " + r.key()
                            + " Value: " + r.value()
                            + " Partition: " + r.partition()
                            + " Offset: " + r.offset()
                            + " Timestamp: " + r.timestamp()));
                }
            } catch ( WakeupException exception) {
                logger.info("Received shutdown signal...");
            } finally {
                consumer.close();
                latch.countDown(); // telling caller code that this thread is done.
            }
        }

        public void shutdown() {
            // to interrupt the consumer.poll() method

            // and will make consumer.poll() to throw an exception WakeupException
            consumer.wakeup();
        }
    }

    public static void main(String[] args) {
        new KafkaConsumerWithThreadsDemo().run();
    }
}

public class KafkaConsumerWithAssignAndSeekDemo {
    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(KafkaConsumerWithAssignAndSeekDemo.class.getName());

        // Configure the consumer
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "my_group"); No needs for group with assign and seek...
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // Possible values : none, earliest, latest

        // Create consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // The Assign and Seek is mostly used to replay data or fetch a specific message.

        // Assign
        TopicPartition partition = new TopicPartition("first_topic", 0);
        consumer.assign(Arrays.asList(partition));

        // Seek
        consumer.seek(partition, 1L);

        // Poll for new data
        while(true) { // Avoid in production - demo purpose only.
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

            records.forEach(r -> logger.info("Key: " + r.key()
                        + " Value: " + r.value()
                        + " Partition: " + r.partition()
                        + " Offset: " + r.offset()
                        + " Timestamp: " + r.timestamp()));
        }
    }
}

dependencies {
    implementation group: 'com.twitter', name: 'twitter-api-java-sdk', version: '1.2.4'
}

kafka-topics --bootstrap-server localhost:9092 --create --topic twitter_tweets --partitions 6 --replication-factor 1

package com.github.alainbouchard.kafka_demo.demo2;

public interface TwitterKafkaProducerInterface {
    void send(String topic, String message);
}

public class TwitterKafkaProducer implements TwitterKafkaProducerInterface {
    private final Logger logger = LoggerFactory.getLogger(TwitterKafkaProducer.class.getName());
    private KafkaProducer<String, String> producer;

    public TwitterKafkaProducer() {
        Properties properties = new Properties();

        // Set Producer Properties
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:29092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // Create Kafka Producer
        producer = new KafkaProducer<String, String>(properties);

        // Create topic:
        // kafka-topics --bootstrap-server localhost:9092 --create --topic twitter_tweets --partitions 6 --replication-factor 1
    }

    public Logger getLogger() {
        return logger;
    }

    @Override
    public void send(String topic, String message) {
        ProducerRecord producerRecord = new ProducerRecord<>(topic, null, message);
        producer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    getLogger().error("Could not send the message to the producer.", exception);
                }
            }
        });
    }
}

public class TwitterListener {
    /***
     * Ref:
     *    https://developer.axonivy.com/api-browser?url=/market-cache/twitter/twitter-connector-product/9.3.0/openapi.json#/Tweets/addOrDeleteRules
     *    https://github.com/twitterdev/twitter-api-java-sdk/tree/d0d6a8ce8db16faf4e3e1841c3a43bd5a56aa069
     *    https://developer.twitter.com/en/docs/twitter-api
     */

    private Logger logger = LoggerFactory.getLogger(TwitterListener.class.getName());
    // API V2 uses BEARER token.
    // TODO: Use environment variable for BEARER_TOKEN.
    private final String BEARER_TOKEN = "";
    protected TwitterApi twitterApi;

    private TwitterKafkaProducerInterface twitterKafkaProducer;

    Function<List<Rule>, List<String>> GetIdsFromRules = r -> r.stream().map(Rule::getId).collect(Collectors.toList());

    public TwitterListener() {
        twitterApi = new TwitterApi();
        TwitterCredentialsBearer credentials = new TwitterCredentialsBearer(BEARER_TOKEN);

        twitterApi.setTwitterCredentials(credentials);
    }

    public Logger getLogger() {
        return logger;
    }

    public TwitterApi getTwitterApi() {
        return twitterApi;
    }

    public void setTwitterKafkaProducer(TwitterKafkaProducerInterface twitterKafkaProducer) {
        this.twitterKafkaProducer = twitterKafkaProducer;
    };

    private void logApiExceptionToString(String description, ApiException e) {
        String text = description
                + " Status code: " + e.getCode()
                + " Reason: " + e.getResponseBody()
                + " Response headers: " + e.getResponseHeaders();
        getLogger().error(text, e);
    }

    private List<Rule> addRule(String value, String tag, boolean dryRun) {
        // Create rule
        RuleNoId ruleNoId = new RuleNoId();
        ruleNoId.setValue(value);
        ruleNoId.setTag(tag);

        // Add rule to list of rules
        List<RuleNoId> ruleNoIds = new ArrayList<>();
        ruleNoIds.add(ruleNoId);

        // Add the list of rules to the request
        AddRulesRequest addRulesRequest = new AddRulesRequest();
        addRulesRequest.add(ruleNoIds);

        AddOrDeleteRulesRequest addOrDeleteRulesRequest = new AddOrDeleteRulesRequest();
        addOrDeleteRulesRequest.setActualInstance(addRulesRequest);

        List<Rule> rules = null;

        try {
            AddOrDeleteRulesResponse result = getTwitterApi().tweets().addOrDeleteRules(addOrDeleteRulesRequest, dryRun);
            getLogger().info(result.toString());

            rules = result.getData();
        } catch (ApiException e) {
            logApiExceptionToString("Exception when calling TweetsApi#addOrDeleteRules", e);
        }

        return rules;
    }

    private AddOrDeleteRulesResponse deleteRules(List<Rule> rules, boolean dryRun) {
        DeleteRulesRequestDelete deleteRulesRequestDelete = new DeleteRulesRequestDelete();
        deleteRulesRequestDelete.ids(GetIdsFromRules.apply(rules));

        DeleteRulesRequest deleteRulesRequest = new DeleteRulesRequest();
        deleteRulesRequest.setDelete(deleteRulesRequestDelete);

        AddOrDeleteRulesRequest addOrDeleteRulesRequestForDelete = new AddOrDeleteRulesRequest();
        addOrDeleteRulesRequestForDelete.setActualInstance(deleteRulesRequest);

        AddOrDeleteRulesResponse result = null;

        try {
            result = this.getTwitterApi().tweets().addOrDeleteRules(addOrDeleteRulesRequestForDelete, dryRun);
            getLogger().info(result.toString());
        } catch (ApiException e) {
            logApiExceptionToString("Exception when calling TweetsApi#addOrDeleteRules", e);
        }

        return result;
    }

    private GetRulesResponse getRules(String paginationToken, List<Rule> rules, Integer maxResults) {
        List<String> ruleIds = rules != null? GetIdsFromRules.apply(rules) : null;
        GetRulesResponse result = null;

        try {
            result = getTwitterApi().tweets().getRules(ruleIds, maxResults, paginationToken);
            getLogger().info(result.toString());
        } catch (ApiException e) {
            logApiExceptionToString("Exception when calling TweetsApi#getRules", e);
        }

        return result;
    }

    private GetRulesResponse getRules(String paginationToken, List<Rule> rules) {
        return this.getRules(paginationToken, rules, 1000);
    }

    private GetRulesResponse getRules(String paginationToken) {
        return this.getRules(paginationToken, null);
    }

    private void searchStream() {
        Set<String> expansions = new HashSet<>(Arrays.asList());
        Set<String> tweetFields = new HashSet<>();
        tweetFields.add("id");
        tweetFields.add("author_id");
        tweetFields.add("created_at");
        tweetFields.add("text");
        Set<String> userFields = new HashSet<>(Arrays.asList());
        Set<String> mediaFields = new HashSet<>(Arrays.asList());
        Set<String> placeFields = new HashSet<>(Arrays.asList());
        Set<String> pollFields = new HashSet<>(Arrays.asList());
        Integer backfillMinutes = null;  // There is a bug in the Twitter API v2 where any specified value will cause an error.

        InputStream result = null;

        try {
            result = getTwitterApi().tweets().searchStream(expansions, tweetFields, userFields, mediaFields, placeFields, pollFields, backfillMinutes);

            try {
                JSON json = new JSON();
                Type localVarReturnType = new TypeToken<FilteredStreamingTweet>(){}.getType();
                BufferedReader reader = new BufferedReader(new InputStreamReader(result));
                String line = reader.readLine();

                while (line != null) {
                    if (line.isEmpty()) {
                        getLogger().info("==> Empty line");
                        line = reader.readLine();
                        continue;
                    }
                    Object jsonObject = json.getGson().fromJson(line, localVarReturnType);
                    String message = jsonObject != null ? jsonObject.toString() : "Null object";
                    getLogger().info(message);
                    twitterKafkaProducer.send("twitter_tweets", message);

                    line = reader.readLine();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        } catch (ApiException e) {
            logApiExceptionToString("Exception when calling TweetsApi#searchStream", e);
        }
    }

    public static void main(String[] args) {
        TwitterListener twitterListener = new TwitterListener();
        twitterListener.setTwitterKafkaProducer(new TwitterKafkaProducer());

        boolean dryRun = false;

        // Delete all existing Rules;
        try {
            GetRulesResponse rulesResponse = twitterListener.getRules(null);
            AddOrDeleteRulesResponse result = twitterListener.deleteRules(rulesResponse.getData(), dryRun);
            twitterListener.getLogger().info("Deleted rules: " + twitterListener.GetIdsFromRules.apply(result.getData()));
        } catch (Exception e) {
            twitterListener.getLogger().error("oops!");  // bug in the TwitterApi SDK.
            e.printStackTrace();
        }

        // Adding Rules;
        twitterListener.addRule( "potus -is:retweet", "Non-retweeted potus tweets", dryRun);
        twitterListener.addRule( "hockey -is:retweet", "Non-retweeted hockey tweets", dryRun);
        twitterListener.addRule( "baseball -is:retweet", "Non-retweeted baseball tweets", dryRun);
        twitterListener.addRule( "bitcoin -is:retweet", "Non-retweeted bitcoin tweets", dryRun);

        // Filter twitter stream;
        twitterListener.searchStream();
    }
}

Producer             Broker 101 partition 0 (leader)
--------             -------------------------------
    |                               |
    +-----[send data to leader]---->+
    |                               |

Producer             Broker 101 partition 0 (leader)
--------             -------------------------------
    |                               |
    +----[send data to leader]----->+
    |                               |
    +<---[respond write reqs]-------+
    |                               |

Producer             Broker 101 part0 (leader)    Broker 102 part0 (replica)    Broker 103 part0 (replica)
--------             -------------------------    --------------------------    --------------------------
    |                            |                            |                            |
    +---[send data to leader]--->+                            |                            |
    |                            |                            |                            |
    |                            +-----[send to replica]----->+                            |
    |                            |                            |                            |
    |                            +-----[send to replicas]--------------------------------->+
    |                            |                            |                            |
    |                            +<--------[ack write]--------+                            |
    |                            |                            |                            |
    |                            +<--------[ack write]-------------------------------------+
    |                            |                            |                            |
    +<---[respond write reqs]----+                            |                            |
    |                            |

        // Make a safer producer
        properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");  // default value with Idempotence=true
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE)); // default value with Idempotence=true
        properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); // with kafka >= 2.0, otherwise use 1.

+----------------+       +--------------------+      +---------------+
| Producer Batch |------>| Compression of the |----->| Kafka Cluster |
+----------------+       | batch of messages  |      +---------------+
                         +--------------------+

        // Improve throughput efficiency of the producer
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "20");
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "32768");  // 32 KB

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.5.2
    ports:
      - 9200:9200
      - 9300:9300
    environment:
      discovery.type: single-node
      node.name: es01
      cluster.name: kafka-cluster

> docker-compose up elasticsearch -d
> docker-compose ps

NAME                            COMMAND                  SERVICE             STATUS              PORTS
kafka-cluster-elasticsearch-1   "/usr/local/bin/dock…"   elasticsearch       running             0.0.0.0:9200->9200/tcp, 0.0.0.0:9300->9300/tcp
kafka-cluster-kafka-1           "/etc/confluent/dock…"   kafka               running             0.0.0.0:29092->29092/tcp
kafka-cluster-zookeeper-1       "/etc/confluent/dock…"   zookeeper           running             0.0.0.0:22181->2181/tcp

> curl localhost:9200/

{
  "name" : "es01",
  "cluster_name" : "kafka-cluster",
  "cluster_uuid" : "-w05UbWdSeOhc4nRGcy8yA",
  "version" : {
    "number" : "7.5.2",
    "build_flavor" : "default",
    "build_type" : "docker",
    "build_hash" : "8bec50e1e0ad29dad5653712cf3bb580cd1afcdf",
    "build_date" : "2020-01-15T12:11:52.313576Z",
    "build_snapshot" : false,
    "lucene_version" : "8.3.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

    implementation group: 'org.elasticsearch.client', name: 'elasticsearch-rest-high-level-client', version: '7.14.0'
    implementation group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.17.2' // Elastic search dependency

public class ElasticsearchClient {
    Logger logger = LoggerFactory.getLogger(ElasticsearchClient.class.getName());

    private RestHighLevelClient restHighLevelClient;

    public ElasticsearchClient() {
        restHighLevelClient = new RestHighLevelClient(
                // TODO: use configuration file or environment variables to set ip and ports, now configured for local docker containers.
                RestClient.builder(new HttpHost("localhost", 9200, "http"),
                        new HttpHost("localhost", 9300, "http")));
    }

    public RestHighLevelClient getRestHighLevelClient() {
        return restHighLevelClient;
    }

    public boolean ping() {
        boolean result = false;
        try {
            result = getRestHighLevelClient().ping(RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("Elasticsearch client received an exception.", e);
        } finally {
            logger.info("Elasticsearch aliveness: " + result);
        }

        return result;
    }

    public IndexResponse toIndex(String index, String jsonSource, String id) {
        IndexRequest indexRequest = new IndexRequest(index)
                .id(id) // Make the entry idempotent.
                .source(jsonSource, XContentType.JSON);

        IndexResponse indexResponse = null;
        try {
            indexResponse = getRestHighLevelClient().index(indexRequest, RequestOptions.DEFAULT);
            logger.info(indexResponse.getId());
        } catch (IOException e) {
            logger.error("Caught and exception.", e);
        }

        return indexResponse;
    }

    public void close() {
        try {
            getRestHighLevelClient().close();
        } catch (IOException e) {
            logger.error("Caught and exception.", e);
        }
    }

    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(ElasticsearchClient.class.getName());

        // Setup Elasticsearch
        ElasticsearchClient elasticsearchClient = new ElasticsearchClient();

        elasticsearchClient.ping();

        // Setup Kafka Consumer
        TwitterKafkaConsumer twitterKafkaConsumer = new TwitterKafkaConsumer();
        twitterKafkaConsumer.subscribe("twitter_tweets");

        // Get data using Kafka Consumer and insert data to the elasticsearch
        while(true) { // TODO: replace with better logic.
            ConsumerRecords<String, String> records = twitterKafkaConsumer.poll(Duration.ofMillis(100));

            records.forEach(record -> {
                FilteredStreamingTweetResponse tweet = twitterKafkaConsumer.mapTweetToObject(record.value());
                elasticsearchClient.toIndex("twitter", record.value(), tweet.getData().getId());

                logger.debug("Key: " + record.key()
                        + " Value: " + record.value()
                        + " Partition: " + record.partition()
                        + " Offset: " + record.offset()
                        + " Timestamp: " + record.timestamp());

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    logger.error("Error while waiting for sleep", e);
                }
            });
        }

        // Tear-down the Elasticsearch
//        elasticsearchClient.close();

        // Tear-down the Kafka Consumer
    }
}

properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false"); // Will require manual offsets commits
properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10"); // Only retrieve 10 records at the time

while(true) {
    ConsumerRecords<String, String> records = twitterKafkaConsumer.poll(Duration.ofMillis(100));

    logger.info(String.format("Received %s records", records.count()));
    records.forEach(record -> {
        // do something synchronous here...
    });
    logger.info("Committing offsets");
    twitterKafkaConsumer.commitSync();
    logger.info("Offsets were committed");
}

kafka-consumer-groups --bootstrap-server localhost:9092 --group kafka-java-demo-elasticsearch --describe

Consumer group 'kafka-java-demo-elasticsearch' has no active members.

GROUP                         TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
kafka-java-demo-elasticsearch twitter_tweets  3          0               39              39              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  0          0               37              37              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  4          0               63              63              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  5          0               43              43              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  2          0               37              37              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  1          0               58              58              -
 -               -
sh-4.4$

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --group kafka-java-demo-elasticsearch --describe

Consumer group 'kafka-java-demo-elasticsearch' has no active members.

GROUP                         TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
kafka-java-demo-elasticsearch twitter_tweets  3          0               39              39              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  0          0               37              37              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  4          0               63              63              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  5          30              43              13              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  2          0               37              37              -
 -               -
kafka-java-demo-elasticsearch twitter_tweets  1          0               58              58              -
 -               -
sh-4.4$

sh-4.4$ kafka-consumer-groups --bootstrap-server localhost:9092 --topic twitter_tweets --group kafka-java-demo-elasticse
arch --reset-offsets --to-earliest --execute

GROUP                          TOPIC                          PARTITION  NEW-OFFSET
kafka-java-demo-elasticsearch  twitter_tweets                 3          0
kafka-java-demo-elasticsearch  twitter_tweets                 0          0
kafka-java-demo-elasticsearch  twitter_tweets                 4          0
kafka-java-demo-elasticsearch  twitter_tweets                 5          0
kafka-java-demo-elasticsearch  twitter_tweets                 2          0
kafka-java-demo-elasticsearch  twitter_tweets                 1          0

sh-4.4$

Sources/Destinations     Connect Cluster             Kafka Cluster          Stream Apps
--------------------     ---------------             -------------          -----------
     [source1]------------->[worker]                   [broker]-------------->[app1]
                            [worker]------------------>[broker]<--------------
     [source2]              [worker]                   [broker]
                            [worker]<------------------[broker]-------------->[app2]
      [sinks]<--------------[worker]                           <--------------

+ [my-java-project]
+-+ [kafka-connect] (to be created with sub folders)
| +-+ [connectors]
| | +-+ [kafka-connect-twitter] (downloaded connector project from github)
| | | +-+ *.jar
| | +-+ connector-standalone.properties
| | | + run.sh
| | | + twitter.properties

sh-4.4$ connect-standalone
USAGE: /usr/bin/connect-standalone [-daemon] connect-standalone.properties

sh-4.4$

$ kafka-topics --bootstrap-server localhost:9092 --create --topic twitter_status_connect --parti
tions 3 --replication-factor 1

WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.

Created topic twitter_status_connect.

$ kafka-topics --bootstrap-server localhost:9092 --create --topic twitter_deletes_connect --part
itions 3 --replication-factor 1

WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.

Created topic twitter_deletes_connect.
$