#555

Build # #555

Build Type

push

Committed by

Commit Message

<a href="https://github.com/evolution-gaming/skafka/commit/<a class=hub.com/evolution-gaming/skafka/commit/<a class="double-link" href="https://git"><a class=hub.com/evolution-gaming/skafka/commit/<a class="double-link" href="https://git"><a class=hub.com/evolution-gaming/skafka/commit/<a class="double-link" href="https://git"><a class=hub.com/evolution-gaming/skafka/commit/eed6d61be4e94f1cef73986198af8596e5aac480">eed6d61be">&lt;a href=&quot;https://github.com/evolution-gaming/skafka/commit/</a><a class="double-link" href="https://github.com/evolution-gaming/skafka/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/evolution-gaming/skafka/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/evolution-gaming/skafka/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/evolution-gaming/skafka/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/evolution-gaming/skafka/commit/eed6d61be4e94f1cef73986198af8596e5aac480">eed6d61be</a><a href="https://github.com/evolution-gaming/skafka/commit/eed6d61be4e94f1cef73986198af8596e5aac480">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/evolution-gaming/skafka/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/evolution-gaming/skafka/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/evolution-gaming/skafka/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/evolution-gaming/skafka/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/evolution-gaming/skafka/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/evolution-gaming/skafka/commit/eed6d61be4e94f1cef73986198af8596e5aac480&quot;&gt;eed6d61be&lt;/a&gt;&lt;a href=&quot;https://github.com/evolution-gaming/skafka/commit/eed6d61be4e94f1cef73986198af8596e5aac480&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/evolution-gaming/skafka/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot... (continued)

Run Details

107 of 107 new or added lines in 6 files covered. (100.0%)

