<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 io.siddhi.extension.io.kafka.source;

import io.siddhi.core.stream.input.source.SourceEventListener;
import io.siddhi.extension.io.kafka.Constants;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static io.siddhi.extension.io.kafka.source.KafkaSource.ADAPTOR_ENABLE_AUTO_COMMIT;

/**
 * This runnable processes each Kafka message and sends it to siddhi.
 */
public class KafkaReplayThread implements Runnable {

    private static final Logger LOG = Logger.getLogger(KafkaReplayThread.class);
    private final KafkaConsumer<byte[], byte[]> consumer;
    // KafkaConsumer is not thread safe, hence we need a lock
    private final Lock consumerLock = new ReentrantLock();
    private final String partitions[];
    private SourceEventListener sourceEventListener;
    private String topics[];
    private volatile boolean paused;
    private volatile boolean inactive;
    private List<TopicPartition> partitionsList = new ArrayList<>();
    private String consumerThreadId;
    private boolean isPartitionWiseThreading = false;
    private boolean isBinaryMessage = false;
    private boolean enableOffsetCommit = false;
    private boolean enableAutoCommit = false;
    private boolean enableAsyncCommit;
    private boolean consumerClosed;
    private ReentrantLock lock;
    private Condition condition;
    private KafkaSource.KafkaSourceState kafkaSourceState;
    private String[] requiredProperties;
    private int trpLength;
    private int startOffset;
    private int endOffset;

    KafkaReplayThread(SourceEventListener sourceEventListener, String[] topics, String[] partitions,
                        Properties props, boolean isPartitionWiseThreading, boolean isBinaryMessage,
                        boolean enableOffsetCommit, boolean enableAsyncCommit, String[] requiredProperties,
                      int startOffset, int endOffset) {
        this.startOffset = startOffset;
        this.endOffset = endOffset;
        this.consumer = new KafkaConsumer<>(props);
        this.sourceEventListener = sourceEventListener;
        this.topics = topics;
        this.partitions = partitions;
        this.isPartitionWiseThreading = isPartitionWiseThreading;
        this.isBinaryMessage = isBinaryMessage;
        this.enableOffsetCommit = false;
        this.enableAutoCommit = Boolean.parseBoolean(props.getProperty(ADAPTOR_ENABLE_AUTO_COMMIT, "true"));
        this.consumerThreadId = buildId();
        this.enableAsyncCommit = enableAsyncCommit;
        lock = new ReentrantLock();
        condition = lock.newCondition();
        this.requiredProperties = requiredProperties;
        if (requiredProperties != null && requiredProperties.length > 0) {
            trpLength = requiredProperties.length;
        } else {
            trpLength = 0;
        }
        if (null != partitions) {
            for (String topic : topics) {
                for (String partition1 : partitions) {
                    TopicPartition partition = new TopicPartition(topic, Integer.parseInt(partition1));
                    LOG.info("Adding partition " + partition1 + " for topic: " + topic);
                    partitionsList.add(partition);
                }
                LOG.info("Adding partitions " + Arrays.toString(partitions) + " for topic: " + topic);
                consumer.assign(partitionsList);
            }
        } else {
            consumer.subscribe(Arrays.asList(topics));
        }
        consumerClosed = false;
        LOG.info("Subscribed for topics: " + Arrays.toString(topics));
    }

    void pause() {
        paused = true;
    }

    void resume() {
//        restore();
//        paused = false;
//        try {
//            lock.lock();
//            condition.signalAll();
//        } finally {
//            lock.unlock();
//        }
    }
//
//    void restore() {
//        final Lock consumerLock = this.consumerLock;
//        if (kafkaSourceState != null && kafkaSourceState.getTopicOffsetMap() != null) {
//            for (String topic : topics) {
//                Map<Integer, Long> offsetMap = kafkaSourceState.getTopicOffsetMap().get(topic);
//                if (null != offsetMap) {
//                    for (Map.Entry<Integer, Long> entry : offsetMap.entrySet()) {
//                        TopicPartition partition = new TopicPartition(topic, entry.getKey());
//                        if (partitionsList.contains(partition)) {
//                            LOG.info("Seeking partition: " + partition + " for topic: " + topic + " offset: " + (entry
//                                    .getValue() + 1));
//                            try {
//                                consumerLock.lock();
//                                consumer.seek(partition, entry.getValue() + 1); //todo check this
//                            } finally {
//                                consumerLock.unlock();
//                            }
//                        }
//                    }
//                }
//            }
//        }
//    }

