<commit_before>package com.bwsw.sj.engine.regular

import java.io.File
import java.net.{InetSocketAddress, URLClassLoader}
import java.util.Properties

import com.aerospike.client.Host
import com.bwsw.common.ObjectSerializer
import com.bwsw.common.tstream.NetworkTimeUUIDGenerator
import com.bwsw.sj.common.ConfigConstants._
import com.bwsw.sj.common.DAL.ConnectionConstants._
import com.bwsw.sj.common.DAL.model._
import com.bwsw.sj.common.DAL.repository.ConnectionRepository
import com.bwsw.sj.common.ModuleConstants._
import com.bwsw.sj.common.StreamConstants._
import com.bwsw.sj.engine.core.PersistentBlockingQueue
import com.bwsw.sj.engine.core.converter.ArrayByteConverter
import com.bwsw.sj.engine.core.environment.ModuleOutput
import com.bwsw.tstreams.agents.consumer.Offsets.{IOffset, Newest}
import com.bwsw.tstreams.agents.consumer.subscriber.BasicSubscribingConsumer
import com.bwsw.tstreams.agents.consumer.{BasicConsumer, BasicConsumerOptions, SubscriberCoordinationOptions}
import com.bwsw.tstreams.agents.producer.InsertionType.SingleElementInsert
import com.bwsw.tstreams.agents.producer.{BasicProducer, BasicProducerOptions, ProducerCoordinationOptions}
import com.bwsw.tstreams.coordination.transactions.transport.impl.TcpTransport
import com.bwsw.tstreams.data.IStorage
import com.bwsw.tstreams.data.aerospike.{AerospikeStorageFactory, AerospikeStorageOptions}
import com.bwsw.tstreams.data.cassandra.{CassandraStorageFactory, CassandraStorageOptions}
import com.bwsw.tstreams.generator.LocalTimeUUIDGenerator
import com.bwsw.tstreams.metadata.{MetadataStorage, MetadataStorageFactory}
import com.bwsw.tstreams.policy.RoundRobinPolicy
import com.bwsw.tstreams.services.BasicStreamService
import com.bwsw.tstreams.streams.BasicStream
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.TopicPartition
import org.slf4j.LoggerFactory

import scala.collection.mutable

/**
  * Class allowing to manage an environment of task
  * Created: 13/04/2016
  *
  * @author <NAME>
  */
class TaskManager() {
  private val logger = LoggerFactory.getLogger(this.getClass)

  val instanceName = System.getenv("INSTANCE_NAME")
  val agentsHost = System.getenv("AGENTS_HOST")
  private val agentsPorts = System.getenv("AGENTS_PORTS").split(",")
  val taskName = System.getenv("TASK_NAME")
  var kafkaOffsetsStorage = mutable.Map[(String, Int), Long]()
  private val kafkaOffsetsStream = taskName + "_kafka_offsets"
  private val stateStream = taskName + "_state"
  private val reportStream = instanceName + "_report"
  private var currentPortNumber = 0
  private val instance = ConnectionRepository.getInstanceService.get(instanceName)
  private val storage = ConnectionRepository.getFileStorage
  private val configService = ConnectionRepository.getConfigService

