Clever Geek Handbook

Big / Bug Data: Analizando el código fuente de Apache Flink

image1.png


Las aplicaciones de Big Data procesan grandes cantidades de información, a menudo en tiempo real. Naturalmente, tales aplicaciones deben ser altamente confiables para que ningún error en el código pueda interferir con el procesamiento de datos. Para lograr una alta confiabilidad, es necesario monitorear de cerca la calidad del código de proyectos desarrollados para esta área. El analizador estático PVS-Studio se ocupa de este problema. Hoy, el proyecto Apache Flink desarrollado por Apache Software Foundation, uno de los líderes en el mercado de software de Big Data, fue elegido como tema de prueba para el analizador.



¿Qué es Apache Flink? Es un marco de código abierto para el procesamiento distribuido de grandes cantidades de datos. Fue desarrollado como una alternativa a Hadoop MapReduce en 2010 en la Universidad Técnica de Berlín. El marco se basa en un motor de ejecución distribuido para aplicaciones de procesamiento por lotes y de flujo. Este motor está escrito en lenguajes Java y Scala. En la actualidad, Apache Flink se puede utilizar en proyectos escritos con Java, Scala, Python e incluso SQL.



Análisis de proyectos



Después de descargar el código fuente del proyecto, comencé a construir el proyecto con el comando 'mvn clean package -DskipTests' especificado en las instrucciones en GitHub . Mientras el ensamblaje estaba en progreso, usando la utilidad CLOC , descubrí que hay 10838 archivos Java en el proyecto, que tienen alrededor de 1.3 millones de líneas de código. Además, había 3833 archivos Java de prueba, que es más de 1/3 de todos los archivos Java. También noté que el proyecto usa el analizador de código estático FindBugs y la utilidad Cobertura, que proporciona información sobre la cobertura del código por pruebas. Con todo esto en mente, queda claro que los desarrolladores de Apache Flink tuvieron cuidado con la calidad del código y la cobertura de prueba durante el desarrollo.



Después de una compilación exitosa, abrí el proyecto en IntelliJ IDEA y lancé el análisis usando el complemento PVS-Studio para IDEA y Android Studio . Las advertencias del analizador se distribuyeron de la siguiente manera:



  • 183 alto;
  • 759 Medio;
  • 545 Bajo.


Aproximadamente 2/3 de los activadores del analizador PVS-Studio se asignaron a archivos de prueba. Teniendo en cuenta este hecho y el tamaño de la base de código del proyecto, podemos decir que los desarrolladores de Apache Flink lograron mantener la calidad del código en su mejor momento.



Habiendo estudiado las advertencias del analizador con más detalle, he elegido las más interesantes en mi opinión. ¡Así que veamos lo que PVS-Studio logró encontrar en este proyecto!





Solo un poco de descuido



V6001 Hay subexpresiones idénticas ' ProcessingData ' a la izquierda y a la derecha del operador '=='. 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);
}


En el contexto de otras expresiones a cambio, este error no es muy llamativo. Al anular el método equals para la clase CheckpointStatistics , el programador cometió un error en la expresión processData == processingData , que no tiene sentido porque siempre es verdadera. De manera similar, el resto de la expresión a cambio debía compararse el campo del objeto actual this y el objeto That : ProcessingData == that.processedData... Esta situación es uno de los patrones de error típicos que se encuentran en las funciones de comparación, que se describen en detalle en el artículo " Vidas malas en funciones de comparación ". Y entonces resulta que sólo "un poco de falta de atención" rompió la lógica para verificar la equivalencia de objetos de la clase CheckpointStatistics .



La expresión siempre es cierta



V6007 La expresión 'input2.length> 0' siempre es verdadera. 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 
  {
    ....
  }
}


En este método, el analizador resultó ser más atento que una persona, a la que decidió informar de una manera peculiar, lo que indica que la expresión input2.length> 0 siempre será verdadera. La razón es que si la longitud de la matriz input2 es 0, entonces la condición input2 == null || input2.length == 0 del primero if en el método será verdadero, y la ejecución del método será interrumpida antes de llegar a la línea con la expresión input2.length> 0 .