    @Override
    public void run() {
        final Lock consumerLock = this.consumerLock;
        while (!inactive) {
            if (paused) {
                lock.lock();
                try {
                    condition.await();
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                } finally {
                    lock.unlock();
                }
            }
            // The time, in milliseconds, spent waiting in poll if data is not available. If 0, returns
            // immediately with any records that are available now. Must not be negative
            ConsumerRecords<byte[], byte[]> records = null;
            try {
                consumerLock.lock();
                // TODO add a huge value because, when there are so many equal group ids, the group balancing
                // takes time and if this value is small, there will be an CommitFailedException while
                // trying to retrieve data
                consumer.seekToBeginning(partitionsList);
                records = consumer.poll(100); // todo check
            } catch (CommitFailedException ex) {
                LOG.warn("Consumer poll() failed." + ex.getMessage(), ex);
            } finally {
                consumerLock.unlock();
            }
            if (null != records) {
//                Map<SequenceKey, Integer> lastReceivedSeqNoMap = null;
//                if (kafkaSourceState.getConsumerLastReceivedSeqNoMap() != null) {
//                    lastReceivedSeqNoMap = kafkaSourceState.getConsumerLastReceivedSeqNoMap().get(consumerThreadId);
//                }
                for (ConsumerRecord record : records) {
                    String[] trpProperties = new String[trpLength];
                    if (!consumerClosed) {
                        if (record.offset() >= startOffset) {
                            if (record.offset() > endOffset) {
                                Thread.currentThread().interrupt();
                                break;
                            }
                            int partition = record.partition();
                            Object event = record.value();
                            Object eventBody = null;
                            String header = null;
                            long eventTimestamp = System.currentTimeMillis();
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Event received in Kafka Event Adaptor with offSet: " + record.offset() +
                                        ", key: " + record.key() + ", topic: " + record.topic() +
                                        ", partition: " + partition + ", recordTimestamp: " + record.timestamp() +
                                        ", eventTimestamp: " + eventTimestamp + ", checksum: " + record.checksum());
                            }
                            for (int i = 0; i < requiredProperties.length; i++) {
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_PARTITION)) {
                                    trpProperties[i] = String.valueOf(record.partition());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_TOPIC)) {
                                    trpProperties[i] = record.topic();
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_KEY)) {
                                    trpProperties[i] = String.valueOf(record.key());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_RECORD_TIMESTAMP)) {
                                    trpProperties[i] = String.valueOf(record.timestamp());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_EVENT_TIMESTAMP)) {
                                    trpProperties[i] = String.valueOf(eventTimestamp);
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_CHECK_SUM)) {
                                    trpProperties[i] = String.valueOf(record.checksum());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_OFFSET)) {
                                    trpProperties[i] = String.valueOf(record.offset());
                                    //todo check for end offset and break the loop
                                }
                            }
                            String transportSyncProperties = "topic:" + record.topic() + ",partition:"
                                    + record.partition() + ",offSet:" + record.offset();
                            String[] transportSyncPropertiesArr = new String[]{transportSyncProperties};
                            sourceEventListener.onEvent(event, trpProperties, transportSyncPropertiesArr);
                            if (record.offset() == endOffset) {
                                Thread.currentThread().interrupt();
                            }
//                        if (lastReceivedSeqNoMap == null) {
//                            sourceEventListener.onEvent(event, trpProperties, transportSyncPropertiesArr);
//                        } else {<commit_msg>Shutdown consumer when interrupting thread<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 io.siddhi.extension.io.kafka.source;

import io.siddhi.core.stream.input.source.SourceEventListener;
import io.siddhi.extension.io.kafka.Constants;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static io.siddhi.extension.io.kafka.source.KafkaSource.ADAPTOR_ENABLE_AUTO_COMMIT;