1050 of 1217 relevant lines covered (86.28%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.06

/skafka/src/main/scala/com/evolutiongaming/skafka/producer/ProducerConfig.scala

package com.evolutiongaming.skafka.producer

import com.evolutiongaming.config.ConfigHelper._
import com.evolutiongaming.skafka.CommonConfig
import com.typesafe.config.{Config, ConfigException}
import org.apache.kafka.clients.producer.{Partitioner, ProducerConfig => C}

import scala.concurrent.duration._
import scala.util.Try

/**
  * Check [[http://kafka.apache.org/documentation/#producerconfigs]]
  */
final case class ProducerConfig(
  common: CommonConfig = CommonConfig.Default,
  batchSize: Int = 16384,
  deliveryTimeout: FiniteDuration = 2.minutes,
  acks: Acks = Acks.One,
  linger: FiniteDuration = 0.millis,
  maxRequestSize: Int = 1048576,
  maxBlock: FiniteDuration = 1.minute,
  bufferMemory: Long = 33554432L,
  compressionType: CompressionType = CompressionType.None,
  retries: Int = Int.MaxValue,
  maxInFlightRequestsPerConnection: Int = 5,
  partitionerClass: Option[Class[_ <: Partitioner]] = None,
  interceptorClasses: List[String] = Nil,
  idempotence: Boolean = false,
  transactionTimeout: FiniteDuration = 1.minute,
  transactionalId: Option[String] = None) {

  def bindings: Map[String, AnyRef] = {
    val bindings1 = Map[String, String](
      (C.BATCH_SIZE_CONFIG, batchSize.toString),
      (C.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeout.toMillis.toString),
      (C.ACKS_CONFIG, acks.names.head.toString),
      (C.LINGER_MS_CONFIG, linger.toMillis.toString),
      (C.MAX_REQUEST_SIZE_CONFIG, maxRequestSize.toString),
      (C.MAX_BLOCK_MS_CONFIG, maxBlock.toMillis.toString),
      (C.BUFFER_MEMORY_CONFIG, bufferMemory.toString),
      (C.COMPRESSION_TYPE_CONFIG, compressionType.toString.toLowerCase),
      (C.RETRIES_CONFIG, retries.toString),
      (C.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, maxInFlightRequestsPerConnection.toString),
      (C.INTERCEPTOR_CLASSES_CONFIG, interceptorClasses mkString ","),
      (C.ENABLE_IDEMPOTENCE_CONFIG, idempotence.toString),
      (C.TRANSACTION_TIMEOUT_CONFIG, transactionTimeout.toMillis.toString))

    val bindings = bindings1 ++
      transactionalId.map { (C.TRANSACTIONAL_ID_CONFIG, _) }
      partitionerClass.map { value => (C.PARTITIONER_CLASS_CONFIG, value) }

    bindings ++ common.bindings
  }

  def properties: java.util.Properties = {
    val properties = new java.util.Properties
    bindings foreach { case (k, v) => properties.put(k, v) }
    properties
  }
}

object ProducerConfig {

  val Default: ProducerConfig = ProducerConfig()

  private implicit val CompressionTypeFromConf = FromConf[CompressionType] { (conf, path) =>
    val str = conf.getString(path)
    val value = CompressionType.Values.find { _.toString equalsIgnoreCase str }
    value getOrElse {
      throw new ConfigException.BadValue(conf.origin(), path, s"Cannot parse CompressionType from $str")
    }
  }

  private implicit val AcksFromConf = FromConf[Acks] { (conf, path) =>
    val str = conf.getString(path)

    val values = Acks.Values.filter(_.names.exists(str.equalsIgnoreCase))

    values.headOption getOrElse {
      throw new ConfigException.BadValue(conf.origin(), path, s"Cannot parse Acks from $str")
    }
  }

  def apply(config: Config): ProducerConfig = {
    apply(config, Default)
  }

  def apply(config: Config, default: => ProducerConfig): ProducerConfig = {

    def get[T: FromConf](path: String, paths: String*) = {
      config.getOpt[T](path, paths: _*)
    }

    def getDuration(path: String, pathMs: => String) = {
      val value = try get[FiniteDuration](path) catch { case _: ConfigException => None }
      value orElse get[Long](pathMs).map { _.millis }
    }

    val partitionerClass = {

      def classOf(name: String) = {

        def classOfClassLoader = {
          val thread = Thread.currentThread()
          val classLoader = thread.getContextClassLoader
          Try { classLoader.loadClass(name) }
        }

        def classOf = {
          Try { Class.forName(name) }
        }

        for {
          a <- classOfClassLoader orElse classOf
          a <- Try { a.asInstanceOf[Class[Partitioner]] }
        } yield a
      }

      for {
        name             <- get[String]("partitioner-class", "partitioner.class")
        partitionerClass <- classOf(name).toOption
      } yield partitionerClass
    }

    ProducerConfig(
      common = CommonConfig(config, default.common),
      acks = get[Acks]("acks") getOrElse default.acks,
      bufferMemory = get[Long](
        "buffer-memory",
        "buffer.memory") getOrElse default.bufferMemory,
      compressionType = get[CompressionType](
        "compression-type",
        "compression.type") getOrElse default.compressionType,
      retries = get[Int]("retries") getOrElse default.retries,
      batchSize = get[Int](
        "batch-size",
        "batch.size") getOrElse default.batchSize,
      deliveryTimeout = getDuration(
        "delivery-timeout",
        "delivery.timeout.ms") getOrElse default.deliveryTimeout,
      linger = getDuration(
        "linger",
        "linger.ms") getOrElse default.linger,
      maxBlock = getDuration(
        "max-block",
        "max.block.ms") getOrElse default.maxBlock,
      maxRequestSize = get[Int](
        "max-request-size",
        "max.request.size") getOrElse default.maxRequestSize,
      maxInFlightRequestsPerConnection = get[Int](
        "max-in-flight-requests-per-connection",
        "max.in.flight.requests.per.connection") getOrElse default.maxInFlightRequestsPerConnection,
      partitionerClass = partitionerClass orElse default.partitionerClass,
      interceptorClasses = get[List[String]](
        "interceptor-classes",
        "interceptor.classes") getOrElse default.interceptorClasses,
      idempotence = get[Boolean](
        "idempotence",
        "enable-idempotence",
        "enable.idempotence") getOrElse default.idempotence,
      transactionTimeout = getDuration(
        "transaction-timeout",
        "transaction.timeout.ms") getOrElse default.transactionTimeout,
      transactionalId = get[String](
        "transactional-id",
        "transactional.id") orElse default.transactionalId)
  }
}

1	package com.evolutiongaming.skafka.producer
2
3	import com.evolutiongaming.config.ConfigHelper._
4	import com.evolutiongaming.skafka.CommonConfig
5	import com.typesafe.config.{Config, ConfigException}
6	import org.apache.kafka.clients.producer.{Partitioner, ProducerConfig => C}
7
8	import scala.concurrent.duration._
9	import scala.util.Try
10
11	/**
12	* Check [[http://kafka.apache.org/documentation/#producerconfigs]]
13	*/
14	final case class ProducerConfig(
15	common: CommonConfig = CommonConfig.Default,
16	batchSize: Int = 16384,
17	deliveryTimeout: FiniteDuration = 2.minutes,
18	acks: Acks = Acks.One,
19	linger: FiniteDuration = 0.millis,
20	maxRequestSize: Int = 1048576,
21	maxBlock: FiniteDuration = 1.minute,
22	bufferMemory: Long = 33554432L,
23	compressionType: CompressionType = CompressionType.None,
24	retries: Int = Int.MaxValue,
25	maxInFlightRequestsPerConnection: Int = 5,
26	partitionerClass: Option[Class[_ <: Partitioner]] = None,
27	interceptorClasses: List[String] = Nil,
28	idempotence: Boolean = false,
29	transactionTimeout: FiniteDuration = 1.minute,
30	transactionalId: Option[String] = None) {
31
32	def bindings: Map[String, AnyRef] = {
33	val bindings1 = Map[String, String](	1✔
34	(C.BATCH_SIZE_CONFIG, batchSize.toString),	1✔
35	(C.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeout.toMillis.toString),	1✔
36	(C.ACKS_CONFIG, acks.names.head.toString),	1✔
37	(C.LINGER_MS_CONFIG, linger.toMillis.toString),	1✔
38	(C.MAX_REQUEST_SIZE_CONFIG, maxRequestSize.toString),	1✔
39	(C.MAX_BLOCK_MS_CONFIG, maxBlock.toMillis.toString),	1✔
40	(C.BUFFER_MEMORY_CONFIG, bufferMemory.toString),	1✔
41	(C.COMPRESSION_TYPE_CONFIG, compressionType.toString.toLowerCase),	1✔
42	(C.RETRIES_CONFIG, retries.toString),	1✔
43	(C.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, maxInFlightRequestsPerConnection.toString),	1✔
44	(C.INTERCEPTOR_CLASSES_CONFIG, interceptorClasses mkString ","),	1✔
45	(C.ENABLE_IDEMPOTENCE_CONFIG, idempotence.toString),	1✔
46	(C.TRANSACTION_TIMEOUT_CONFIG, transactionTimeout.toMillis.toString))	1✔
47
48	val bindings = bindings1 ++	1✔
49	transactionalId.map { (C.TRANSACTIONAL_ID_CONFIG, _) }	1✔
50	partitionerClass.map { value => (C.PARTITIONER_CLASS_CONFIG, value) }	×
51
52	bindings ++ common.bindings	1✔
53	}
54
55	def properties: java.util.Properties = {
56	val properties = new java.util.Properties	1✔
57	bindings foreach { case (k, v) => properties.put(k, v) }	1✔
58	properties
59	}
60	}
61
62	object ProducerConfig {
63
64	val Default: ProducerConfig = ProducerConfig()	1✔
65
66	private implicit val CompressionTypeFromConf = FromConf[CompressionType] { (conf, path) =>	1✔
67	val str = conf.getString(path)	1✔
68	val value = CompressionType.Values.find { _.toString equalsIgnoreCase str }	1✔
69	value getOrElse {	1✔
70	throw new ConfigException.BadValue(conf.origin(), path, s"Cannot parse CompressionType from $str")	×
71	}
72	}
73
74	private implicit val AcksFromConf = FromConf[Acks] { (conf, path) =>	1✔
75	val str = conf.getString(path)	1✔
76
77	val values = Acks.Values.filter(_.names.exists(str.equalsIgnoreCase))	1✔
78
79	values.headOption getOrElse {	1✔
80	throw new ConfigException.BadValue(conf.origin(), path, s"Cannot parse Acks from $str")	×
81	}
82	}
83
84	def apply(config: Config): ProducerConfig = {
85	apply(config, Default)	1✔
86	}
87
88	def apply(config: Config, default: => ProducerConfig): ProducerConfig = {
89
90	def get[T: FromConf](path: String, paths: String*) = {
91	config.getOpt[T](path, paths: _*)	1✔
92	}
93
94	def getDuration(path: String, pathMs: => String) = {
95	val value = try get[FiniteDuration](path) catch { case _: ConfigException => None }	1✔
96	value orElse get[Long](pathMs).map { _.millis }	1✔
97	}
98
99	val partitionerClass = {
100
101	def classOf(name: String) = {
102
103	def classOfClassLoader = {
104	val thread = Thread.currentThread()	1✔
105	val classLoader = thread.getContextClassLoader	1✔
106	Try { classLoader.loadClass(name) }	1✔
107	}
108
109	def classOf = {
110	Try { Class.forName(name) }	×
111	}
112
113	for {
114	a <- classOfClassLoader orElse classOf	1✔
115	a <- Try { a.asInstanceOf[Class[Partitioner]] }	1✔
116	} yield a
117	}
118
119	for {
120	name <- get[String]("partitioner-class", "partitioner.class")	1✔
121	partitionerClass <- classOf(name).toOption	1✔
122	} yield partitionerClass
123	}
124
125	ProducerConfig(	1✔
126	common = CommonConfig(config, default.common),	1✔
127	acks = get[Acks]("acks") getOrElse default.acks,	1✔
128	bufferMemory = get[Long](	1✔
129	"buffer-memory",	1✔
130	"buffer.memory") getOrElse default.bufferMemory,	1✔
131	compressionType = get[CompressionType](	1✔
132	"compression-type",	1✔
133	"compression.type") getOrElse default.compressionType,	1✔
134	retries = get[Int]("retries") getOrElse default.retries,	1✔
135	batchSize = get[Int](	1✔
136	"batch-size",	1✔
137	"batch.size") getOrElse default.batchSize,	1✔
138	deliveryTimeout = getDuration(
139	"delivery-timeout",	1✔
140	"delivery.timeout.ms") getOrElse default.deliveryTimeout,	1✔
141	linger = getDuration(
142	"linger",	1✔
143	"linger.ms") getOrElse default.linger,	1✔
144	maxBlock = getDuration(
145	"max-block",	1✔
146	"max.block.ms") getOrElse default.maxBlock,	1✔
147	maxRequestSize = get[Int](	1✔
148	"max-request-size",	1✔
149	"max.request.size") getOrElse default.maxRequestSize,	1✔
150	maxInFlightRequestsPerConnection = get[Int](	1✔
151	"max-in-flight-requests-per-connection",	1✔
152	"max.in.flight.requests.per.connection") getOrElse default.maxInFlightRequestsPerConnection,	1✔
153	partitionerClass = partitionerClass orElse default.partitionerClass,	1✔
154	interceptorClasses = get[List[String]](	1✔
155	"interceptor-classes",	1✔
156	"interceptor.classes") getOrElse default.interceptorClasses,	1✔
157	idempotence = get[Boolean](	1✔
158	"idempotence",	1✔
159	"enable-idempotence",	1✔
160	"enable.idempotence") getOrElse default.idempotence,	1✔
161	transactionTimeout = getDuration(
162	"transaction-timeout",	1✔
163	"transaction.timeout.ms") getOrElse default.transactionTimeout,	1✔
164	transactionalId = get[String](	1✔
165	"transactional-id",	1✔
166	"transactional.id") orElse default.transactionalId)	1✔
167	}
168	}

evolution-gaming / skafka / #555

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous