<commit_before>/*
# Licensed Materials - Property of IBM
# Copyright IBM Corp. 2015  
 */
package kafka;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Future;

import com.ibm.streamsx.topology.TStream;
import com.ibm.streamsx.topology.Topology;
import com.ibm.streamsx.topology.context.ContextProperties;
import com.ibm.streamsx.topology.context.StreamsContextFactory;
import com.ibm.streamsx.topology.function.Function;
import com.ibm.streamsx.topology.function.UnaryOperator;
import com.ibm.streamsx.topology.messaging.kafka.ConsumerConnector;
import com.ibm.streamsx.topology.messaging.kafka.ProducerConnector;
import com.ibm.streamsx.topology.tuple.Message;
import com.ibm.streamsx.topology.tuple.SimpleMessage;


/**
 * Demonstrate integrating with the Apache Kafka messaging system
 * <a href="http://kafka.apache.org">http://kafka.apache.org</a>.
 * <p>
 * Connectors are used to create a bridge between topology streams
 * and a Kafka cluster:
 * <ul>
 * <li>{@link com.ibm.streamsx.topology.messaging.kafka.ConsumerConnector ConsumerConnector} - subscribe to Kafka topics and create streams of messages.</li>
 * <li>{@link com.ibm.streamsx.topology.messaging.kafka.ProducerConnector ProducerConnector} - publish streams of messages to Kafka topics.</li>
 * </ul>
 * <p>
 * The sample publishes some messages to a Kafka topic.  
 * It also subscribes to the topic and reports the messages received.
 * The messages received may include messages from prior runs of the sample.
 * <p>
 * The sample requires a running Kafka cluster with the following
 * characteristics:
 * <ul>
 * <li>the kafka topic (@code kafkaSampleTopic} has been created.
 * e.g.<br>
 * {@code ${KAFKA_HOME}/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafkaSampleTopic}
 * </li>
 * <li>the Kafka cluster's zookeeper connection is {@code localhost:2181}</li>
 * <li>the Kafka cluster's brokers addresses is {@code localhost:9092}</li>
 * </ul>
 * <p>
 * Required IBM Streams environment variables:
 * <ul>
 * <li>STREAMS_INSTALL - the Streams installation directory</li>
 * <li>STREAMS_DOMAIN_ID - the Streams domain to use for context {@code DISTRIBUTED}
 * <li>STREAMS_INSTANCE_ID - the Streams instance to use for context {@code DISTRIBUTED}
 * </ul>
 * <p>
 * See the Apache Kafka link above for information about setting up a Kafka
 * cluster and creating a topic.
 * <p>
 * This may be executed as (from the {@code samples/java/functional directory} )
 * as:
 * <UL>
 * <LI>{@code ant run.kafka.distributed} - Using Apache Ant, this will run in distributed mode.</li>
 * <LI>{@code ant run.kafka} - Using Apache Ant, this will run in standalone mode.</li>
 * <LI>
 * {@code java -cp functionalsamples.jar:../com.ibm.streamsx.topology/lib/com.ibm.streamsx.topology.jar:$STREAMS_INSTALL/lib/com.ibm.streams.operator.samples.jar kafka.KafkaSample} <i>[CONTEXT_TYPE]</i>
 * - Run directly from the command line with a specific
 * {@code StreamsContext} where <i>CONTEXT_TYPE</i> is one of:
 * </LI>
 * <UL>
 * <LI>{@code DISTRIBUTED} - Run as an IBM Streams distributed
 * application.</LI>
 * <LI>{@code STANDALONE} - Run as an IBM Streams standalone
 * application.</LI>
 * <LI>{@code BUNDLE} - Create an IBM Streams application bundle.</LI>
 * <LI>{@code TOOLKIT} - Create an IBM Streams application toolkit.</LI>
 * </UL>
 * <LI>
 * An application execution within your IDE once you set the class path to include the correct jars.</LI>
 * </UL>

 */
public class KafkaSample {
    private static final String ZOOKEEPER_CONNECT = "localhost:2181";    
    private static final String KAFKA_METADATA_BROKER_LIST = "localhost:9092";
    
    private static final String TOPIC = "kafkaSampleTopic";