/**
 * This runnable processes each Kafka message and sends it to siddhi.
 */
public class KafkaReplayThread implements Runnable {

    private static final Logger LOG = Logger.getLogger(KafkaReplayThread.class);
    private final KafkaConsumer<byte[], byte[]> consumer;
    // KafkaConsumer is not thread safe, hence we need a lock
    private final Lock consumerLock = new ReentrantLock();
    private final String partitions[];
    private SourceEventListener sourceEventListener;
    private String topics[];
    private volatile boolean paused;
    private volatile boolean inactive;
    private List<TopicPartition> partitionsList = new ArrayList<>();
    private String consumerThreadId;
    private boolean isPartitionWiseThreading = false;
    private boolean isBinaryMessage = false;
    private boolean enableOffsetCommit = false;
    private boolean enableAutoCommit = false;
    private boolean enableAsyncCommit;
    private boolean consumerClosed;
    private ReentrantLock lock;
    private Condition condition;
    private KafkaSource.KafkaSourceState kafkaSourceState;
    private String[] requiredProperties;
    private int trpLength;
    private int startOffset;
    private int endOffset;

    KafkaReplayThread(SourceEventListener sourceEventListener, String[] topics, String[] partitions,
                        Properties props, boolean isPartitionWiseThreading, boolean isBinaryMessage,
                        boolean enableOffsetCommit, boolean enableAsyncCommit, String[] requiredProperties,
                      int startOffset, int endOffset) {
        this.startOffset = startOffset;
        this.endOffset = endOffset;
        this.consumer = new KafkaConsumer<>(props);
        this.sourceEventListener = sourceEventListener;
        this.topics = topics;
        this.partitions = partitions;
        this.isPartitionWiseThreading = isPartitionWiseThreading;
        this.isBinaryMessage = isBinaryMessage;
        this.enableOffsetCommit = false;
        this.enableAutoCommit = Boolean.parseBoolean(props.getProperty(ADAPTOR_ENABLE_AUTO_COMMIT, "true"));
        this.consumerThreadId = buildId();
        this.enableAsyncCommit = enableAsyncCommit;
        lock = new ReentrantLock();
        condition = lock.newCondition();
        this.requiredProperties = requiredProperties;
        if (requiredProperties != null && requiredProperties.length > 0) {
            trpLength = requiredProperties.length;
        } else {
            trpLength = 0;
        }
        if (null != partitions) {
            for (String topic : topics) {
                for (String partition1 : partitions) {
                    TopicPartition partition = new TopicPartition(topic, Integer.parseInt(partition1));
                    LOG.info("Adding partition " + partition1 + " for topic: " + topic);
                    partitionsList.add(partition);
                }
                LOG.info("Adding partitions " + Arrays.toString(partitions) + " for topic: " + topic);
                consumer.assign(partitionsList);
            }
        } else {
            consumer.subscribe(Arrays.asList(topics));
        }
        consumerClosed = false;
        LOG.info("Subscribed for topics: " + Arrays.toString(topics));
    }

    void pause() {
        paused = true;
    }

    void resume() {
//        restore();
//        paused = false;
//        try {
//            lock.lock();
//            condition.signalAll();
//        } finally {
//            lock.unlock();
//        }
    }
//
//    void restore() {
//        final Lock consumerLock = this.consumerLock;
//        if (kafkaSourceState != null && kafkaSourceState.getTopicOffsetMap() != null) {
//            for (String topic : topics) {
//                Map<Integer, Long> offsetMap = kafkaSourceState.getTopicOffsetMap().get(topic);
//                if (null != offsetMap) {
//                    for (Map.Entry<Integer, Long> entry : offsetMap.entrySet()) {
//                        TopicPartition partition = new TopicPartition(topic, entry.getKey());
//                        if (partitionsList.contains(partition)) {
//                            LOG.info("Seeking partition: " + partition + " for topic: " + topic + " offset: " + (entry
//                                    .getValue() + 1));
//                            try {
//                                consumerLock.lock();
//                                consumer.seek(partition, entry.getValue() + 1); //todo check this
//                            } finally {
//                                consumerLock.unlock();
//                            }
//                        }
//                    }
//                }
//            }
//        }
//    }

