<commit_before>/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.extension.siddhi.io.kafka.multidc.source;

import org.apache.log4j.Logger;
import org.wso2.extension.siddhi.io.kafka.sink.KafkaSink;
import org.wso2.extension.siddhi.io.kafka.source.KafkaSource;
import org.wso2.siddhi.annotation.Example;
import org.wso2.siddhi.annotation.Extension;
import org.wso2.siddhi.annotation.Parameter;
import org.wso2.siddhi.annotation.util.DataType;
import org.wso2.siddhi.core.config.SiddhiAppContext;
import org.wso2.siddhi.core.exception.ConnectionUnavailableException;
import org.wso2.siddhi.core.stream.input.source.Source;
import org.wso2.siddhi.core.stream.input.source.SourceEventListener;
import org.wso2.siddhi.core.util.config.ConfigReader;
import org.wso2.siddhi.core.util.transport.OptionHolder;
import org.wso2.siddhi.query.api.definition.StreamDefinition;
import org.wso2.siddhi.query.api.exception.SiddhiAppValidationException;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * This class implements a Kafka multi data center source to receives events from a kafka cluster.
 */
@Extension(
        name = "kafkaMultiDC",
        namespace = "source",
        description = "The Kafka Multi-Datacenter(DC) source receives records from the same topic in brokers " +
                "deployed in two different kafka clusters. It filters out all the duplicate messages and ensures" +
                "that the events are received in the correct order using sequential numbering. It receives events in" +
                " formats such as `TEXT`, `XML` JSON` and `Binary`.The Kafka Source creates the default partition" +
                " '0' for a given topic, if the topic has not yet been created in the Kafka cluster.",
        parameters = {
                @Parameter(name = "bootstrap.servers",
                        description = "This contains the kafka server list which the kafka source "
                                + "listens to. This is given using comma-separated values. "
                                + "eg: 'localhost:9092,localhost:9093' ",
                        type = {DataType.STRING}),
                @Parameter(name = "topic",
                        description = "This is the topic that the source listens to. eg: 'topic_one' ",
                        type = {DataType.STRING}),
                @Parameter(name = "partition.no",
                        description = "This is the partition number of the given topic.",
                        type = {DataType.INT},
                        optional = true,
                        defaultValue = "0"),
                @Parameter(name = "is.binary.message",
                        description = "In order to receive the binary events via the Kafka Multi-DC source, the " +
                                "value of this parameter needs to be set to 'True'.",
                        type = {DataType.BOOL},
                        optional = true,
                        defaultValue = "false"),
                @Parameter(name = "optional.configuration",
                        description = "This contains all the other possible configurations with which the consumer "
                                + "can be created."
                                + "eg: producer.type:async,batch.size:200",
                        type = {DataType.STRING},
                        optional = true,
                        defaultValue = "null")
        },
        examples = {
                @Example(
                        syntax = "@App:name('TestExecutionPlan') \n" +
                                "define stream BarStream (symbol string, price float, volume long); \n" +
                                "@info(name = 'query1') \n" +
                                "@source("
                                + "type='kafkaMultiDC', "
                                + "topic='kafka_topic', "
                                + "bootstrap.servers='host1:9092,host1:9093', "
                                + "partition.no='1', "
                                + "@map(type='xml'))\n" +
                                "Define stream FooStream (symbol string, price float, volume long);\n" +
                                "from FooStream select symbol, price, volume insert into BarStream;\n",
                        description = "The following query listens to 'kafka_topic' topic, deployed in the broker " +
                                "host1:9092 and host1:9093, with partition 1. A thread is created for " +
                                "each broker. The receiving xml events are mapped to a siddhi event " +
                                "and sent to the FooStream.")
        }
)
public class KafkaMultiDCSource extends Source {
    private static final String KAFKA_TOPIC = "topic";
    private static final String KAFKA_PARTITION_NO = "partition.no";
    private static final Logger LOG = Logger.getLogger(KafkaMultiDCSource.class);
    private static final String LAST_RECEIVED_SEQ_NO_KEY = "lastConsumedSeqNo";
    private SourceEventListener eventListener;
    private Map<String, KafkaSource> sources = new HashMap<>();
    private String[] bootstrapServers;
    private SourceSynchronizer synchronizer;