    private static final int PUB_DELAY_MSEC = 5*1000;
    private static final String uniq = new SimpleDateFormat("HH:mm:ss.SSS").format(new Date());
    private boolean captureArtifacts = false;
    private boolean setAppTracingLevel = false;
    private java.util.logging.Level appTracingLevel = java.util.logging.Level.FINE;<commit_msg>Fix javadoc for "java -cp ..."<commit_after>/*
# Licensed Materials - Property of IBM
# Copyright IBM Corp. 2015  
 */
package kafka;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Future;

import com.ibm.streamsx.topology.TStream;
import com.ibm.streamsx.topology.Topology;
import com.ibm.streamsx.topology.context.ContextProperties;
import com.ibm.streamsx.topology.context.StreamsContextFactory;
import com.ibm.streamsx.topology.function.Function;
import com.ibm.streamsx.topology.function.UnaryOperator;
import com.ibm.streamsx.topology.messaging.kafka.ConsumerConnector;
import com.ibm.streamsx.topology.messaging.kafka.ProducerConnector;
import com.ibm.streamsx.topology.tuple.Message;
import com.ibm.streamsx.topology.tuple.SimpleMessage;


/**
 * Demonstrate integrating with the Apache Kafka messaging system
 * <a href="http://kafka.apache.org">http://kafka.apache.org</a>.
 * <p>
 * Connectors are used to create a bridge between topology streams
 * and a Kafka cluster:
 * <ul>
 * <li>{@link com.ibm.streamsx.topology.messaging.kafka.ConsumerConnector ConsumerConnector} - subscribe to Kafka topics and create streams of messages.</li>
 * <li>{@link com.ibm.streamsx.topology.messaging.kafka.ProducerConnector ProducerConnector} - publish streams of messages to Kafka topics.</li>
 * </ul>
 * <p>
 * The sample publishes some messages to a Kafka topic.  
 * It also subscribes to the topic and reports the messages received.
 * The messages received may include messages from prior runs of the sample.
 * <p>
 * The sample requires a running Kafka cluster with the following
 * characteristics:
 * <ul>
 * <li>the kafka topic (@code kafkaSampleTopic} has been created.
 * e.g.<br>
 * {@code ${KAFKA_HOME}/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafkaSampleTopic}
 * </li>
 * <li>the Kafka cluster's zookeeper connection is {@code localhost:2181}</li>
 * <li>the Kafka cluster's brokers addresses is {@code localhost:9092}</li>
 * </ul>
 * <p>
 * Required IBM Streams environment variables:
 * <ul>
 * <li>STREAMS_INSTALL - the Streams installation directory</li>
 * <li>STREAMS_DOMAIN_ID - the Streams domain to use for context {@code DISTRIBUTED}
 * <li>STREAMS_INSTANCE_ID - the Streams instance to use for context {@code DISTRIBUTED}
 * </ul>
 * <p>
 * See the Apache Kafka link above for information about setting up a Kafka
 * cluster and creating a topic.
 * <p>
 * This may be executed as (from the {@code samples/java/functional directory} )
 * as:
 * <UL>
 * <LI>{@code ant run.kafka.distributed} - Using Apache Ant, this will run in distributed mode.</li>
 * <LI>{@code ant run.kafka} - Using Apache Ant, this will run in standalone mode.</li>
 * <LI>
 * {@code java -cp functionalsamples.jar:../../../com.ibm.streamsx.topology/lib/com.ibm.streamsx.topology.jar:$STREAMS_INSTALL/lib/com.ibm.streams.operator.samples.jar kafka.KafkaSample} <i>[CONTEXT_TYPE]</i>
 * - Run directly from the command line with a specific
 * {@code StreamsContext} where <i>CONTEXT_TYPE</i> is one of:
 * </LI>
 * <UL>
 * <LI>{@code DISTRIBUTED} - Run as an IBM Streams distributed
 * application.</LI>
 * <LI>{@code STANDALONE} - Run as an IBM Streams standalone
 * application.</LI>
 * <LI>{@code BUNDLE} - Create an IBM Streams application bundle.</LI>
 * <LI>{@code TOOLKIT} - Create an IBM Streams application toolkit.</LI>
 * </UL>
 * <LI>
 * An application execution within your IDE once you set the class path to include the correct jars.</LI>
 * </UL>

 */
public class KafkaSample {
    private static final String ZOOKEEPER_CONNECT = "localhost:2181";    
    private static final String KAFKA_METADATA_BROKER_LIST = "localhost:9092";
    
    private static final String TOPIC = "kafkaSampleTopic";

    private static final int PUB_DELAY_MSEC = 5*1000;
    private static final String uniq = new SimpleDateFormat("HH:mm:ss.SSS").format(new Date());
    private boolean captureArtifacts = false;
    private boolean setAppTracingLevel = false;
    private java.util.logging.Level appTracingLevel = java.util.logging.Level.FINE;
{'fraction_non_alphanumeric': 0.11886366337810532, 'fraction_numerical': 0.007369547129442529, 'mean_word_length': 4.372924648786718, 'pattern_counts': {'":': 0, '<': 109, '<?xml version=': 0, '>': 109, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7243435', 'n_tokens_mistral': 2721, 'n_tokens_neox': 2646, 'n_words': 788}