    @Override
    public void run() {
        final Lock consumerLock = this.consumerLock;
        while (!inactive) {
            if (paused) {
                lock.lock();
                try {
                    condition.await();
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                } finally {
                    lock.unlock();
                }
            }
            // The time, in milliseconds, spent waiting in poll if data is not available. If 0, returns
            // immediately with any records that are available now. Must not be negative
            ConsumerRecords<byte[], byte[]> records = null;
            try {
                consumerLock.lock();
                // TODO add a huge value because, when there are so many equal group ids, the group balancing
                // takes time and if this value is small, there will be an CommitFailedException while
                // trying to retrieve data
                consumer.seekToBeginning(partitionsList);
                records = consumer.poll(100); // todo check
            } catch (CommitFailedException ex) {
                LOG.warn("Consumer poll() failed." + ex.getMessage(), ex);
            } finally {
                consumerLock.unlock();
            }
            if (null != records) {
//                Map<SequenceKey, Integer> lastReceivedSeqNoMap = null;
//                if (kafkaSourceState.getConsumerLastReceivedSeqNoMap() != null) {
//                    lastReceivedSeqNoMap = kafkaSourceState.getConsumerLastReceivedSeqNoMap().get(consumerThreadId);
//                }
                for (ConsumerRecord record : records) {
                    String[] trpProperties = new String[trpLength];
                    if (!consumerClosed) {
                        if (record.offset() >= startOffset) {
                            if (record.offset() > endOffset) {
                                shutdownConsumer();
                                Thread.currentThread().interrupt();
                                break;
                            }
                            int partition = record.partition();
                            Object event = record.value();
                            Object eventBody = null;
                            String header = null;
                            long eventTimestamp = System.currentTimeMillis();
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Event received in Kafka Event Adaptor with offSet: " + record.offset() +
                                        ", key: " + record.key() + ", topic: " + record.topic() +
                                        ", partition: " + partition + ", recordTimestamp: " + record.timestamp() +
                                        ", eventTimestamp: " + eventTimestamp + ", checksum: " + record.checksum());
                            }
                            for (int i = 0; i < requiredProperties.length; i++) {
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_PARTITION)) {
                                    trpProperties[i] = String.valueOf(record.partition());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_TOPIC)) {
                                    trpProperties[i] = record.topic();
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_KEY)) {
                                    trpProperties[i] = String.valueOf(record.key());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_RECORD_TIMESTAMP)) {
                                    trpProperties[i] = String.valueOf(record.timestamp());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_EVENT_TIMESTAMP)) {
                                    trpProperties[i] = String.valueOf(eventTimestamp);
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_CHECK_SUM)) {
                                    trpProperties[i] = String.valueOf(record.checksum());
                                }
                                if (requiredProperties[i].equalsIgnoreCase(Constants.TRP_OFFSET)) {
                                    trpProperties[i] = String.valueOf(record.offset());
                                    //todo check for end offset and break the loop
                                }
                            }
                            String transportSyncProperties = "topic:" + record.topic() + ",partition:"
                                    + record.partition() + ",offSet:" + record.offset();
                            String[] transportSyncPropertiesArr = new String[]{transportSyncProperties};
                            sourceEventListener.onEvent(event, trpProperties, transportSyncPropertiesArr);
                            if (record.offset() == endOffset) {
                                shutdownConsumer();
                                Thread.currentThread().interrupt();
                            }
//                        if (lastReceivedSeqNoMap == null) {
//                            sourceEventListener.onEvent(event, trpProperties, transportSyncPropertiesArr);
//                        } else {
{'fraction_non_alphanumeric': 0.09369611842381934, 'fraction_numerical': 0.002018587829597544, 'mean_word_length': 1.646044286191165, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 46, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1894412', 'n_tokens_mistral': 5615, 'n_tokens_neox': 5416, 'n_words': 1328}