ビッグ/バグデータ：ApacheFlinkソースコードの分析

ビッグデータアプリケーションは、多くの場合リアルタイムで大量の情報を処理します。当然、このようなアプリケーションは、コードのエラーがデータ処理に干渉しないように、信頼性が高い必要があります。高い信頼性を実現するには、この分野で開発されたプロジェクトのコードの品質を注意深く監視する必要があります。 PVS-Studio静的アナライザーはこの問題に対処します。今日、ビッグデータソフトウェア市場のリーダーの1つであるApache SoftwareFoundationによって開発されたApacheFlinkプロジェクトが、アナライザーのテスト対象として選ばれました。



Apache Flinkとは何ですか？これは、大量のデータを分散処理するためのオープンソースフレームワークです。これは、2010年にベルリン工科大学でHadoopMapReduceの代替として開発されました。このフレームワークは、バッチおよびストリームデータ処理アプリケーション用の分散実行エンジンに基づいています。このエンジンはJavaとScalaで書かれています。現在、Apache Flinkは、Java、Scala、Python、さらにはSQLを使用して作成されたプロジェクトで使用できます。

プロジェクト分析

プロジェクトのソースコードをダウンロードした後、GitHubの手順で指定されたコマンド「mvncleanpackage-DskipTests」を使用してプロジェクトのビルドを開始しました 。アセンブリの進行中に、CLOCユーティリティを使用して、 プロジェクトに10838個のJavaファイルがあり、約130万行のコードがあることがわかりました。さらに、すでに3833個のテストJavaファイルがあり、これはすべてのJavaファイルの1/3以上です。また、プロジェクトでFindBugs静的コードアナライザーと、テストによるコードカバレッジに関する情報を提供するCoberturaユーティリティが使用されていることにも気づきました。これらすべてを念頭に置いて、ApacheFlink開発者が開発中のコード品質とテストカバレッジに注意を払っていたことが明らかになりました。



ビルドが成功した後、IntelliJ IDEAでプロジェクトを開き、PVS-Studio forIDEAとAndroidStudioプラグインを使用して分析を開始しました 。アナライザーの警告は次のように配布されました。

• 183高;
• 759中;
• 545低。

PVS-Studioアナライザートリガーの約2/3がテストファイルに割り当てられました。この事実とプロジェクトのコードベースのサイズを考慮すると、ApacheFlink開発者はコード品質を最高の状態に保つことができたと言えます。



アナライザーの警告をより詳細に調査したので、私は自分の意見で最も興味深いものを選択しました。それでは、PVS-Studioがこのプロジェクトで何を見つけたのか見てみましょう！

ちょっと不注意

V6001「==」演算子の左側と右側に同一のサブ式「processedData」があります。 CheckpointStatistics.java（229）

