#563

Build # #563

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/d835877bba502cdf4ec48a3b12f3f3630fe64901">d835877bb<a href="https://github.com/evolution-gaming/skafka/commit/d835877bba502cdf4ec48a3b12f3f3630fe64901">&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/d835877bba502cdf4ec48a3b12f3f3630fe64901&quot;&gt;d835877bb&lt;/a&gt;&lt;a href=&quot;https://github.com/evolution-gaming/skafka/commit/d835877bba502cdf4ec48a3b12f3f3630fe64901&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/evolution-gaming/skafka/commit/d835877bba502cdf4ec48a3b12f3f363&lt;/a&gt;0fe64901&quot;>&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/evolution-gaming/skafka/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/evolution-gaming/skafka/commit/<a class="double-link" href="https://github.com/evolution-gaming/skafka/commit/3654a05794713a1531945db469498e90271d0867">3654a0579</a><a href="https://github.com/evolution-gaming/skafka/commit/d835877bba502cdf4ec48a3b12f3f3630fe64901">&amp;quot;&amp;gt;3654a0579&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;3654a0579&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/evolution-gaming/skafka/commit/d835877bba502cdf4ec48a3b12f3f3630fe64&lt;/a&gt;901&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/evolution-gaming/skafka/commit/<a class="double-link" href="https://github.com/evolution-gaming/skafka/commit/b76b257d5">b76b257d5">b76b257d5</a>

Run Details

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

1042 of 1239 relevant lines covered (84.1%)

0.84 hits per line

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

80.0

/skafka/src/main/scala/com/evolutiongaming/skafka/consumer/ConsumerLogging.scala

package com.evolutiongaming.skafka.consumer

import java.util.regex.Pattern

import cats.Monad
import cats.data.{NonEmptySet => Nes, NonEmptyMap => Nem}
import cats.implicits._
import com.evolutiongaming.catshelper.Log
import com.evolutiongaming.skafka.{Offset, OffsetAndMetadata, Topic, TopicPartition}
import com.evolutiongaming.smetrics.MeasureDuration

import scala.concurrent.duration.FiniteDuration

object ConsumerLogging {

