TwitterStreamListener.java

    @Override
    public void onStatus(Status status) {

        List<String> matchedTerms = new ArrayList<>(getMatchedTerms(status.getText()));

        if (!status.isPossiblySensitive() && !matchedTerms.isEmpty()) {

            String tweetID = String.valueOf(status.getId());
            String tweetText = status.getText();
            System.out.println("Matched terms  - " + matchedTerms);

            System.out.println("sending tweet to kafka...");

            producer.send(new ProducerRecord<>(TWEETS_TOPIC, tweetID, tweetText),
                    new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println("Successfully sent tweet with ID " + tweetID + " and text '" + tweetText + "' to topic " + TWEETS_TOPIC);
                }
            });
        }

    }