<commit_before>/**
  * Copyright 2015 Yahoo Inc. Licensed under the Apache License, Version 2.0
  * See accompanying LICENSE file.
  */

package kafka.manager.actor.cluster

import java.io.Closeable
import java.net.InetAddress
import java.nio.ByteBuffer
import java.util.Properties
import java.util.concurrent.{ConcurrentLinkedDeque, TimeUnit}

import akka.actor.{ActorContext, ActorPath, ActorRef, Props}
import akka.pattern._
import com.github.benmanes.caffeine.cache.{Cache, Caffeine, RemovalCause, RemovalListener}
import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache}
import grizzled.slf4j.Logging
import kafka.admin.AdminClient
import kafka.api.PartitionOffsetRequestInfo
import kafka.common.{OffsetAndMetadata, TopicAndPartition}
import kafka.manager._
import kafka.manager.base.cluster.{BaseClusterQueryActor, BaseClusterQueryCommandActor}
import kafka.manager.base.{LongRunningPoolActor, LongRunningPoolConfig}
import kafka.manager.features.{ClusterFeatures, KMDeleteTopicFeature, KMPollConsumersFeature}
import kafka.manager.model.ActorModel._
import kafka.manager.model._
import kafka.manager.utils.ZkUtils
import kafka.manager.utils.zero81.{PreferredReplicaLeaderElectionCommand, ReassignPartitionCommand}
import kafka.manager.utils.one10.{GroupMetadata, GroupMetadataKey, MemberMetadata, OffsetKey}
import org.apache.curator.framework.CuratorFramework
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode
import org.apache.curator.framework.recipes.cache._
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.consumer.{Consumer, ConsumerRecords, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.requests.DescribeGroupsResponse
import org.joda.time.{DateTime, DateTimeZone}

import scala.collection.JavaConversions.{mapAsScalaMap, _}
import scala.collection.concurrent.TrieMap
import scala.collection.immutable.Map
import scala.collection.mutable
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success, Try}
import org.apache.kafka.clients.consumer.ConsumerConfig._
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
import org.apache.kafka.common.protocol.Errors

/**
  * @author hiral
  */
import kafka.manager.utils._

import scala.collection.JavaConverters._

case class KafkaAdminClientActorConfig(clusterContext: ClusterContext, longRunningPoolConfig: LongRunningPoolConfig, kafkaStateActorPath: ActorPath, consumerProperties: Option[Properties])
case class KafkaAdminClientActor(config: KafkaAdminClientActorConfig) extends BaseClusterQueryActor with LongRunningPoolActor {

  private[this] var adminClientOption : Option[AdminClient] = None

  protected implicit val clusterContext: ClusterContext = config.clusterContext
  override protected def longRunningPoolConfig: LongRunningPoolConfig = config.longRunningPoolConfig
<commit_msg>Fail fast if required properties are missing from consumer properties<commit_after>/**
  * Copyright 2015 Yahoo Inc. Licensed under the Apache License, Version 2.0
  * See accompanying LICENSE file.
  */

package kafka.manager.actor.cluster

import java.io.Closeable
import java.net.InetAddress
import java.nio.ByteBuffer
import java.util.Properties
import java.util.concurrent.{ConcurrentLinkedDeque, TimeUnit}

import akka.actor.{ActorContext, ActorPath, ActorRef, Props}
import akka.pattern._
import com.github.benmanes.caffeine.cache.{Cache, Caffeine, RemovalCause, RemovalListener}
import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache}
import grizzled.slf4j.Logging
import kafka.admin.AdminClient
import kafka.api.PartitionOffsetRequestInfo
import kafka.common.{OffsetAndMetadata, TopicAndPartition}
import kafka.manager._
import kafka.manager.base.cluster.{BaseClusterQueryActor, BaseClusterQueryCommandActor}
import kafka.manager.base.{LongRunningPoolActor, LongRunningPoolConfig}
import kafka.manager.features.{ClusterFeatures, KMDeleteTopicFeature, KMPollConsumersFeature}
import kafka.manager.model.ActorModel._
import kafka.manager.model._
import kafka.manager.utils.ZkUtils
import kafka.manager.utils.zero81.{PreferredReplicaLeaderElectionCommand, ReassignPartitionCommand}
import kafka.manager.utils.one10.{GroupMetadata, GroupMetadataKey, MemberMetadata, OffsetKey}
import org.apache.curator.framework.CuratorFramework
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode
import org.apache.curator.framework.recipes.cache._
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.consumer.{Consumer, ConsumerRecords, KafkaConsumer}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.requests.DescribeGroupsResponse
import org.joda.time.{DateTime, DateTimeZone}

import scala.collection.JavaConversions.{mapAsScalaMap, _}
import scala.collection.concurrent.TrieMap
import scala.collection.immutable.Map
import scala.collection.mutable
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success, Try}
import org.apache.kafka.clients.consumer.ConsumerConfig._
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
import org.apache.kafka.common.protocol.Errors

/**
  * @author hiral
  */
import kafka.manager.utils._

import scala.collection.JavaConverters._

case class KafkaAdminClientActorConfig(clusterContext: ClusterContext, longRunningPoolConfig: LongRunningPoolConfig, kafkaStateActorPath: ActorPath, consumerProperties: Option[Properties]) {
  validate()

  def validate(): Unit = {
    // Fail if required properties are missing
    consumerProperties.foreach(p => {
      require(p.contains(BOOTSTRAP_SERVERS_CONFIG), s"$BOOTSTRAP_SERVERS_CONFIG is missing from consumer properties")
      require(p.contains(KEY_DESERIALIZER_CLASS_CONFIG), s"$KEY_DESERIALIZER_CLASS_CONFIG is missing from consumer properties")
      require(p.contains(VALUE_DESERIALIZER_CLASS_CONFIG), s"$VALUE_DESERIALIZER_CLASS_CONFIG is missing from consumer properties")
      require(p.contains(GROUP_ID_CONFIG), s"$GROUP_ID_CONFIG is missing from consumer properties")
    })
  }

}
case class KafkaAdminClientActor(config: KafkaAdminClientActorConfig) extends BaseClusterQueryActor with LongRunningPoolActor {

  private[this] var adminClientOption : Option[AdminClient] = None

  protected implicit val clusterContext: ClusterContext = config.clusterContext
  override protected def longRunningPoolConfig: LongRunningPoolConfig = config.longRunningPoolConfig

{'fraction_non_alphanumeric': 0.08325666973321068, 'fraction_numerical': 0.0033731984053971173, 'mean_word_length': 7.103105590062111, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, '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': '975036', 'n_tokens_mistral': 1880, 'n_tokens_neox': 1837, 'n_words': 370}