@Override
public boolean equals(Object o) 
{
  ....
  CheckpointStatistics that = (CheckpointStatistics) o;
  return id == that.id &&
    savepoint == that.savepoint &&
    triggerTimestamp == that.triggerTimestamp &&
    latestAckTimestamp == that.latestAckTimestamp &&
    stateSize == that.stateSize &&
    duration == that.duration &&
    alignmentBuffered == that.alignmentBuffered &&
    processedData == processedData &&                // <=
    persistedData == that.persistedData &&
    numSubtasks == that.numSubtasks &&
    numAckSubtasks == that.numAckSubtasks &&
    status == that.status &&
    Objects.equals(checkpointType, that.checkpointType) &&
    Objects.equals(
      checkpointStatisticsPerTask, 
      that.checkpointStatisticsPerTask);
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

見返りに他の表現の背景に対して 、このエラーはそれほど目立ったものではありません。CheckpointStatisticsクラスのequalsメソッドをオーバーライドする と、 プログラマーはprocessedData == processedData式でエラーを発生させました 。これは常にtrueであるため意味がありません。同様に、返される式の残りの部分は、 現在のオブジェクトthisとobject That： processedData == that.processedDataのフィールドと比較されます。 ..。この状況は、比較関数に見られる典型的なエラーパターンの1つであり、「悪は比較関数に住んでいます」の記事で詳しく説明されています 。そのため、「少しの不注意」だけで、CheckpointStatisticsクラスのオブジェクトの同等性をチェックするためのロジックが壊れたことが わかりました。

式は常に真です

V6007式 'input2.length> 0'は常に真です。Operator.java（283）

public static <T> Operator<T> createUnionCascade(Operator<T> input1, 
                                                 Operator<T>... input2) 
{
  if (input2 == null || input2.length == 0) 
  {
    return input1;                                // <=
  } 
  else if (input2.length == 1 && input1 == null) 
  {
    return input2[0];
  }
  ....
  if (input1 != null) 
  {
    ....
  } 
  else if (input2.length > 0 && input2[0] != null) // <=
  {
    ....
  } 
  else 
  {
    ....
  }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

この方法では、アナライザーは人よりも注意深いことが判明し、独自の方法で報告することを決定しました。これは、式 input2.length> 0が常に真であることを示しています。その理由は、input2配列の長さ が0の場合、条件 input2 == null ||であるためです。メソッド内の最初の ifのinput2.length == 0がtrueになり、式input2.length> 0の行に到達する前に、メソッドの実行が中断され ます。

オールシーイングアナライザー

V6007式 'slotSharingGroup == null'は常にfalseです。StreamGraphGenerator.java（510）

private <T> Collection<Integer> transformFeedback(....)
{
  ....
  String slotSharingGroup = determineSlotSharingGroup(null, allFeedbackIds);
  if (slotSharingGroup == null)
  {
    slotSharingGroup = "SlotSharingGroup-" + iterate.getId();
  }
  ....
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

アナライザーは、slotSharingGroup == nullが常にfalseであると報告し ました。これは、determineSlotSharingGroupメソッド がnullを返さないことを示してい ます。アナライザーは、このメソッドが返すことができるすべての値を計算することができたほどスマートですか？すべてを自分でチェックしてみましょう。

public class StreamGraphGenerator 
{
  ....
  public static final String DEFAULT_SLOT_SHARING_GROUP = "default";
  ....
  private String determineSlotSharingGroup(String specifiedGroup, 
                                           Collection<Integer> inputIds) 
  {
    if (specifiedGroup != null)
    {
      return specifiedGroup; // <= 1
    }
    else
    {
      String inputGroup = null;
      for (int id: inputIds)
      {
        String inputGroupCandidate = streamGraph.getSlotSharingGroup(id);
        if (inputGroup == null)
        {
          inputGroup = inputGroupCandidate;
        }
        else if (!inputGroup.equals(inputGroupCandidate))
        {
          return DEFAULT_SLOT_SHARING_GROUP; // <= 2
        }
      }
      return inputGroup == null 
             ? DEFAULT_SLOT_SHARING_GROUP 
             : inputGroup; // <= 3
    }
  }
  ....
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

すべてのリターンを通過する順序で、 このメソッドを取り戻すことができるものを確認します。

• 最初の戻り値は、指定されたGroupメソッドに引数を返しますが、nullでない場合に限ります。
• return for DEFAULT_SLOT_SHARING_GROUP, ;
• return inputGroup, null. DEFAULT_SLOT_SHARING_GROUP.

アナライザーは、determineSlotSharingGroupメソッドから nullを返すことの不可能性を実際に計算でき、 これについて警告し、slotSharingGroup == nullチェックが無意味であることを示していることが わかりました。この状況は誤りではありませんが、アナライザーのこのような追加の保護により、他の場合にエラーを検出できます。たとえば、特定の条件下でnullを返すメソッドが必要な場合です 。

それらをすべて集める

V6007式 ' currentCount <= lastEnd'は常にtrueです。 CountSlidingWindowAssigner.java（75）



V6007式 'lastStart <= currentCount'は常にtrueです。 CountSlidingWindowAssigner.java（75）

@Override
public Collection<CountWindow> assignWindows(....) throws IOException 
{
  Long countValue = count.value();
  long currentCount = countValue == null ? 0L : countValue;
  count.update(currentCount + 1);
  long lastId = currentCount / windowSlide;
  long lastStart = lastId * windowSlide;
  long lastEnd = lastStart + windowSize - 1;
  List<CountWindow> windows = new ArrayList<>();
  while (lastId >= 0 && 
         lastStart <= currentCount && 
         currentCount <= lastEnd) 
  {
    if (lastStart <= currentCount && currentCount <= lastEnd) // <=
    {
      windows.add(new CountWindow(lastId));
    }
    lastId--;
    lastStart -= windowSlide;
    lastEnd -= windowSlide;
  }
  return windows;
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

アナライザーは、currentCount <= lastEndおよび lastStart <= currentCountという式が常にtrueであることを警告し ます。実際、whileループの条件を見ると、 まったく同じ式があります。つまり、ループ内ではこれらの式は常にtrueになるため、ループ内で作成されたCountWindowタイプのすべてのオブジェクトがウィンドウリストに追加さ れ ます。この無意味なチェックの外観には多くのオプションがあり、最初に頭に浮かぶのは、リファクタリングアーティファクトまたは開発者の安心感です。しかし、何か他のものをチェックしたいのであれば、それは間違いかもしれません...

引数の順序が正しくありません

V6029メソッドに渡される引数の順序が正しくない可能性があります： 'hasBufferForReleasedChannel'、 'hasBufferForRemovedChannel'。NettyMessageClientDecoderDelegateTest.java（165）、NettyMessageClientDecoderDelegateTest.java（166）

private void testNettyMessageClientDecoding(
       boolean hasEmptyBuffer,
       boolean hasBufferForReleasedChannel,
       boolean hasBufferForRemovedChannel) throws Exception 
{
  ....
  List<BufferResponse> messages = createMessageList (
    hasEmptyBuffer,
    hasBufferForReleasedChannel,
    hasBufferForRemovedChannel);
  ....
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

名前付きパラメーターを使用してメソッドを呼び出す機能がJavaにないため、開発者は残酷な冗談を言うことがあります。これは、アナライザーがcreateMessageListメソッドをポイントしたときに起こったこととまったく同じ です。このメソッドの定義を見ると、hasBufferForReleasedChannelパラメーターの前にhasBufferForRemovedChannelパラメーターをメソッドに渡す必要があることが明らかになり ます 。

private List<BufferResponse> createMessageList(
  boolean hasEmptyBuffer,
  boolean hasBufferForRemovedChannel,
  boolean hasBufferForReleasedChannel) 
{
  ....
  if (hasBufferForReleasedChannel) {
    addBufferResponse(messages, 
                      releasedInputChannelId, 
                      Buffer.DataType.DATA_BUFFER, 
                      BUFFER_SIZE, 
                      seqNumber++);
  }
  if (hasBufferForRemovedChannel) {
    addBufferResponse(messages, 
                      new InputChannelID(), 
                      Buffer.DataType.DATA_BUFFER, 
                      BUFFER_SIZE, 
                      seqNumber++);
  }
  ....
  return messages;
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ただし、メソッドを呼び出すときに、開発者はこれらの引数の順序を混同しているため、混合された引数の値が異なる場合、createMessageListメソッドのロジック が壊れます。

ああ、このコピー＆ペースト

V6032メソッド「seekToFirst」の本体が別のメソッド「seekToLast」の本体と完全に同等であるのは奇妙です。RocksIteratorWrapper.java（53）、RocksIteratorWrapper.java（59）

public class RocksIteratorWrapper implements RocksIteratorInterface, Closeable {
  ....
  private RocksIterator iterator;
  ....

  @Override
  public void seekToFirst() {
    iterator.seekToFirst(); // <=
    status(); 
  }
  
  @Override
  public void seekToLast() {
    iterator.seekToFirst();  // <=
    status();
  }
  
  ....
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

遺体 seekToFirstと seekToLast方法がある同じ。さらに、両方のメソッドがコードで使用されています。



ここで何かが汚れています！実際、イテレーターオブジェクトのメソッドを見ると 、アナライザーがどのエラーを見つけるのに役立ったかが明らかになります。

public class RocksIterator extends AbstractRocksIterator<RocksDB>
{
  ....
}

public abstract class AbstractRocksIterator<....> extends ....
{
  ....
  public void seekToFirst() // <=
  {
    assert this.isOwningHandle();
    this.seekToFirst0(this.nativeHandle_);
  }
  
  public void seekToLast() // <=
  {
    assert this.isOwningHandle();
    this.seekToLast0(this.nativeHandle_);
  }
  ....
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

メソッド seekToLastクラス RocksIteratorWrapperは、同じクラスのコピーアンドペーストメソッドseekToFirstによって作成されたことが わかりました。ただし、何らかの理由で、開発者はイテレータの seekToFirstメソッド呼び出しをseekToLastに置き換えるのを忘れていまし た 。

フォーマット文字列との混同

V6046フォーマットが正しくありません。異なる数のフォーマットアイテムが予想されます。使用されていない引数：1。UnsignedTypeConversionITCase.java（102）

public static void prepareMariaDB() throws IllegalStateException {
  ....
  if (!initDbSuccess) {
    throw new IllegalStateException(
      String.format(
        "Initialize MySQL database instance failed after {} attempts," + // <=
        " please open an issue.", INITIALIZE_DB_MAX_RETRY));
  }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

String.formatメソッドとJavaロガーのフォーマット文字列 は異なります。引数の置換が「％」文字を使用して指定されるString.formatメソッドのフォーマット文字列とは異なり、 ロガーフォーマット文字列は代わりに「{}」文字の組み合わせを使用します。この混乱のため、このエラーが発生しました。フォーマット文字列として、文字列はString.formatメソッドに渡されます。このメソッドは 、ロガーで使用されていた別の場所からコピーされた可能性があります。その結果、INITIALIZE_DB_MAX_RETRYフィールド値はIllegalStateExceptionメッセージで置き換えられません 。 '{}'の代わりに、この例外をキャッチまたはログに記録した人は、データベースへの接続が何回試行されたかを知ることはできません。

異常な分布

V6048この式は簡略化できます。操作のOperand'index 'は0に等しい。CollectionUtil.java（76）

public static <T> Collection<List<T>> partition(Collection<T> elements, 
                                                int numBuckets) 
{
  Map<Integer, List<T>> buckets = new HashMap<>(numBuckets);
  
  int initialCapacity = elements.size() / numBuckets;

  int index = 0;
  for (T element : elements) 
  {
    int bucket = index % numBuckets;                                 // <=
    buckets.computeIfAbsent(bucket, 
                            key -> new ArrayList<>(initialCapacity))

           .add(element); 
  }

  return buckets.values();
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

パーティションメソッド は、要素コレクションの要素を複数のセグメントに分割してから、 それらのセグメントを返します。ただし、アナライザーで指摘されたエラーのため、分離は発生しません。インデックスは常に0であるため 、セグメント番号index％numBucketsを決定するために使用される式 は、常に0になります。当初、このメソッドのコードがリファクタリングされたと考えていました。その結果、forループに インデックス変数の増分を追加するのを忘れていました 。しかし、コミットを見て このメソッドが追加された場合、このエラーはこのメソッドに付随することが判明しました。コードの修正バージョン：

public static <T> Collection<List<T>> partition(Collection<T> elements, 
                                                int numBuckets) 
{
  Map<Integer, List<T>> buckets = new HashMap<>(numBuckets);
  
  int initialCapacity = elements.size() / numBuckets;

  int index = 0;
  for (T element : elements) 
  {
    int bucket = index % numBuckets; 
    buckets.computeIfAbsent(bucket, 
                            key -> new ArrayList<>(initialCapacity))
           .add(element);
    index++;
  }

  return buckets.values();
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

互換性のないタイプ

V6066引数として渡されたオブジェクトのタイプは、コレクションのタイプと互換性がありません：String、ListStateDescriptor <NextTransactionalIdHint>。FlinkKafkaProducer.java（1083）

public interface OperatorStateStore 
{
  Set<String> getRegisteredStateNames();
}
public class FlinkKafkaProducer<IN> extends ....
{
  ....
  private static final 
  ListStateDescriptor<FlinkKafkaProducer.NextTransactionalIdHint>
  NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR = ....;

  @Override
  public void initializeState(FunctionInitializationContext context).... 
  {
    ....
    if (context.getOperatorStateStore()
               .getRegisteredStateNames()
               .contains(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR))    // <=
    {
       migrateNextTransactionalIdHindState(context);
    }
    ....
  }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

アナライザーが指す式は常にfalseになります。つまり、migrateNextTransactionalIdHindStateメソッドの呼び出し は発生しません。Set <String> --ListStateDescriptor <FlinkKafkaProducer.NextTransactionalIdHint>タイプのコレクションで、まったく異なるタイプの要素 を誰かが探しているのは どうしてですか？アナライザーの助けがなければ、このようなエラーは非常に長い間コードに存在していた可能性があります。これは、目に入ったことがなく、このメソッドの徹底的なテストなしでは見つけることができないためです。

非原子変数の変化

V6074揮発性変数の非原子修飾。'currentNumAcknowledgedSubtasks'を検査します。保留中のチェックポイント統計.java（131）

boolean reportSubtaskStats(JobVertexID jobVertexId, SubtaskStateStats subtask) {
  TaskStateStats taskStateStats = taskStats.get(jobVertexId);

  if (taskStateStats != null && taskStateStats.reportSubtaskStats(subtask)) {
    currentNumAcknowledgedSubtasks++;                // <=
    latestAcknowledgedSubtask = subtask;

    currentStateSize += subtask.getStateSize();      // <=

    long processedData = subtask.getProcessedData();
    if (processedData > 0) {
      currentProcessedData += processedData;         // <=
    }

    long persistedData = subtask.getPersistedData();
    if (persistedData > 0) {
      currentPersistedData += persistedData;         // <=
    }
    return true;
  } else {
    return false;
  }
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

さらに、同じ方法でさらに3つのアナライザー警告：

• V6074揮発性変数の非原子修飾。'currentStateSize'を検査します。BookingCheckpointStats.java（134）
• V6074揮発性変数の非原子修飾。'currentProcessedData'を検査します。保留中のチェックポイント統計.java（138）
• V6074揮発性変数の非原子修飾。'currentPersistedData'を検査します。BookingCheckpointStats.java（143）

アナライザーは、メソッド内の4つの揮発性フィールドが非アトミックに変化することを示唆しました 。そして、いつものように、アナライザーは正しいことがわかります。なぜなら、++および + =操作は 、実際には、いくつかの読み取り-変更-書き込み操作のシーケンスだからです。ご存知のように、フィールドの揮発性の値は すべてのスレッドに表示されます。つまり、レースの状態により、フィールドの変更の一部が失われる可能性があります。これに関する詳細情報は、診断の説明で読むことができます。

結論

ビッグデータプロジェクトでは、信頼性が重要な要件の1つであるため、プロジェクト内のコードの品質を注意深く監視する必要があります。 Apache Flinkの開発者は、いくつかのツールによってこれを支援され、かなりの数のテストも作成しました。ただし、このような状況でも、PVS-Studioアナライザーはエラーを検出できました。エラーを完全に排除することは不可能ですが、さまざまな静的コード分析ツールを定期的に使用することで、この理想に近づくことができます。はい、正確に定期的に。静的分析は、定期的に使用する場合にのみその有効性を示します。これについては、この記事で詳しく説明し ます。

この記事を英語を話す聴衆と共有したい場合は、翻訳リンクValeryKomarovを使用してください。 ビッグ/バグデータ：ApacheFlinkソースコードの分析。