# Belay the Metamorphosis: analyzing Kafka project

Have you ever wondered what bugs might be lurking in the project source code of global companies? Don't miss the chance to spot interesting bugs detected by the PVS\-Studio static analyzer in the open\-source Apache Kafka project\.

![1171_kafka/image1.png](https://import.viva64.com/docx/blog/1171_kafka/image1.png)

## Introduction

Apache Kafka is a well\-known [open\-source](https://github.com/apache/kafka) project that's mostly written in Java\. LinkedIn developed it in 2011 as a message broker, i\.e\. a data pipeline for various system components\. Today, it's one of the most popular solutions in its category\.

Ready to take a look under the hood?

<details>
   <summary>P\\\.S\\\.</summary>

Just wanted to leave a quick note about the title\. It references Franz Kafka's "The Metamorphosis", where the main character turns into monstrous vermin\. Our static analyzer fights to keep your projects from ~~transfiguring into monstrous vermin~~ transforming into one colossal bug, so say no to "The Metamorphosis"\.


</details>
 

## Oh no, bugs

### All humor is rooted with pain

These aren't my words; the quote belongs to Richard Pryor\. But what does that matter? The first thing I'd like to tell you about is a silly error\. Yet, after many attempts to understand why the program doesn't work properly, it's frustrating to encounter something like the example below:

```cpp
@Override
public KeyValueIterator<Windowed<K>, V> backwardFetch(
  K keyFrom,
  K keyTo,
  Instant timeFrom,
  Instant timeTo) {
  ....
  if (keyFrom == null && keyFrom == null) {   // <=
    kvSubMap = kvMap;
  } else if (keyFrom == null) {
    kvSubMap = kvMap.headMap(keyTo, true);
  } else if (keyTo == null) {
    kvSubMap = kvMap.tailMap(keyFrom, true);
  } else {
    // keyFrom != null and KeyTo != null 
    kvSubMap = kvMap.subMap(keyFrom, true, keyTo, true);
  } 
  ....
}
```

As you can see, here's something that no developer can avoid—a trivial typo\. In the very first condition, developers wanted to use the following logical expression:

```cpp
keyFrom == null && keyTo == null
```

The analyzer issued two warnings:

[V6001](https://pvs-studio.com/en/docs/warnings/v6001/) There are identical sub\-expressions 'keyFrom \=\= null' to the left and to the right of the '&&' operator\. ReadOnlyWindowStoreStub\.java 327, ReadOnlyWindowStoreStub\.java 327

[V6007](https://pvs-studio.com/en/docs/warnings/v6007/) Expression 'keyFrom \=\= null' is always false\. ReadOnlyWindowStoreStub\.java 329

We can see why\. Such chucklesome typos are timeless for every developer\. While we can spend a lot of time searching for them, and yet it won't be a piece of cake to recall where it has lurked\.

In the same class, there is exactly the same error in another method\. I think it's fair to call this copypasta\.

```cpp
@Override
public KeyValueIterator<Windowed<K>, V> fetch(
  K keyFrom,
  K keyTo,
  Instant timeFrom,
  Instant timeTo) {
  ....
  NavigableMap<K, V> kvMap = data.get(now);
  if (kvMap != null) {
    NavigableMap<K, V> kvSubMap;
    if (keyFrom == null && keyFrom == null) {      // <=
      kvSubMap = kvMap;
    } else if (keyFrom == null) {
      kvSubMap = kvMap.headMap(keyTo, true);
    } else if (keyTo == null) {
      kvSubMap = kvMap.tailMap(keyFrom, true);
    } else {
      // keyFrom != null and KeyTo != null
      kvSubMap = kvMap.subMap(keyFrom, true, keyTo, true);
    }
  }
  ....
}
```

Here are the same warnings:

[V6007](https://pvs-studio.com/en/docs/warnings/v6007/) Expression 'keyFrom \=\= null' is always false\. ReadOnlyWindowStoreStub\.java 273  

[V6001](https://pvs-studio.com/en/docs/warnings/v6001/) There are identical sub\-expressions 'keyFrom \=\= null' to the left and to the right of the '&&' operator\. ReadOnlyWindowStoreStub\.java 271, ReadOnlyWindowStoreStub\.java 271

No need to worry—we won't have to look at hundreds of code lines at once\. PVS\-Studio is great at handling such simple things\. How about tackling something a little more challenging?

### Mutable synchronized

What's the purpose of the _synchronized_ keyword in Java? Here, I'll only focus on the synchronized methods, not blocks\. According to the [Oracle docs](https://docs.oracle.com/javase/specs/jls/se8/html/jls-17.html#jls-17.1), the _synchronized_ keyword declares a method as synchronized to ensure a thread\-safe interaction with an instance\. If a thread invokes a synchronized method of the instance, other threads that try to invoke synchronized methods of the same instance will be blocked \(i\.e\. their execution will be suspended\)\. They'll be blocked until the method invoked by the first thread processes its execution\. This is needed when the instance is visible to more than one thread\. The read/write operations of such instances should be executed only via synchronized methods\. 

The developers broke the rule in the _Sensor_ class, as illustrated in the simplified code fragment below\. The read/write operations on the instance field are executed through both synchronized and unsynchronized methods\. It may lead to a race condition and make the output unpredictable\.

```cpp
private final Map<MetricName, KafkaMetric> metrics;

public void checkQuotas(long timeMs) {                  // <=
  for (KafkaMetric metric : this.metrics.values()) {
    MetricConfig config = metric.config();
    if (config != null) {
      ....
    }
  }
  ....
}  

public synchronized boolean add(CompoundStat stat,      // <=
                                MetricConfig config) {       
  ....
  if (!metrics.containsKey(metric.metricName())) {         
    metrics.put(metric.metricName(), metric);            
  }  
  ....
}  

public synchronized boolean add(MetricName metricName,  // <=
                                MeasurableStat stat, 
                                MetricConfig config) {  
  if (hasExpired()) {
    return false;
  } else if (metrics.containsKey(metricName)) {
    return true;
  } else {
    ....
    metrics.put(metric.metricName(), metric);
    return true;
  }
}
```

The analyzer warning looks like this:

[V6102](https://pvs-studio.com/en/docs/warnings/v6102/) Inconsistent synchronization of the 'metrics' field\. Consider synchronizing the field on all usages\. Sensor\.java 49, Sensor\.java 254

If different threads can change the instance state at once, the methods that allow this should be synchronized\. If the program doesn't anticipate that several threads can interact with the instance, it's pointless to make its methods synchronized\. In the worst case, it can even damage the program performance\.

There are plenty of such errors in the program\. Here's a similar code fragment for which the analyzer issued the warning:

```cpp
private final PrefixKeyFormatter prefixKeyFormatter; 

@Override
public synchronized void destroy() {                // <=
  ....
  Bytes keyPrefix = prefixKeyFormatter.getPrefix();
  ....
}

@Override
public void addToBatch(....) {                      // <=
  physicalStore.addToBatch(
    new KeyValue<>(
    prefixKeyFormatter.addPrefix(record.key),
    record.value
    ), batch
  );
} 

@Override
public synchronized void deleteRange(....) {        // <=
  physicalStore.deleteRange(
    prefixKeyFormatter.addPrefix(keyFrom),
    prefixKeyFormatter.addPrefix(keyTo)
  );
}

@Override
public synchronized void put(....) {                // <=
  physicalStore.put(
    prefixKeyFormatter.addPrefix(key),
    value
  );
}
```

The analyzer warning:

[V6102](https://pvs-studio.com/en/docs/warnings/v6102/) Inconsistent synchronization of the 'prefixKeyFormatter' field\. Consider synchronizing the field on all usages\. LogicalKeyValueSegment\.java 60, LogicalKeyValueSegment\.java 247

### Iterator, iterator, and iterator again\.\.\.

In the example, there are two rather unpleasant errors within one line at once\. I'll explain their nature within the part of the article\. Here's a code snippet:

```cpp
private final Map<String, Uuid> topicIds = new HashMap(); 

private Map<String, KafkaFutureVoid> handleDeleteTopicsUsingNames(....) { 
  ....
  Collection<String> topicNames = new ArrayList<>(topicNameCollection);

  for (final String topicName : topicNames) {
    KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();

    if (allTopics.remove(topicName) == null) {
      ....
    } else {
      topicNames.remove(topicIds.remove(topicName));      // <=
      future.complete(null);
    }
    ....
  }
}
```

That's what the analyzer shows us:

[V6066](https://pvs-studio.com/en/docs/warnings/v6066/) The type of object passed as argument is incompatible with the type of collection: String, Uuid\. MockAdminClient\.java 569

[V6053](https://pvs-studio.com/en/docs/warnings/v6053/) The 'topicNames' collection of 'ArrayList' type is modified while iteration is in progress\. ConcurrentModificationException may occur\. MockAdminClient\.java 569

Now that's a big dilemma\! What's going on here, and how should we address it?\! 

First, let's talk about collections and generics\. Using the generic types of collections helps us avoid _ClassCastExceptions_ and cumbersome constructs where we convert types\. 

If we specify a certain data type when initializing a collection and add an incompatible type, the compiler won't compile the code\. 

Here's an example:

```cpp
public class Test {
  public static void main(String[] args) {
    Set<String> set = new HashSet<>();
    set.add("str");
    set.add(UUID.randomUUID()); // java.util.UUID cannot be converted to
                                // java.lang.String
  }
}
```

However, if we delete an incompatible type from our _Set_, no exception will be thrown\. The method returns _false_\. 

Here's an example:

```cpp
public class Test {
  public static void main(String[] args) {
    Set<String> set = new HashSet<>();
    set.add("abc");
    set.add("def");
    System.out.println(set.remove(new Integer(13))); // false
  }
}
```

It's a waste of time\. Most likely, if we encounter something like this in the code, this is an error\. I suggest you go back to the code at the beginning of this subchapter and try to spot a similar case\.

Second, let's talk about the _Iterator_\. We can talk about iterating through collections for a long time\. I don't want to bore you or digress from the main topic, so I'll just cover the key points to ensure we understand why we get the warning\.

So, how do we iterate through the collection here? Here is what the _for_ loop in the code fragment looks like:

```cpp
for (Type collectionElem : collection) {
  ....
}
```

The _for_ loop entry is just syntactic sugar\. The construction is [equivalent](https://docs.oracle.com/javase/specs/jls/se7/html/jls-14.html#jls-14.14.2) to this one:

```cpp
for (Iterator<Type> iter = collection.iterator(); iter.hasNext();) {
  Type collectionElem = iter.next();
  ....
}
```

We're basically working with the collection iterator\. All right, that's sorted\! Now, let's discuss _ConcurrentModificationException_\.

_ConcurrentModificationException_ is an exception that covers a range of situations both in single\-threaded and multi\-threaded programs\. Here, we're focusing on single\-threading\. We can find an explanation quite easily\. Let's take a peek at the [Oracle docs](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/ConcurrentModificationException.html): a method can throw the exception when it detects parallel modification of an object that doesn't support it\. In our case, while the iterator is running, we delete objects from the collection\. This may cause the iterator to throw a _ConcurrentModificationException\._

How does the iterator know when to throw the exception? If we look at the _ArrayList_ collection, we see that its parent, _AbstactList_, has the _modCount_ field that stores the number of modifications to the collection:

```cpp
protected transient int modCount = 0;
```

Here are some usages of the _modCount_ counter in the _ArrayList_ class:

```cpp
public boolean add(E e) {
  modCount++;
  add(e, elementData, size);
  return true;
}

private void fastRemove(Object[] es, int i) {
  modCount++;
  final int newSize;
  if ((newSize = size - 1) > i)
    System.arraycopy(es, i + 1, es, i, newSize - i);
  es[size = newSize] = null;
}
```

So, the counter is incremented each time when the collection is modified\.

Btw, the _fastRemove_ method is used in the _remove_ method, which we use inside the loop\.

Here's the small code fragment of the _ArrayList_ iterator inner workings:

```cpp
private class Itr implements Iterator<E> {
  ....
  int expectedModCount = modCount;            

  final void checkForComodification() {
  if (modCount != expectedModCount)               // <=
    throw new ConcurrentModificationException();
  }

  public E next() {
    checkForComodification();              
    ....
  }
    
  public void remove() {
    ....
    checkForComodification();             

    try {
      ArrayList.this.remove(lastRet);   
      ....
      expectedModCount = modCount;
    } catch (IndexOutOfBoundsException ex) {
      throw new ConcurrentModificationException();
    }
  }
  ....
  public void add(E e) {
    checkForComodification();            
    try {
      ....
      ArrayList.this.add(i, e);        
      ....
      expectedModCount = modCount;     
    } catch (IndexOutOfBoundsException ex) {
      throw new ConcurrentModificationException();
    }
  }
}
```

Let me explain that last fragment\. If the collection modifications don't match the expected number of modifications \(which is the sum of the initial modifications before the iterator was created and the number of the iterator operations\), a _ConcurrentModificationException_ is thrown\. That's only possible when we modify the collection using its methods while iterating over it \(i\.e\. **in parallel** with the iterator\)\. That's what the second warning is about\.

So, I've explained you the analyzer messages\. Now let's put it all together: 

We attempt to delete an element from the collection when the _Iterator_ is still running: 

```cpp
topicNames.remove(topicIds.remove(topicName)); 
// topicsNames – Collection<String>
// topicsIds – Map<String, UUID>
```

However, since the incompatible element is passed to _ArrayList_ for deletion \(_the remove method returns a UUID object from topicIds_\), the modification count won't increase, but the object won't be deleted\. Simply put, that code section is rudimentary\.

I'd venture to guess that the developer's intent is clear\. If that's the case, one way to fix these two warnings could be as follows:

```cpp
Collection<String> topicNames = new ArrayList<>(topicNameCollection);

List<String> removableItems = new ArrayList<>();

for (final String topicName : topicNames) {
  KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();

  if (allTopics.remove(topicName) == null) {
    ....
  } else {
    topicIds.remove(topicName);
    removableItems.add(topicName);
    future.complete(null);
  }
  ....
}
topicNames.removeAll(removableItems);
```

### Void, sweet void

Where would we go without our all\-time favorite _null_ and its potential problems, right? Let me show you the code fragment for which the analyzer issued the following warning:

[V6008](https://pvs-studio.com/en/docs/warnings/v6008/) Potential null dereference of 'oldMember' in function 'removeStaticMember'\. ConsumerGroup\.java 311, ConsumerGroup\.java 323

```cpp
@Override
public void removeMember(String memberId) {
  ConsumerGroupMember oldMember = members.remove(memberId);
  ....
  removeStaticMember(oldMember);
  ....
}

private void removeStaticMember(ConsumerGroupMember oldMember) {
  if (oldMember.instanceId() != null) {
    staticMembers.remove(oldMember.instanceId());
  }
}
```

If _members_ doesn't contain an object with the _memberId_ key, _oldMember_ will be _null_\. It can lead to a _NullPointerException_ in the _removeStaticMember_ method\.

Boom\! The parameter is checked for _null_:

```cpp
if (oldMember != null && oldMember.instanceId() != null) {
```

The next error will be the last one in the article—I'd like to wrap things up on a positive note\. The code below—as well as the one at the beginning of this article—has a common and silly typo\. However, it can certainly lead to unpleasant consequences\. 

Let's take a look at this code fragment:

```cpp
protected SchemaAndValue roundTrip(...., SchemaAndValue input) {
  String serialized = Values.convertToString(input.schema(),
                                             input.value());

  if (input != null && input.value() != null) {   
    ....
  }
  ....
}
```

Yeah, that's right\. The method actually accesses the _input_ object first, and then checks whether it's referencing _null_\.

[V6060](https://pvs-studio.com/en/docs/warnings/v6060/) The 'input' reference was utilized before it was verified against null\. ValuesTest\.java 1212, ValuesTest\.java 1213

Again, I'll note that such typos are ok\. However, they can lead to some pretty nasty results\. It's tough and inefficient to search for these things in the code manually\. 

## Conclusion

In sum, I'd like to circle back to the previous point\. Manually searching through the code for all these errors is a very time\-consuming and tedious task\. It's not unusual for issues like the ones I've shown to lurk in code for a long time\. The last bug dates back to 2018\. That's why it's a good idea to use static analysis tools\. If you'd like to know more about PVS\-Studio, the tool we have used to detect all those errors, you can find out more [here](https://pvs-studio.com/en/pvs-studio/)\.

That's all\. Let's wrap things up here\. "Oh, and in case I don't see ya, good afternoon, good evening, and good night\."

![1171_kafka/image2.gif](https://import.viva64.com/docx/blog/1171_kafka/image2.gif)

I almost forgot\! Catch a [link](https://pvs-studio.com/en/order/open-source-license/) to learn more about a free license for open\-source projects\.