  def apply[F[_] : Monad : MeasureDuration, K, V](
    log: Log[F],
    consumer: Consumer[F, K, V]
  ): Consumer[F, K, V] = {

    new Consumer[F, K, V] {

      def assign(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.assign(partitions)
          d <- d
          _ <- log.debug(s"assign in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }")
        } yield a
      }

      val assignment = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.assignment
          d <- d
          _ <- log.debug(s"assignment in ${ d.toMillis }ms, result: ${ a.toList.mkString_(", ") }")
        } yield a
      }

      // TODO RebalanceListener1 implement logging - https://github.com/evolution-gaming/skafka/issues/127
      def subscribe(topics: Nes[Topic], listener: RebalanceListener1[F]) = consumer.subscribe(topics, listener)

      // TODO RebalanceListener1 implement logging - https://github.com/evolution-gaming/skafka/issues/127
      def subscribe(pattern: Pattern, listener: RebalanceListener1[F]) = consumer.subscribe(pattern, listener)

      def subscribe(topics: Nes[Topic], listener: Option[RebalanceListener[F]]) = {

        val listenerLogging = (listener getOrElse RebalanceListener.empty[F]).withLogging(log)

        for {
          d <- MeasureDuration[F].start
          a <- consumer.subscribe(topics, listenerLogging.some)
          d <- d
          _ <- log.debug(s"subscribe in ${ d.toMillis }ms, topics: ${ topics.mkString_(", ") }, listener: ${ listener.isDefined }")
        } yield a
      }

      def subscribe(pattern: Pattern, listener: Option[RebalanceListener[F]]) = {

        val listenerLogging = (listener getOrElse RebalanceListener.empty[F]).withLogging(log)

        for {
          d <- MeasureDuration[F].start
          a <- consumer.subscribe(pattern, listenerLogging.some)
          d <- d
          _ <- log.debug(s"subscribe in ${ d.toMillis }ms, pattern: $pattern, listener: ${ listener.isDefined }")
        } yield a
      }

      val subscription = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.subscription
          d <- d
          _ <- log.debug(s"subscription in ${ d.toMillis }ms, result: ${ a.toList.mkString_(", ") }")
        } yield a
      }

      val unsubscribe = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.unsubscribe
          d <- d
          _ <- log.debug(s"unsubscribe in ${ d.toMillis }ms")
        } yield a
      }

      def poll(timeout: FiniteDuration) = {

        def show(records: ConsumerRecords[K, V]) = {
          ConsumerRecords.summaryShow.show(records)
        }

        for {
          d <- MeasureDuration[F].start
          a <- consumer.poll(timeout)
          d <- d
          _ <- log.debug(s"poll timeout: $timeout in ${ d.toMillis }ms, result: ${ show(a) }")
        } yield a
      }

      val commit = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.commit
          d <- d
          _ <- log.debug(s"commit in ${ d.toMillis }ms")
        } yield a
      }

      def commit(timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.commit(timeout)
          d <- d
          _ <- log.debug(s"commit in ${ d.toMillis }ms, timeout: $timeout")
        } yield a
      }

      def commit(offsets: Nem[TopicPartition, OffsetAndMetadata]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.commit(offsets)
          d <- d
          _ <- log.debug(s"commit in ${ d.toMillis }ms, offsets: ${ offsets.mkString_(", ") }")
        } yield a
      }

      def commit(offsets: Nem[TopicPartition, OffsetAndMetadata], timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.commit(offsets, timeout)
          d <- d
          _ <- log.debug(s"commit in ${ d.toMillis }ms, offsets: ${ offsets.mkString_(", ") }, timeout: $timeout")
        } yield a
      }

      val commitLater = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.commitLater
          d <- d
          _ <- log.debug(s"commitLater in ${ d.toMillis }ms")
        } yield a
      }

      def commitLater(offsets: Nem[TopicPartition, OffsetAndMetadata]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.commitLater(offsets)
          d <- d
          _ <- log.debug(s"commitLater in ${ d.toMillis }ms, offsets: ${ offsets.mkString_(", ") }")
        } yield a
      }

      def seek(partition: TopicPartition, offset: Offset) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.seek(partition, offset)
          d <- d
          _ <- log.debug(s"seek in ${ d.toMillis }ms, partition: $partition, offset: $offset")
        } yield a
      }

      def seek(partition: TopicPartition, offsetAndMetadata: OffsetAndMetadata) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.seek(partition, offsetAndMetadata)
          d <- d
          _ <- log.debug(s"seek in ${ d.toMillis }ms, partition: $partition, offset: $offsetAndMetadata")
        } yield a
      }

      def seekToBeginning(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.seekToBeginning(partitions)
          d <- d
          _ <- log.debug(s"seekToBeginning in ${ d.toMillis }ms, partition: ${ partitions.mkString_(", ") }")
        } yield a
      }

      def seekToEnd(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.seekToEnd(partitions)
          d <- d
          _ <- log.debug(s"seekToEnd in ${ d.toMillis }ms, partition: ${ partitions.mkString_(", ") }")
        } yield a
      }

      def position(partition: TopicPartition) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.position(partition)
          d <- d
          _ <- log.debug(s"position in ${ d.toMillis }ms, partition: $partition, result: $a")
        } yield a
      }

      def position(partition: TopicPartition, timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.position(partition, timeout)
          d <- d
          _ <- log.debug(s"position in ${ d.toMillis }ms, partition: $partition, timeout: $timeout, result: $a")
        } yield a
      }

      def committed(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.committed(partitions)
          d <- d
          _ <- log.debug(s"committed in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, result: $a")
        } yield a
      }

      def committed(partitions: Nes[TopicPartition], timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.committed(partitions, timeout)
          d <- d
          _ <- log.debug(s"committed in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, timeout: $timeout, result: $a")
        } yield a
      }

      def partitions(topic: Topic) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.partitions(topic)
          d <- d
          _ <- log.debug(s"partitions in ${ d.toMillis }ms, topic: $topic, result: ${ a.mkString(", ") }")
        } yield a
      }

      def partitions(topic: Topic, timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.partitions(topic, timeout)
          d <- d
          _ <- log.debug(s"partitions in ${ d.toMillis }ms, topic: $topic, timeout: $timeout, result: ${ a.mkString(", ") }")
        } yield a
      }

      val topics = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.topics
          d <- d
          _ <- log.debug(s"topics in ${ d.toMillis }ms, result: ${ a.mkString(", ") }")
        } yield a
      }

      def topics(timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.topics(timeout)
          d <- d
          _ <- log.debug(s"topics in ${ d.toMillis }ms, timeout: $timeout, result: ${ a.mkString(", ") }")
        } yield a
      }

      def pause(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.pause(partitions)
          d <- d
          _ <- log.debug(s"pause in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }")
        } yield a
      }

      val paused = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.paused
          d <- d
          _ <- log.debug(s"paused in ${ d.toMillis }ms")
        } yield a
      }

      def resume(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.resume(partitions)
          d <- d
          _ <- log.debug(s"resume in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }")
        } yield a
      }

      def offsetsForTimes(timestampsToSearch: Map[TopicPartition, Offset]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.offsetsForTimes(timestampsToSearch)
          d <- d
          _ <- log.debug(s"offsetsForTimes in ${ d.toMillis }ms, partitions: ${ timestampsToSearch.mkString(", ") }, result: ${ a.mkString(", ") }")
        } yield a
      }

      def offsetsForTimes(timestampsToSearch: Map[TopicPartition, Offset], timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.offsetsForTimes(timestampsToSearch, timeout)
          d <- d
          _ <- log.debug(s"offsetsForTimes in ${ d.toMillis }ms, partitions: ${ timestampsToSearch.mkString(", ") }, timeout: $timeout, result: ${ a.mkString(", ") }")
        } yield a
      }

      def beginningOffsets(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.beginningOffsets(partitions)
          d <- d
          _ <- log.debug(s"beginningOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, result: ${ a.mkString(", ") }")
        } yield a
      }

      def beginningOffsets(partitions: Nes[TopicPartition], timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.beginningOffsets(partitions, timeout)
          d <- d
          _ <- log.debug(s"beginningOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, timeout: $timeout, result: ${ a.mkString(", ") }")
        } yield a
      }

      def endOffsets(partitions: Nes[TopicPartition]) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.endOffsets(partitions)
          d <- d
          _ <- log.debug(s"endOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, result: ${ a.mkString(", ") }")
        } yield a
      }

      def endOffsets(partitions: Nes[TopicPartition], timeout: FiniteDuration) = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.endOffsets(partitions, timeout)
          d <- d
          _ <- log.debug(s"endOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, timeout: $timeout, result: ${ a.mkString(", ") }")
        } yield a
      }

      def groupMetadata = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.groupMetadata
          d <- d
          _ <- log.debug(s"groupMetadata in ${ d.toMillis }ms, result: $a")
        } yield a
      }

      val wakeup = {
        for {
          d <- MeasureDuration[F].start
          a <- consumer.wakeup
          d <- d
          _ <- log.debug(s"wakeup in ${ d.toMillis }ms")
        } yield a
      }
    }
  }
}