    @Override
    public void init(SourceEventListener sourceEventListener, OptionHolder optionHolder, String[] transportPropertyNames,
                     ConfigReader configReader, SiddhiAppContext siddhiAppContext) {
        this.eventListener = sourceEventListener;
        String serverList = optionHolder.validateAndGetStaticValue(KafkaSource
            .ADAPTOR_SUBSCRIBER_ZOOKEEPER_CONNECT_SERVERS);<commit_msg>Fix line length<commit_after>/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.wso2.extension.siddhi.io.kafka.multidc.source;

import org.apache.log4j.Logger;
import org.wso2.extension.siddhi.io.kafka.sink.KafkaSink;
import org.wso2.extension.siddhi.io.kafka.source.KafkaSource;
import org.wso2.siddhi.annotation.Example;
import org.wso2.siddhi.annotation.Extension;
import org.wso2.siddhi.annotation.Parameter;
import org.wso2.siddhi.annotation.util.DataType;
import org.wso2.siddhi.core.config.SiddhiAppContext;
import org.wso2.siddhi.core.exception.ConnectionUnavailableException;
import org.wso2.siddhi.core.stream.input.source.Source;
import org.wso2.siddhi.core.stream.input.source.SourceEventListener;
import org.wso2.siddhi.core.util.config.ConfigReader;
import org.wso2.siddhi.core.util.transport.OptionHolder;
import org.wso2.siddhi.query.api.definition.StreamDefinition;
import org.wso2.siddhi.query.api.exception.SiddhiAppValidationException;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * This class implements a Kafka multi data center source to receives events from a kafka cluster.
 */
@Extension(
        name = "kafkaMultiDC",
        namespace = "source",
        description = "The Kafka Multi-Datacenter(DC) source receives records from the same topic in brokers " +
                "deployed in two different kafka clusters. It filters out all the duplicate messages and ensures" +
                "that the events are received in the correct order using sequential numbering. It receives events in" +
                " formats such as `TEXT`, `XML` JSON` and `Binary`.The Kafka Source creates the default partition" +
                " '0' for a given topic, if the topic has not yet been created in the Kafka cluster.",
        parameters = {
                @Parameter(name = "bootstrap.servers",
                        description = "This contains the kafka server list which the kafka source "
                                + "listens to. This is given using comma-separated values. "
                                + "eg: 'localhost:9092,localhost:9093' ",
                        type = {DataType.STRING}),
                @Parameter(name = "topic",
                        description = "This is the topic that the source listens to. eg: 'topic_one' ",
                        type = {DataType.STRING}),
                @Parameter(name = "partition.no",
                        description = "This is the partition number of the given topic.",
                        type = {DataType.INT},
                        optional = true,
                        defaultValue = "0"),
                @Parameter(name = "is.binary.message",
                        description = "In order to receive the binary events via the Kafka Multi-DC source, the " +
                                "value of this parameter needs to be set to 'True'.",
                        type = {DataType.BOOL},
                        optional = true,
                        defaultValue = "false"),
                @Parameter(name = "optional.configuration",
                        description = "This contains all the other possible configurations with which the consumer "
                                + "can be created."
                                + "eg: producer.type:async,batch.size:200",
                        type = {DataType.STRING},
                        optional = true,
                        defaultValue = "null")
        },
        examples = {
                @Example(
                        syntax = "@App:name('TestExecutionPlan') \n" +
                                "define stream BarStream (symbol string, price float, volume long); \n" +
                                "@info(name = 'query1') \n" +
                                "@source("
                                + "type='kafkaMultiDC', "
                                + "topic='kafka_topic', "
                                + "bootstrap.servers='host1:9092,host1:9093', "
                                + "partition.no='1', "
                                + "@map(type='xml'))\n" +
                                "Define stream FooStream (symbol string, price float, volume long);\n" +
                                "from FooStream select symbol, price, volume insert into BarStream;\n",
                        description = "The following query listens to 'kafka_topic' topic, deployed in the broker " +
                                "host1:9092 and host1:9093, with partition 1. A thread is created for " +
                                "each broker. The receiving xml events are mapped to a siddhi event " +
                                "and sent to the FooStream.")
        }
)
public class KafkaMultiDCSource extends Source {
    private static final String KAFKA_TOPIC = "topic";
    private static final String KAFKA_PARTITION_NO = "partition.no";
    private static final Logger LOG = Logger.getLogger(KafkaMultiDCSource.class);
    private static final String LAST_RECEIVED_SEQ_NO_KEY = "lastConsumedSeqNo";
    private SourceEventListener eventListener;
    private Map<String, KafkaSource> sources = new HashMap<>();
    private String[] bootstrapServers;
    private SourceSynchronizer synchronizer;

    @Override
    public void init(SourceEventListener sourceEventListener, OptionHolder optionHolder, String[]
            transportPropertyNames, ConfigReader configReader, SiddhiAppContext siddhiAppContext) {
        this.eventListener = sourceEventListener;
        String serverList = optionHolder.validateAndGetStaticValue(KafkaSource
            .ADAPTOR_SUBSCRIBER_ZOOKEEPER_CONNECT_SERVERS);
{'fraction_non_alphanumeric': 0.08703804984764435, 'fraction_numerical': 0.009844519103054927, 'mean_word_length': 2.0476190476190474, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1798834', 'n_tokens_mistral': 3344, 'n_tokens_neox': 3210, 'n_words': 993}