  private val txnPreload = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> txnPreloadTag)
  ).head.value.toInt
  private val dataPreload = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> dataPreloadTag)
  ).head.value.toInt
  private val consumerKeepAliveInterval = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> consumerKeepAliveInternalTag)
  ).head.value.toInt
  private val transportTimeout = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> transportTimeoutTag)
  ).head.value.toInt
  private val txnTTL = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> txnTTLTag)
  ).head.value.toInt
  private val txnKeepAliveInterval = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> txnKeepAliveIntervalTag)
  ).head.value.toInt
  private val producerKeepAliveInterval = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> producerKeepAliveIntervalTag)
  ).head.value.toInt
  private val streamTTL = configService.getByParameters(
    Map("domain" -> "t-streams", "name" -> streamTTLTag)
  ).head.value.toInt

  private val zkTimeout = configService.getByParameters(
    Map("domain" -> "zk", "name" -> zkSessionTimeoutTag)
  ).head.value.toInt

  assert(agentsPorts.length == (instance.inputs.length + instance.outputs.length + 3),
    "Not enough ports for t-stream consumers/producers ")

  private val fileMetadata: FileMetadata = ConnectionRepository.getFileMetadataService.getByParameters(
    Map("specification.name" -> instance.moduleName,<commit_msg>SJ-2007 hot fix<commit_after>package com.bwsw.sj.engine.regular

import java.io.File
import java.net.{InetSocketAddress, URLClassLoader}
import java.util.Properties

import com.aerospike.client.Host
import com.bwsw.common.ObjectSerializer
import com.bwsw.common.tstream.NetworkTimeUUIDGenerator
import com.bwsw.sj.common.ConfigConstants._
import com.bwsw.sj.common.DAL.ConnectionConstants._
import com.bwsw.sj.common.DAL.model._
import com.bwsw.sj.common.DAL.repository.ConnectionRepository
import com.bwsw.sj.common.ModuleConstants._
import com.bwsw.sj.common.StreamConstants._
import com.bwsw.sj.engine.core.PersistentBlockingQueue
import com.bwsw.sj.engine.core.converter.ArrayByteConverter
import com.bwsw.sj.engine.core.environment.ModuleOutput
import com.bwsw.tstreams.agents.consumer.Offsets.{IOffset, Newest}
import com.bwsw.tstreams.agents.consumer.subscriber.BasicSubscribingConsumer
import com.bwsw.tstreams.agents.consumer.{BasicConsumer, BasicConsumerOptions, SubscriberCoordinationOptions}
import com.bwsw.tstreams.agents.producer.InsertionType.SingleElementInsert
import com.bwsw.tstreams.agents.producer.{BasicProducer, BasicProducerOptions, ProducerCoordinationOptions}
import com.bwsw.tstreams.coordination.transactions.transport.impl.TcpTransport
import com.bwsw.tstreams.data.IStorage
import com.bwsw.tstreams.data.aerospike.{AerospikeStorageFactory, AerospikeStorageOptions}
import com.bwsw.tstreams.data.cassandra.{CassandraStorageFactory, CassandraStorageOptions}
import com.bwsw.tstreams.generator.LocalTimeUUIDGenerator
import com.bwsw.tstreams.metadata.{MetadataStorage, MetadataStorageFactory}
import com.bwsw.tstreams.policy.RoundRobinPolicy
import com.bwsw.tstreams.services.BasicStreamService
import com.bwsw.tstreams.streams.BasicStream
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.TopicPartition
import org.slf4j.LoggerFactory

import scala.collection.mutable

/**
  * Class allowing to manage an environment of task
  * Created: 13/04/2016
  *
  * @author <NAME>
  */
class TaskManager() {
  private val logger = LoggerFactory.getLogger(this.getClass)

  val instanceName = System.getenv("INSTANCE_NAME")
  val agentsHost = System.getenv("AGENTS_HOST")
  private val agentsPorts = System.getenv("AGENTS_PORTS").split(",")
  val taskName = System.getenv("TASK_NAME")
  var kafkaOffsetsStorage = mutable.Map[(String, Int), Long]()
  private val kafkaOffsetsStream = taskName + "_kafka_offsets"
  private val stateStream = taskName + "_state"
  private val reportStream = instanceName + "_report"
  private var currentPortNumber = 0
  private val instance = ConnectionRepository.getInstanceService.get(instanceName)
  private val storage = ConnectionRepository.getFileStorage
  private val configService = ConnectionRepository.getConfigService

  private val txnPreload = configService.get(txnPreloadTag).value.toInt
  private val dataPreload = configService.get(dataPreloadTag).value.toInt
  private val consumerKeepAliveInterval = configService.get(consumerKeepAliveInternalTag).value.toInt
  private val transportTimeout = configService.get(transportTimeoutTag).value.toInt
  private val txnTTL = configService.get(txnTTLTag).value.toInt
  private val txnKeepAliveInterval = configService.get(txnKeepAliveIntervalTag).value.toInt
  private val producerKeepAliveInterval = configService.get(producerKeepAliveIntervalTag).value.toInt
  private val streamTTL = configService.get(streamTTLTag).value.toInt

  private val zkTimeout = configService.get(zkSessionTimeoutTag).value.toInt

  assert(agentsPorts.length == (instance.inputs.length + instance.outputs.length + 3),
    "Not enough ports for t-stream consumers/producers ")

  private val fileMetadata: FileMetadata = ConnectionRepository.getFileMetadataService.getByParameters(
    Map("specification.name" -> instance.moduleName,
{'fraction_non_alphanumeric': 0.10024125452352231, 'fraction_numerical': 0.0031363088057901087, 'mean_word_length': 5.978956228956229, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 25, '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': '5111170', 'n_tokens_mistral': 2589, 'n_tokens_neox': 2536, 'n_words': 440}