1	package com.evolutiongaming.skafka.consumer
2
3	import java.util.regex.Pattern
4
5	import cats.Monad
6	import cats.data.{NonEmptySet => Nes, NonEmptyMap => Nem}
7	import cats.implicits._
8	import com.evolutiongaming.catshelper.Log
9	import com.evolutiongaming.skafka.{Offset, OffsetAndMetadata, Topic, TopicPartition}
10	import com.evolutiongaming.smetrics.MeasureDuration
11
12	import scala.concurrent.duration.FiniteDuration
13
14	object ConsumerLogging {
15
16	def apply[F[_] : Monad : MeasureDuration, K, V](
17	log: Log[F],
18	consumer: Consumer[F, K, V]
19	): Consumer[F, K, V] = {
20
21	new Consumer[F, K, V] {	1✔
22
23	def assign(partitions: Nes[TopicPartition]) = {
24	for {
25	d <- MeasureDuration[F].start	1✔
26	a <- consumer.assign(partitions)	1✔
27	d <- d	1✔
28	_ <- log.debug(s"assign in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }")	×
29	} yield a
30	}
31
32	val assignment = {
33	for {
34	d <- MeasureDuration[F].start	1✔
35	a <- consumer.assignment	1✔
36	d <- d	1✔
37	_ <- log.debug(s"assignment in ${ d.toMillis }ms, result: ${ a.toList.mkString_(", ") }")	1✔
38	} yield a
39	}
40
41	// TODO RebalanceListener1 implement logging - https://github.com/evolution-gaming/skafka/issues/127
42	def subscribe(topics: Nes[Topic], listener: RebalanceListener1[F]) = consumer.subscribe(topics, listener)	×
43
44	// TODO RebalanceListener1 implement logging - https://github.com/evolution-gaming/skafka/issues/127
45	def subscribe(pattern: Pattern, listener: RebalanceListener1[F]) = consumer.subscribe(pattern, listener)	×
46
47	def subscribe(topics: Nes[Topic], listener: Option[RebalanceListener[F]]) = {
48
49	val listenerLogging = (listener getOrElse RebalanceListener.empty[F]).withLogging(log)	1✔
50
51	for {
52	d <- MeasureDuration[F].start	1✔
53	a <- consumer.subscribe(topics, listenerLogging.some)	1✔
54	d <- d	1✔
55	_ <- log.debug(s"subscribe in ${ d.toMillis }ms, topics: ${ topics.mkString_(", ") }, listener: ${ listener.isDefined }")	×
56	} yield a
57	}
58
59	def subscribe(pattern: Pattern, listener: Option[RebalanceListener[F]]) = {
60
61	val listenerLogging = (listener getOrElse RebalanceListener.empty[F]).withLogging(log)	1✔
62
63	for {
64	d <- MeasureDuration[F].start	1✔
65	a <- consumer.subscribe(pattern, listenerLogging.some)	1✔
66	d <- d	1✔
67	_ <- log.debug(s"subscribe in ${ d.toMillis }ms, pattern: $pattern, listener: ${ listener.isDefined }")	×
68	} yield a
69	}
70
71	val subscription = {
72	for {
73	d <- MeasureDuration[F].start	1✔
74	a <- consumer.subscription	1✔
75	d <- d	1✔
76	_ <- log.debug(s"subscription in ${ d.toMillis }ms, result: ${ a.toList.mkString_(", ") }")	1✔
77	} yield a
78	}
79
80	val unsubscribe = {
81	for {
82	d <- MeasureDuration[F].start	1✔
83	a <- consumer.unsubscribe	1✔
84	d <- d	1✔
85	_ <- log.debug(s"unsubscribe in ${ d.toMillis }ms")	1✔
86	} yield a
87	}
88
89	def poll(timeout: FiniteDuration) = {
90
91	def show(records: ConsumerRecords[K, V]) = {
92	ConsumerRecords.summaryShow.show(records)	×
93	}
94
95	for {
96	d <- MeasureDuration[F].start	1✔
97	a <- consumer.poll(timeout)	1✔
98	d <- d	1✔
99	_ <- log.debug(s"poll timeout: $timeout in ${ d.toMillis }ms, result: ${ show(a) }")	×
100	} yield a
101	}
102
103	val commit = {
104	for {
105	d <- MeasureDuration[F].start	1✔
106	a <- consumer.commit	1✔
107	d <- d	1✔
108	_ <- log.debug(s"commit in ${ d.toMillis }ms")	×
109	} yield a
110	}
111
112	def commit(timeout: FiniteDuration) = {
113	for {
114	d <- MeasureDuration[F].start	1✔
115	a <- consumer.commit(timeout)	1✔
116	d <- d	1✔
117	_ <- log.debug(s"commit in ${ d.toMillis }ms, timeout: $timeout")	1✔
118	} yield a
119	}
120
121	def commit(offsets: Nem[TopicPartition, OffsetAndMetadata]) = {
122	for {
123	d <- MeasureDuration[F].start	1✔
124	a <- consumer.commit(offsets)	1✔
125	d <- d	1✔
126	_ <- log.debug(s"commit in ${ d.toMillis }ms, offsets: ${ offsets.mkString_(", ") }")	×
127	} yield a
128	}
129
130	def commit(offsets: Nem[TopicPartition, OffsetAndMetadata], timeout: FiniteDuration) = {
131	for {
132	d <- MeasureDuration[F].start	1✔
133	a <- consumer.commit(offsets, timeout)	1✔
134	d <- d	1✔
135	_ <- log.debug(s"commit in ${ d.toMillis }ms, offsets: ${ offsets.mkString_(", ") }, timeout: $timeout")	×
136	} yield a
137	}
138
139	val commitLater = {
140	for {
141	d <- MeasureDuration[F].start	1✔
142	a <- consumer.commitLater	1✔
143	d <- d	1✔
144	_ <- log.debug(s"commitLater in ${ d.toMillis }ms")	×
145	} yield a
146	}
147
148	def commitLater(offsets: Nem[TopicPartition, OffsetAndMetadata]) = {
149	for {
150	d <- MeasureDuration[F].start	1✔
151	a <- consumer.commitLater(offsets)	1✔
152	d <- d	1✔
153	_ <- log.debug(s"commitLater in ${ d.toMillis }ms, offsets: ${ offsets.mkString_(", ") }")	×
154	} yield a
155	}
156
157	def seek(partition: TopicPartition, offset: Offset) = {
158	for {
159	d <- MeasureDuration[F].start	1✔
160	a <- consumer.seek(partition, offset)	1✔
161	d <- d	1✔
162	_ <- log.debug(s"seek in ${ d.toMillis }ms, partition: $partition, offset: $offset")	1✔
163	} yield a
164	}
165
166	def seek(partition: TopicPartition, offsetAndMetadata: OffsetAndMetadata) = {
167	for {
168	d <- MeasureDuration[F].start	×
169	a <- consumer.seek(partition, offsetAndMetadata)	×
170	d <- d	×
171	_ <- log.debug(s"seek in ${ d.toMillis }ms, partition: $partition, offset: $offsetAndMetadata")	×
172	} yield a
173	}
174
175	def seekToBeginning(partitions: Nes[TopicPartition]) = {
176	for {
177	d <- MeasureDuration[F].start	1✔
178	a <- consumer.seekToBeginning(partitions)	1✔
179	d <- d	1✔
180	_ <- log.debug(s"seekToBeginning in ${ d.toMillis }ms, partition: ${ partitions.mkString_(", ") }")	×
181	} yield a
182	}
183
184	def seekToEnd(partitions: Nes[TopicPartition]) = {
185	for {
186	d <- MeasureDuration[F].start	1✔
187	a <- consumer.seekToEnd(partitions)	1✔
188	d <- d	1✔
189	_ <- log.debug(s"seekToEnd in ${ d.toMillis }ms, partition: ${ partitions.mkString_(", ") }")	×
190	} yield a
191	}
192
193	def position(partition: TopicPartition) = {
194	for {
195	d <- MeasureDuration[F].start	1✔
196	a <- consumer.position(partition)	1✔
197	d <- d	1✔
198	_ <- log.debug(s"position in ${ d.toMillis }ms, partition: $partition, result: $a")	1✔
199	} yield a
200	}
201
202	def position(partition: TopicPartition, timeout: FiniteDuration) = {
203	for {
204	d <- MeasureDuration[F].start	1✔
205	a <- consumer.position(partition, timeout)	1✔
206	d <- d	1✔
207	_ <- log.debug(s"position in ${ d.toMillis }ms, partition: $partition, timeout: $timeout, result: $a")	×
208	} yield a
209	}
210
211	def committed(partitions: Nes[TopicPartition]) = {
212	for {
213	d <- MeasureDuration[F].start	1✔
214	a <- consumer.committed(partitions)	1✔
215	d <- d	1✔
216	_ <- log.debug(s"committed in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, result: $a")	1✔
217	} yield a
218	}
219
220	def committed(partitions: Nes[TopicPartition], timeout: FiniteDuration) = {
221	for {
222	d <- MeasureDuration[F].start	1✔
223	a <- consumer.committed(partitions, timeout)	1✔
224	d <- d	1✔
225	_ <- log.debug(s"committed in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, timeout: $timeout, result: $a")	1✔
226	} yield a
227	}
228
229	def partitions(topic: Topic) = {
230	for {
231	d <- MeasureDuration[F].start	1✔
232	a <- consumer.partitions(topic)	1✔
233	d <- d	1✔
234	_ <- log.debug(s"partitions in ${ d.toMillis }ms, topic: $topic, result: ${ a.mkString(", ") }")	×
235	} yield a
236	}
237
238	def partitions(topic: Topic, timeout: FiniteDuration) = {
239	for {
240	d <- MeasureDuration[F].start	1✔
241	a <- consumer.partitions(topic, timeout)	1✔
242	d <- d	1✔
243	_ <- log.debug(s"partitions in ${ d.toMillis }ms, topic: $topic, timeout: $timeout, result: ${ a.mkString(", ") }")	1✔
244	} yield a
245	}
246
247	val topics = {
248	for {
249	d <- MeasureDuration[F].start	1✔
250	a <- consumer.topics	1✔
251	d <- d	1✔
252	_ <- log.debug(s"topics in ${ d.toMillis }ms, result: ${ a.mkString(", ") }")	×
253	} yield a
254	}
255
256	def topics(timeout: FiniteDuration) = {
257	for {
258	d <- MeasureDuration[F].start	1✔
259	a <- consumer.topics(timeout)	1✔
260	d <- d	1✔
261	_ <- log.debug(s"topics in ${ d.toMillis }ms, timeout: $timeout, result: ${ a.mkString(", ") }")	×
262	} yield a
263	}
264
265	def pause(partitions: Nes[TopicPartition]) = {
266	for {
267	d <- MeasureDuration[F].start	1✔
268	a <- consumer.pause(partitions)	1✔
269	d <- d	1✔
270	_ <- log.debug(s"pause in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }")	×
271	} yield a
272	}
273
274	val paused = {
275	for {
276	d <- MeasureDuration[F].start	1✔
277	a <- consumer.paused	1✔
278	d <- d	1✔
279	_ <- log.debug(s"paused in ${ d.toMillis }ms")	1✔
280	} yield a
281	}
282
283	def resume(partitions: Nes[TopicPartition]) = {
284	for {
285	d <- MeasureDuration[F].start	1✔
286	a <- consumer.resume(partitions)	1✔
287	d <- d	1✔
288	_ <- log.debug(s"resume in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }")	×
289	} yield a
290	}
291
292	def offsetsForTimes(timestampsToSearch: Map[TopicPartition, Offset]) = {
293	for {
294	d <- MeasureDuration[F].start	1✔
295	a <- consumer.offsetsForTimes(timestampsToSearch)	1✔
296	d <- d	1✔
297	_ <- log.debug(s"offsetsForTimes in ${ d.toMillis }ms, partitions: ${ timestampsToSearch.mkString(", ") }, result: ${ a.mkString(", ") }")	1✔
298	} yield a
299	}
300
301	def offsetsForTimes(timestampsToSearch: Map[TopicPartition, Offset], timeout: FiniteDuration) = {
302	for {
303	d <- MeasureDuration[F].start	1✔
304	a <- consumer.offsetsForTimes(timestampsToSearch, timeout)	1✔
305	d <- d	1✔
306	_ <- log.debug(s"offsetsForTimes in ${ d.toMillis }ms, partitions: ${ timestampsToSearch.mkString(", ") }, timeout: $timeout, result: ${ a.mkString(", ") }")	×
307	} yield a
308	}
309
310	def beginningOffsets(partitions: Nes[TopicPartition]) = {
311	for {
312	d <- MeasureDuration[F].start	1✔
313	a <- consumer.beginningOffsets(partitions)	1✔
314	d <- d	1✔
315	_ <- log.debug(s"beginningOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, result: ${ a.mkString(", ") }")	×
316	} yield a
317	}
318
319	def beginningOffsets(partitions: Nes[TopicPartition], timeout: FiniteDuration) = {
320	for {
321	d <- MeasureDuration[F].start	1✔
322	a <- consumer.beginningOffsets(partitions, timeout)	1✔
323	d <- d	1✔
324	_ <- log.debug(s"beginningOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, timeout: $timeout, result: ${ a.mkString(", ") }")	×
325	} yield a
326	}
327
328	def endOffsets(partitions: Nes[TopicPartition]) = {
329	for {
330	d <- MeasureDuration[F].start	1✔
331	a <- consumer.endOffsets(partitions)	1✔
332	d <- d	1✔
333	_ <- log.debug(s"endOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, result: ${ a.mkString(", ") }")	×
334	} yield a
335	}
336
337	def endOffsets(partitions: Nes[TopicPartition], timeout: FiniteDuration) = {
338	for {
339	d <- MeasureDuration[F].start	1✔
340	a <- consumer.endOffsets(partitions, timeout)	1✔
341	d <- d	1✔
342	_ <- log.debug(s"endOffsets in ${ d.toMillis }ms, partitions: ${ partitions.mkString_(", ") }, timeout: $timeout, result: ${ a.mkString(", ") }")	×
343	} yield a
344	}
345
346	def groupMetadata = {
347	for {
348	d <- MeasureDuration[F].start	1✔
349	a <- consumer.groupMetadata	1✔
350	d <- d	1✔
351	_ <- log.debug(s"groupMetadata in ${ d.toMillis }ms, result: $a")	×
352	} yield a
353	}
354
355	val wakeup = {
356	for {
357	d <- MeasureDuration[F].start	1✔
358	a <- consumer.wakeup	1✔
359	d <- d	1✔
360	_ <- log.debug(s"wakeup in ${ d.toMillis }ms")	1✔
361	} yield a
362	}
363	}
364	}
365	}

evolution-gaming / skafka / #563

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