Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 0 additions & 39 deletions core/src/main/scala/kafka/utils/CoreUtils.scala
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ package kafka.utils

import java.io._
import java.nio._
import java.nio.channels._
import java.util.concurrent.locks.{Lock, ReadWriteLock}
import java.lang.management._
import java.util.{Base64, Properties, UUID}
Expand Down Expand Up @@ -50,12 +49,6 @@ import scala.annotation.nowarn
object CoreUtils {
private val logger = Logger(getClass)

/**
* Return the smallest element in `iterable` if it is not empty. Otherwise return `ifEmpty`.
*/
def min[A, B >: A](iterable: Iterable[A], ifEmpty: A)(implicit cmp: Ordering[B]): A =
if (iterable.isEmpty) ifEmpty else iterable.min(cmp)

/**
* Do the given action and log any exceptions thrown without rethrowing them.
*
Expand Down Expand Up @@ -134,30 +127,6 @@ object CoreUtils {
}
}

/**
* Unregister the mbean with the given name, if there is one registered
* @param name The mbean name to unregister
*/
def unregisterMBean(name: String): Unit = {
val mbs = ManagementFactory.getPlatformMBeanServer()
mbs synchronized {
val objName = new ObjectName(name)
if (mbs.isRegistered(objName))
mbs.unregisterMBean(objName)
}
}

/**
* Read some bytes into the provided buffer, and return the number of bytes read. If the
* channel has been closed or we get -1 on the read for any reason, throw an EOFException
*/
def read(channel: ReadableByteChannel, buffer: ByteBuffer): Int = {
channel.read(buffer) match {
case -1 => throw new EOFException("Received -1 when reading from channel, socket has likely been closed.")
case n => n
}
}

/**
* This method gets comma separated values which contains key,value pairs and returns a map of
* key value pairs. the format of allCSVal is key1:val1, key2:val2 ....
Expand Down Expand Up @@ -195,14 +164,6 @@ object CoreUtils {
constructor.newInstance(args: _*)
}

/**
* Create a circular (looping) iterator over a collection.
* @param coll An iterable over the underlying collection.
* @return A circular iterator over the collection.
*/
def circularIterator[T](coll: Iterable[T]) =
for (_ <- Iterator.continually(1); t <- coll) yield t

/**
* Execute the given function inside the lock
*/
Expand Down
29 changes: 5 additions & 24 deletions core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
Original file line number Diff line number Diff line change
Expand Up @@ -60,9 +60,9 @@ class CoreUtilsTest extends Logging {

CoreUtils.tryAll(Seq(
() => recordingFunction(Right("valid-0")),
() => recordingFunction(Left(new TestException("exception-1"))),
() => recordingFunction(Left(TestException("exception-1"))),
() => recordingFunction(Right("valid-2")),
() => recordingFunction(Left(new TestException("exception-3")))
() => recordingFunction(Left(TestException("exception-3")))
))
var expected = Map(
"valid-0" -> Right("valid-0"),
Expand All @@ -85,8 +85,8 @@ class CoreUtilsTest extends Logging {

recorded.clear()
CoreUtils.tryAll(Seq(
() => recordingFunction(Left(new TestException("exception-0"))),
() => recordingFunction(Left(new TestException("exception-1")))
() => recordingFunction(Left(TestException("exception-0"))),
() => recordingFunction(Left(TestException("exception-1")))
))
expected = Map(
"exception-0" -> Left(TestException("exception-0")),
Expand All @@ -95,25 +95,6 @@ class CoreUtilsTest extends Logging {
assertEquals(expected, recorded)
}

@Test
def testCircularIterator(): Unit = {
val l = List(1, 2)
val itl = CoreUtils.circularIterator(l)
assertEquals(1, itl.next())
assertEquals(2, itl.next())
assertEquals(1, itl.next())
assertEquals(2, itl.next())
assertFalse(itl.isEmpty)

val s = Set(1, 2)
val its = CoreUtils.circularIterator(s)
assertEquals(1, its.next())
assertEquals(2, its.next())
assertEquals(1, its.next())
assertEquals(2, its.next())
assertEquals(1, its.next())
}

@Test
def testReadBytes(): Unit = {
for(testCase <- List("", "a", "abcd")) {
Expand Down Expand Up @@ -223,7 +204,7 @@ class CoreUtilsTest extends Logging {
val map = new ConcurrentHashMap[Int, AtomicInteger]().asScala
implicit val executionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As you are cleaning up some minor problems, feel free to type this implicit val as this is the recommended way:

Suggested change
implicit val executionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))
implicit val executionContext: ExecutionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've add to use ExecutionContextExecutorService instead of ExecutionContext because we call shutdownNow() below.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks good!

try {
Await.result(Future.traverse(1 to count) { i =>
Await.result(Future.traverse(1 to count) { _ =>
Future {
CoreUtils.atomicGetOrUpdate(map, 0, {
createdCount.incrementAndGet
Expand Down