Apache 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 17,006 units with 150,195 lines of code in units (70.7% of code).
    • 4 very complex units (474 lines of code)
    • 30 complex units (3,452 lines of code)
    • 267 medium complex units (14,614 lines of code)
    • 890 simple units (25,176 lines of code)
    • 15,815 very simple units (106,479 lines of code)
<1% | 2% | 9% | 16% | 70%
Legend:
51+
26-50
11-25
6-10
1-5
Alternative Visuals
Conditional Complexity per Extension
java<1% | 1% | 7% | 14% | 75%
scala<1% | 4% | 17% | 22% | 54%
Legend:
51+
26-50
11-25
6-10
1-5
Conditional Complexity per Logical Component
primary logical decomposition
clients<1% | 1% | 7% | 13% | 76%
connect<1% | 3% | 8% | 14% | 73%
core<1% | 4% | 17% | 23% | 54%
tools0% | 4% | 4% | 21% | 68%
generator0% | 5% | 29% | 21% | 43%
streams0% | 0% | 3% | 14% | 81%
log4j-appender0% | 0% | 24% | 0% | 75%
examples0% | 0% | 0% | 37% | 62%
jmh-benchmarks0% | 0% | 0% | 4% | 95%
Legend:
51+
26-50
11-25
6-10
1-5
Most Complex Units
Top 50 most complex units
Unit# linesMcCabe index# params
protected static Object convertTo()
in connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
185 58 3
def handle()
in core/src/main/scala/kafka/server/KafkaApis.scala
67 57 1
public static AbstractResponse parseResponse()
in clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
111 53 3
public static AbstractRequest parseRequest()
in clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
111 53 3
protected static SchemaAndValue parse()
in connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
166 49 2
private def endTransaction()
in core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
146 46 6
protected static Schema commonSchemaFor()
in connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
67 43 2
String fieldDefault()
in generator/src/main/java/org/apache/kafka/message/FieldSpec.java
147 40 2
public Schema asConnectSchema()
in connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
101 40 1
public void onCompletion()
in connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java
172 39 2
private JsonNode convertToJson()
in connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
108 39 2
void checkSchemaMatches()
in connect/api/src/main/java/org/apache/kafka/connect/header/ConnectHeaders.java
75 39 1
def checkArgs()
in core/src/main/scala/kafka/admin/ConfigCommand.scala
58 38 0
def handleFetchRequest()
in core/src/main/scala/kafka/server/KafkaApis.scala
196 37 1
private def doLoadGroupsAndOffsets()
in core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
135 36 2
public static Object parseType()
in clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
91 35 3
def main()
in core/src/main/scala/kafka/tools/JmxTool.scala
186 34 1
def handleDescribeClientQuotas()
in core/src/main/scala/kafka/server/AdminManager.scala
74 34 3
void pollSelectionKeys()
in clients/src/main/java/org/apache/kafka/common/network/Selector.java
98 33 3
private String fieldTypeToSchemaType()
in generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java
92 31 5
private[admin] def alterConfig()
in core/src/main/scala/kafka/admin/ConfigCommand.scala
84 30 2
override def process()
in core/src/main/scala/kafka/controller/KafkaController.scala
66 30 1
public int read()
in clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
79 29 1
public static void main()
in tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java
242 29 1
private CompletedFetch initializeCompletedFetch()
in clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
108 28 1
public static void main()
in tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java
105 28 1
def storeOffsets()
in core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
112 28 8
private def processFetchRequest()
in core/src/main/scala/kafka/server/AbstractFetcherThread.scala
87 27 3
def handleOffsetCommitRequest()
in core/src/main/scala/kafka/server/KafkaApis.scala
108 27 1
private def doHandleStateChanges()
in core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
123 27 3
public LoginContext login()
in clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java
144 26 0
override def onComplete()
in core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerRequestCompletionHandler.scala
126 26 1
private def prepareOffsetsToReset()
in core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
91 26 2
def shutdown()
in core/src/main/scala/kafka/server/KafkaServer.scala
65 26 0
public void activateOptions()
in log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java
55 25 0
public void authenticate()
in clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
72 24 0
private void generateVariableLengthFieldSize()
in generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
165 24 3
public void handle()
in clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
82 23 2
def appendTransactionToLog()
in core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
119 23 5
def checkArgs()
in core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
56 23 0
private def controlledShutdown()
in core/src/main/scala/kafka/server/KafkaServer.scala
120 23 0
def validateMessagesAndAssignOffsetsCompressed()
in core/src/main/scala/kafka/log/LogValidator.scala
98 23 15
private void doHandshake()
in clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
82 22 0
def completeTransitionTo()
in core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala
60 22 1
public void tick()
in connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
91 22 0
public ObjectNode asJsonSchema()
in connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
75 22 1
public AlterPartitionReassignmentsResult alterPartitionReassignments()
in clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
130 21 2
public void handle()
in clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
76 21 2
def checkArgs()
in core/src/main/scala/kafka/admin/AclCommand.scala
24 21 0
def has()
in core/src/main/scala/kafka/admin/TopicCommand.scala
73 21 1