Analizador que todo lo ve



La expresión V6007 'slotSharingGroup == null' siempre es falsa. StreamGraphGenerator.java (510) 



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


El analizador informó que slotSharingGroup == null siempre es falso. Esto sugiere que el método determineSlotSharingGroup nunca devolverá un valor nulo . ¿Es el analizador tan inteligente que podría calcular todos los valores que este método puede devolver? Mejor revisemos todo nosotros mismos: 



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


En el orden en que revisamos todas las devoluciones y vemos qué puede recuperar este método:



  • La primera devolución devolverá el argumento al método especificadoGroup , pero solo si no es nulo .
  • return for DEFAULT_SLOT_SHARING_GROUP, ;
  • return inputGroup, null. DEFAULT_SLOT_SHARING_GROUP.


Resulta que el analizador realmente pudo calcular la imposibilidad de devolver un valor nulo desde el método determineSlotSharingGroup y nos advirtió sobre esto, señalando la falta de sentido de verificar slotSharingGroup == null . Y aunque esta situación no es errónea, dicha protección adicional del analizador puede detectar un error en algún otro caso. Por ejemplo, cuando desee que un método devuelva un valor nulo en determinadas condiciones.



Colecciónalos todos



V6007 La expresión 'currentCount <= lastEnd' siempre es verdadera. CountSlidingWindowAssigner.java (75)



V6007 La expresión 'lastStart <= currentCount' siempre es verdadera. 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;
}


El analizador advierte que las expresiones currentCount <= lastEnd y lastStart <= currentCount son siempre verdaderas. Y de hecho, si observa la condición del ciclo while , entonces existen exactamente las mismas expresiones. Esto significa que dentro del bucle estas expresiones siempre serán verdaderas, por lo que todos los objetos del tipo CountWindow creados en el bucle se agregarán a la lista de ventanas . Hay muchas opciones para la aparición de este cheque sin sentido, y lo primero que me viene a la mente es un artefacto de refactorización o la tranquilidad del desarrollador. Pero puede ser un error, si quisiera comprobar otra cosa ...



Orden de argumentos incorrecto



V6029 Posible orden incorrecto de los argumentos pasados ​​al método: '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);
  ....
}


La falta de capacidad de Java para llamar a un método con parámetros nombrados a veces juega una broma cruel con los desarrolladores. Esto es exactamente lo que sucedió cuando el analizador apuntó al método createMessageList . Al observar la definición de este método, queda claro que el parámetro hasBufferForRemovedChannel debe pasarse al método antes que el parámetro hasBufferForReleasedChannel : 



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


Sin embargo, al llamar al método, el desarrollador ha mezclado el orden de estos argumentos, por lo que la lógica del método createMessageList se romperá si los valores de los argumentos mixtos difieren.



Oh, este copiar y pegar



V6032 Es extraño que el cuerpo del método 'seekToFirst' sea totalmente equivalente al cuerpo de otro método '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();
  }
  
  ....
}


Los cuerpos de los métodos seekToFirst y seekToLast son los mismos. Además, ambos métodos se utilizan en el código.



¡Algo está inmundo aquí! De hecho, si observa qué métodos tiene el objeto iterador , quedará claro qué error ayudó a encontrar el analizador: 



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


Resulta que el método seekToLast clase RocksIteratorWrapper fue creado por el método de copiar y pegar seekToFirst de la misma clase. Sin embargo, por alguna razón, el desarrollador se olvidó de reemplazar el iterador 's seekToFirst método de llamada con seekToLast .



Confusión con cadenas de formato



V6046 Formato incorrecto. Se espera un número diferente de elementos de formato. Argumentos no utilizados: 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));
  }
}


