apache / pekko-connectors-kafka - Conditional Complexity

Intro

Conditional complexity (also called cyclomatic complexity) is a term used to measure the complexity of software. The term refers to the number of possible paths through a program function. A higher value ofter means higher maintenance and testing costs (infosecinstitute.com).
Conditional complexity is calculated by counting all conditions in the program that can affect the execution path (e.g. if statement, loops, switches, and/or operators, try and catch blocks...).
Conditional complexity is measured at the unit level (methods, functions...).
Units are classified in four categories based on the measured McCabe index: 1-5 (simple units), 6-10 (medium complex units), 11-25 (complex units), 26+ (very complex units).

Learn more...

Conditional Complexity Overall

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

0 very complex units (0 lines of code)
0 complex units (0 lines of code)
4 medium complex units (206 lines of code)
25 simple units (690 lines of code)
485 very simple units (5,101 lines of code)

Legend:

51+

26-50

11-25

6-10

1-5

Alternative Visuals

3D view of all units

Conditional Complexity per Extension

51+

26-50

11-25

6-10

1-5

Conditional Complexity per Logical Component

primary logical decomposition

51+

26-50

11-25

6-10

1-5

Most Complex Units

Top 50 most complex units

Unit	# lines	McCabe index	# params
def toCsv() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/CsvFormatter.scala	62	15	1
private def handleMetadataRequest() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	60	12	1
def handleSubscription() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	47	11	1
def poll() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	37	11	0
private def requiresQuotesOrSplit() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/CsvFormatter.scala	16	9	1
def pollForMetrics() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala	42	9	5
private def createTopicAndFill() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PerfFixtureHelpers.scala	36	9	4
private def maybeCommitTransaction() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalProducerStage.scala	15	9	2
override def apply() in project/CopyrightHeader.scala	14	8	2
private def updateBatchForPartition() in core/src/main/scala/org/apache/pekko/kafka/internal/CommitObservationLogic.scala	15	8	3
private def consumer[T]() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/InflightMetrics.scala	20	7	2
def consumerAtLeastOnceBatched() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaConsumerBenchmarks.scala	49	7	1
def updateBatch() in core/src/main/scala/org/apache/pekko/kafka/internal/CommitObservationLogic.scala	23	7	1
private def produce() in core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala	21	7	3
def waitUntilCluster() in testkit/src/main/scala/org/apache/pekko/kafka/testkit/internal/KafkaTestKitChecks.scala	41	7	5
public int compareTo() in testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/KafkaContainerCluster.java	13	7	1
def apply() in cluster-sharding/src/main/scala/org/apache/pekko/kafka/cluster/sharding/KafkaClusterSharding.scala	40	7	1
def consumeTransformProduceTransaction() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/KafkaTransactionBenchmarks.scala	58	6	1
def consumeTransformProduceTransaction() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/ReactiveKafkaTransactionBenchmarks.scala	32	6	2
private def checkForCompletion() in core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala	13	6	0
private def scheduleCommit() in core/src/main/scala/org/apache/pekko/kafka/internal/CommitCollectorStage.scala	16	6	0
protected def messageHandling: PartialFunction[() in core/src/main/scala/org/apache/pekko/kafka/internal/BaseSingleSourceLogic.scala	11	6	2
def stopping: Receive = LoggingReceive.withLabel() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	11	6	1
private def commit() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	42	6	3
private def checkForCompletion() in core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala	8	6	0
private def parseFirstMessage() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalProducerStage.scala	14	6	3
override def revoke() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	48	6	1
private def drainAndComplete() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	26	6	0
protected void containerIsStarting() in testkit/src/main/java/org/apache/pekko/kafka/testkit/internal/PekkoConnectorsKafkaContainer.java	66	6	2
override def apply() in project/CopyrightHeader.scala	11	5	2
def apply[K, V]() in core/src/main/scala/org/apache/pekko/kafka/ConsumerSettings.scala	58	5	3
def apply[K, V]() in core/src/main/scala/org/apache/pekko/kafka/ProducerSettings.scala	40	5	3
def withMetadata() in core/src/main/scala/org/apache/pekko/kafka/ConsumerMessage.scala	9	5	1
def offsets: Map[GroupTopicPartition, Long] = offsetsAndMetadata.view.mapValues() in core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala	7	5	1
private[internal] def committerFor() in core/src/main/scala/org/apache/pekko/kafka/internal/MessageBuilder.scala	25	5	1
private def collectOffset() in core/src/main/scala/org/apache/pekko/kafka/internal/CommittingProducerSinkStage.scala	17	5	1
protected def closeProducer() in core/src/main/scala/org/apache/pekko/kafka/internal/DeferredProducer.scala	10	5	0
private def pump() in core/src/main/scala/org/apache/pekko/kafka/internal/BaseSingleSourceLogic.scala	10	5	0
def expectSettings: Receive = LoggingReceive.withLabel() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	13	5	1
def checkOverlappingRequests() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	9	5	3
private def applySettings() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	22	5	2
private def processResult() in core/src/main/scala/org/apache/pekko/kafka/internal/KafkaConsumerActor.scala	23	5	3
override def preStart() in core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala	20	5	0
override def performShutdown() in core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala	23	5	0
protected def messageHandling: PartialFunction[() in core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala	10	5	2
protected def initialInHandler() in core/src/main/scala/org/apache/pekko/kafka/internal/DefaultProducerStage.scala	34	5	0
private def drainHandling: PartialFunction[() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	22	5	2
override protected def logic() in core/src/main/scala/org/apache/pekko/kafka/internal/TransactionalSources.scala	56	5	4
def backoff() in core/src/main/scala/org/apache/pekko/kafka/internal/ConnectionChecker.scala	15	5	2
def run() in benchmarks/src/main/scala/org/apache/pekko/kafka/benchmarks/PekkoConnectorsCommittableSinkFixtures.scala	28	4	2