Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
34 changes: 7 additions & 27 deletions core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.locks.ReentrantLock
import java.nio.ByteBuffer
import java.util.regex.Pattern

import org.junit.jupiter.api.Assertions._
import kafka.utils.CoreUtils.inLock
import org.apache.kafka.common.KafkaException
Expand All @@ -34,7 +33,7 @@ import org.slf4j.event.Level
import scala.jdk.CollectionConverters._
import scala.collection.mutable
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutorService, Future}

class CoreUtilsTest extends Logging {

Expand All @@ -60,9 +59,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 +84,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 +94,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 @@ -221,9 +201,9 @@ class CoreUtilsTest extends Logging {
val nThreads = 5
val createdCount = new AtomicInteger
val map = new ConcurrentHashMap[Int, AtomicInteger]().asScala
implicit val executionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))
implicit val executionContext: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(nThreads))
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