Las cadenas de formato del método String.format y los registradores de Java son diferentes. En contraste con la cadena de formato del método String.format , donde las sustituciones de argumentos se especifican usando el carácter '%', las cadenas de formato del registrador usan la combinación de caracteres '{}' en su lugar. Debido a esta confusión, se produjo este error. Como cadena de formato, se pasa una cadena al método String.format , que lo más probable es que se haya copiado de otro lugar donde se usó en algún registrador. Como resultado, el valor del campo INITIALIZE_DB_MAX_RETRY no se sustituirá en el mensaje IllegalStateException . en lugar de '{}', y la persona que detecta o registra esta excepción nunca sabrá cuántos intentos de conectarse a la base de datos se realizaron.



Distribución anormal



V6048 Esta expresión se puede simplificar. El operando 'índice' en la operación es igual a 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();
}


El método de partición divide los elementos de la colección de elementos en varios segmentos y luego devuelve esos segmentos. Sin embargo, debido al error señalado por el analizador, no se producirá ninguna separación. La expresión utilizada para determinar el número de segmento index% numBuckets siempre será 0, porque el índice siempre es 0. Originalmente pensé que el código para este método fue refactorizado, como resultado de lo cual se olvidaron de agregar un incremento de la variable de índice en el ciclo for . Pero mirando el compromisodonde se agregó este método, resultó que este error vino junto con este método. Versión corregida del código: 



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


Tipo incompatible



V6066 El tipo de objeto pasado como argumento es incompatible con el tipo de colección: 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);
    }
    ....
  }
}


La expresión a la que apunta el analizador siempre será falsa, lo que significa que la llamada al método migrateNextTransactionalIdHindState nunca sucederá. ¿Cómo sucedió que alguien busca en una colección de tipo Set <String> un elemento de un tipo completamente diferente - ListStateDescriptor <FlinkKafkaProducer.NextTransactionalIdHint> ? Sin la ayuda del analizador, tal error, muy probablemente, habría vivido en el código durante mucho tiempo, ya que no llama la atención y es simplemente imposible encontrarlo sin una verificación exhaustiva de este método.



Cambio de variable no atómica



V6074 Modificación no atómica de variable volátil. Inspeccione 'currentNumAcknowledgedSubtaks'. PendingCheckpointStats.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;
  }
}


Más 3 advertencias más del analizador con el mismo método:



  • V6074 Modificación no atómica de variable volátil. Inspeccione 'currentStateSize'. PendingCheckpointStats.java (134)
  • V6074 Modificación no atómica de variable volátil. Inspeccione 'currentProcessedData'. PendingCheckpointStats.java (138)
  • V6074 Modificación no atómica de variable volátil. Inspeccione 'currentPersistedData'. PendingCheckpointStats.java (143)


El analizador sugirió que hasta 4 campos volátiles en el método cambian de forma no atómica. Y el analizador, como siempre, tiene razón, porque las operaciones ++ y + = son , de hecho, una secuencia de varias operaciones de lectura-modificación-escritura. Como sabe, el valor volátil de un campo es visible para todos los subprocesos, lo que significa que parte de los cambios del campo se pueden perder debido a una condición de carrera. Puede leer información más detallada sobre esto en la descripción de los diagnósticos.



Conclusión



En los proyectos de Big Data, la confiabilidad es uno de los requisitos clave; por lo tanto, la calidad del código en ellos debe ser monitoreada de cerca. Los desarrolladores de Apache Flink han sido asistidos en esto por varias herramientas, y también han escrito un número significativo de pruebas. Sin embargo, incluso en tales condiciones, el analizador PVS-Studio pudo encontrar errores. Es imposible deshacerse por completo de los errores, pero el uso regular de varias herramientas de análisis de código estático lo ayudará a acercarse a este ideal. Sí, exactamente con regularidad. Solo con el uso regular, el análisis estático muestra su efectividad, que se describe con más detalle en este artículo .





Si desea compartir este artículo con una audiencia de habla inglesa, utilice el enlace de traducción: Valery Komarov. Big / Bug Data: análisis del código fuente de Apache Flink .


More articles:


All Articles