apache / pekko-connectors-kafka

Intro

Unit size measurements show the distribution of size of units of code (methods, functions...).
Units are classified in four categories based on their size (lines of code): 1-20 (small units), 20-50 (medium size units), 51-100 (long units), 101+ (very long units).
You should aim at keeping units small (< 20 lines). Long units may become "bloaters", code that have increased to such gargantuan proportions that they are hard to work with.

Learn more...

Unit Size Overall

There are 514 units with 5,997 lines of code in units (60.5% of code).

2 very long units (251 lines of code)
12 long units (764 lines of code)
50 medium size units (1,525 lines of code)
113 small units (1,576 lines of code)
337 very small units (1,881 lines of code)

Legend:

101+

51-100

21-50

11-20

1-10

Unit Size per Extension

101+

51-100

21-50

11-20

1-10

Unit Size per Logical Component

primary logical decomposition

101+

51-100

21-50

11-20

1-10

Alternative Visuals

3D view of all units

Longest Units

Top 50 longest units

Unit	# lines	McCabe index	# params
def withBootstrapServers() in core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala	143	3	1
def getZooKeeperImage() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/KafkaTestkitTestcontainersSettings.scala	108	1	0
def producerFactory: ProducerSettings[K, V] => Producer[K, V] = _ => createKafkaProducer() in core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala	82	1	0
def apply[K, V]() in core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala	72	1	3
override def stop() in core/src/main/scala/org/apache/pekko/kafka/javadsl/Consumer.scala	71	1	0
protected void containerIsStarting() in testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/PekkoConnectorsKafkaContainer.java	66	6	2
def toCsv() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/CsvFormatter.scala	62	15	1
public KafkaContainerCluster() in testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java	62	4	9
def apply[K, V]() in core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala	61	1	3
private def handleMetadataRequest() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	60	12	1
def consumeTransformProduceTransaction() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.scala	58	6	1
def apply[K, V]() in core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala	58	5	3
override protected def logic() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	56	5	4
def messageExtractor[M]() in cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala	56	1	4
def consumerAtLeastOnceBatched() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala	49	7	1
override def revoke() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	48	6	1
def handleSubscription() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	47	11	1
def pollForMetrics() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala	42	9	5
private def commit() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	42	6	3
def create() in core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala	42	1	6
def waitUntilCluster() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala	41	7	5
def apply[K, V]() in core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala	40	5	3
private String readinessCheckScript() in testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java	40	1	0
def apply() in cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala	40	7	1
override def stop() in core/src/main/scala/org/apache/pekko/kafka/scaladsl/Consumer.scala	37	1	0
def plainSink[K, V]() in core/src/main/scala/org/apache/pekko/kafka/scaladsl/Producer.scala	37	2	2
def poll() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	37	11	0
private def createTopicAndFill() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PerfFixtureHelpers.scala	36	9	4
def consumerAtLeastOnceCommitEveryPoll() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala	35	4	0
protected def initialInHandler() in core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala	34	5	0
def consumeTransformProduceTransaction() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionBenchmarks.scala	32	6	2
def createKafkaConsumerAsync() in core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala	32	1	0
def source[K, V]() in core/src/main/scala/org/apache/pekko/kafka/scaladsl/Transactional.scala	30	2	3
def startCluster() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/TestcontainersKafka.scala	30	4	0
def consumeCommitAtMostOnce() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala	29	3	2
private def protectPartition[K, V]() in core/src/main/scala/org/apache/pekko/kafka/internal/ConsumerResetProtection.scala	29	4	5
def apply() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/KafkaTestkitTestcontainersSettings.scala	29	1	1
def run() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala	28	4	2
def plainFlow() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaProducerBenchmarks.scala	28	4	2
def initialize() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionFixtureGen.scala	27	1	1
private def createConsumerSettings() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionFixtures.scala	27	1	1
def attachControl[A, B]() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/javadsl/ConsumerControlFactory.scala	27	1	2
private def createConsumerSettings() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala	26	1	1
override protected def logic() in core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala	26	1	4
protected def configureSubscription() in core/src/main/scala/org/apache/pekko/kafka/internal/SourceLogicSubscription.scala	26	4	2
override def onMessage() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	26	2	2
private def drainAndComplete() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	26	6	0
def producerDefaults: ProducerSettings[String, String] = producerDefaults() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKit.scala	26	1	2
def consumePlainInflightMetrics() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaConsumerBenchmarks.scala	25	1	5
def reporter() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/Timed.scala	25	1	1