Belay the Metamorphosis: Kafka プロジェクトの分析

您有沒有想過跨國公司的專案原始碼中可能潛藏著哪些錯誤？不要錯過在開源 Apache Kafka 專案中發現 PVS-Studio 靜態分析器偵測到的有趣錯誤的機會。

介紹

Apache Kafka 是一個著名的開源項目，主要用 Java 寫。 LinkedIn 於 2011 年將其開發為訊息代理，即各種系統元件的資料管道。如今，它已成為同類產品中最受歡迎的解決方案之一。

準備好看看引擎蓋下的內容了嗎？

喔不，蟲子

所有的幽默都源自於痛苦

這不是我的話；這句話出自理查德·普賴爾之口。但這有什麼關係呢？我想告訴你的第一件事是一個愚蠢的錯誤。然而，在多次嘗試理解程式無法正常運作的原因後，遇到以下範例的情況令人沮喪：

@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);
  } 
  ....
}

如您所見，這是任何開發人員都無法避免的事情——一個微不足道的拼字錯誤。在第一個條件下，開發人員希望使用下列邏輯表達式：

keyFrom == null && keyTo == null

分析器發出兩個警告：

V6001 在「&&」運算子的左邊和右邊有相同的子運算式「keyFrom == null」。 ReadOnlyWindowStoreStub.java 327、ReadOnlyWindowStoreStub.java 327

V6007 表達式「keyFrom == null」總是 false。 ReadOnlyWindowStoreStub.java 329

我們可以明白為什麼。對於每個開發人員來說，這種可笑的打字錯誤都是永恆的。雖然我們可以花很多時間尋找它們，但要回憶起它們潛伏的地方可不是小菜一碟。

在同一個類別中，另一個方法中存在完全相同的錯誤。我認為稱其為複製麵食是公平的。

@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);
    }
  }
  ....
}

以下是相同的警告：

V6007 表達式「keyFrom == null」總是 false。 ReadOnlyWindowStoreStub.java 273

V6001 在「&&」運算子的左邊和右邊有相同的子運算式「keyFrom == null」。 ReadOnlyWindowStoreStub.java 271, ReadOnlyWindowStoreStub.java 271

不用擔心——我們不必一次查看數百行程式碼。 PVS-Studio 非常擅長處理這類簡單的事情。解決一些更具挑戰性的事情怎麼樣？

可變同步

Java 中 synchronized 關鍵字的用途是什麼？在這裡，我將只關注同步方法，而不是區塊。根據 Oracle 文檔，synchronized 關鍵字將方法聲明為同步，以確保與實例的線程安全互動。如果一個執行緒呼叫該實例的同步方法，則嘗試呼叫相同實例的同步方法的其他執行緒將被阻塞（即它們的執行將被掛起）。它們將被阻塞，直到第一個執行緒呼叫的方法處理其執行。當實例對多個執行緒可見時，需要執行此操作。此類實例的讀取/寫入操作只能透過同步方法執行。

開發人員違反了 Sensor 類別中的規則，如下面的簡化程式碼片段所示。實例欄位的讀取/寫入操作可以透過同步和非同步兩種方式執行。它可能會導致競爭條件並使輸出變得不可預測。

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;
  }
}

分析器警告如下：

V6102 “metrics”欄位同步不一致。考慮在所有用途上同步該欄位。感測器.java 49，感測器.java 254

如果不同的執行緒可以同時變更實例狀態，則允許此操作的方法應該同步。如果程式沒有預料到多個執行緒可以與實例交互，則使其方法同步是沒有意義的。最壞的情況下，甚至會損害程式效能。

程式中有很多這樣的錯誤。這是分析器發出警告的類似程式碼片段：

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
  );
}

分析器警告：

V6102 “prefixKeyFormatter”欄位同步不一致。考慮在所有用途上同步該欄位。 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:

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 The type of object passed as argument is incompatible with the type of collection: String, Uuid. MockAdminClient.java 569

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:

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:

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:

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

The for loop entry is just syntactic sugar. The construction is equivalent to this one:

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: 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:

protected transient int modCount = 0;

Here are some usages of the modCount counter in the ArrayList class:

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:

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:

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:

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 Potential null dereference of 'oldMember' in function 'removeStaticMember'. ConsumerGroup.java 311, ConsumerGroup.java 323

@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:

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:

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 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.

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."

I almost forgot! Catch a link to learn more about a free license for open-source projects.

以上がBelay the Metamorphosis: Kafka プロジェクトの分析の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。