# Big/Bug Data: анализируем исходный код Apache Flink Приложения, использующиеся в области Big Data, обрабатывают огромные объемы информации, причем часто это происходит в реальном времени\. Естественно, такие приложения должны обладать высокой надежностью, чтобы никакая ошибка в коде не могла помешать обработке данных\. Для достижения высокой надежности необходимо пристально следить за качеством кода проектов, разрабатываемых для этой области\. Решением данной проблемы и занимается статический анализатор PVS\-Studio\. Сегодня в качестве подопытного для анализатора был выбран проект Apache Flink, разработанный организацией Apache Software Foundation \- одним из лидеров на рынке ПО для Big Data\. ![0781_Apache_Flink_ru/image1.png](https://import.viva64.com/docx/blog/0781_Apache_Flink_ru/image1.png) Что же такое Apache Flink? Это open\-source фреймворк для распределенной обработки больших объемов данных\. Он был разработан как альтернатива Hadoop MapReduce в 2010 году в Техническом университете Берлина\. Основу фреймворка составляет движок распределенного исполнения приложений пакетной и потоковой обработки данных\. Написан этот движок на языках Java и Scala\. Сегодня Apache Flink возможно использовать в проектах, написанных с использованием языков Java, Scala, Python и даже SQL\. ## Анализ проекта Загрузив исходный код проекта, я запустил сборку проекта командой 'mvn clean package \-DskipTests', указанной в инструкции на [GitHub](https://github.com/apache/flink)\. Пока шла сборка, я при помощи утилиты [CLOC](https://github.com/AlDanial/cloc) выяснил что в проекте 10838 Java\-файлов, в которых имеется около 1\.3 миллионов строк кода\. Причем тестовых Java файлов оказалось аж 3833 штуки, а это больше 1/3 всех Java файлов\. Также я заметил, что в проекте используется статический анализатор кода FindBugs и утилита Cobertura, предоставляющая информацию о покрытии кода тестами\. Учитывая все это, становится ясно, что разработчики Apache Flink тщательно следили за качеством кода и покрытием тестами при разработке\. После удачной сборки я открыл проект в IntelliJ IDEA и запустил анализ при помощи плагина [PVS\-Studio for IDEA and Android Studio](https://plugins.jetbrains.com/plugin/12263-pvs-studio-for-idea-and-android-studio)\. Предупреждения анализатора распределились следующим образом: * 183 High; * 759 Medium; * 545 Low\. Примерно 2/3 срабатываний анализатора PVS\-Studio выданы на тестовые файлы\. Если учесть этот факт и размер кодовой базы проекта, то можно сказать, что разработчикам Apache Flink удалось сохранить качество кода на высоте\. Изучив предупреждения анализатора более подробно, я выбрал самые интересные на мой взгляд\. Так давайте же посмотрим, что же удалось найти PVS\-Studio в данном проекте\! [![getTrialImageLink](https://cdn.pvs-studio.ru/media/get_trial_insert_ru.png)](https://pvs-studio.ru/ru/pvs-studio-download/) ## Всего лишь немного невнимательности [V6001](https://pvs-studio.ru/ru/docs/warnings/v6001/) There are identical sub\-expressions 'processedData' to the left and to the right of the '\=\=' operator\. CheckpointStatistics\.java\(229\) ```cpp @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); } ``` На фоне других выражений в _return_ данная ошибка не сильно бросается в глаза\. При переопределении метода _equals_ для класса _CheckpointStatistics_ программист допустил ошибку в выражении _processedData \=\= processedData_, которое не имеет смысла, потому что всегда истинно\. Аналогично остальным выражениям в _return_ должны были сравниваться поля текущего объекта _this_ и объекта _that_: _processedData \=\= that\.processedData_\. Данная ситуация это один из типичных паттернов ошибок, найденных в функциях сравнения, которые подробно описаны в статье "[Зло живет в функциях сравнения](https://pvs-studio.ru/ru/blog/posts/cpp/0509/)"\. Вот так и получается, что всего лишь "немного невнимательности" сломало логику проверки эквивалентности объектов класса _CheckpointStatistics_\. ## Выражение всегда истинно [V6007](https://pvs-studio.ru/ru/docs/warnings/v6007/) Expression 'input2\.length \> 0' is always true\. Operator\.java\(283\) ```cpp public static Operator createUnionCascade(Operator input1, Operator... 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 \|\| input2\.length \=\= 0_ первого _if _в методе будет истинно, и выполнение метода будет прервано, так и не дойдя до строчки с выражением _input2\.length \> 0_\. ## Всевидящий анализатор [V6007](https://pvs-studio.ru/ru/docs/warnings/v6007/) Expression 'slotSharingGroup \=\= null' is always false\. StreamGraphGenerator\.java\(510\) ```cpp private Collection transformFeedback(....) { .... String slotSharingGroup = determineSlotSharingGroup(null, allFeedbackIds); if (slotSharingGroup == null) { slotSharingGroup = "SlotSharingGroup-" + iterate.getId(); } .... } ``` Анализатор сообщил, что выражение _slotSharingGroup \=\= null_ всегда ложно\. А это наталкивает на мысль о том, что метод _determineSlotSharingGroup_ никогда не вернет _null_\. Неужели анализатор настолько умный, что смог вычислить все значения, которые может вернуть этот метод? Давайте\-ка лучше все перепроверим сами: ```cpp public class StreamGraphGenerator { .... public static final String DEFAULT_SLOT_SHARING_GROUP = "default"; .... private String determineSlotSharingGroup(String specifiedGroup, Collection 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 } } .... } ``` По порядку пройдемся по всем _return_ и посмотрим, что же может вернуть данный метод: 1. В первом _return_ вернется аргумент метода _specifiedGroup_, но только если он не будет равен _null_\. 1. _return_ в цикле _for_ вернет значение статического финального поля _DEFAULT\_SLOT\_SHARING\_GROUP_, инициализированного строковым литералом; 1. И последний return в методе вернет значение переменной _inputGroup_, если оно не будет равно _null_\. В противном случае вернется значение поля _DEFAULT\_SLOT\_SHARING\_GROUP_\. Получается, что анализатор действительно смог вычислить невозможность возврата _null_ из метода _determineSlotSharingGroup_ и предупредил нас об этом, указав на бессмысленность проверки _slotSharingGroup \=\= null_\. И хотя данная ситуация не является ошибочной, однако подобная дополнительная защита анализатора сможет обнаружить ошибку в каком\-нибудь другом случае\. Например, когда необходимо, чтобы метод возвращал _null_ при определенных условиях\. ## Собери их всех [V6007](https://pvs-studio.ru/ru/docs/warnings/v6007/) Expression 'currentCount <\= lastEnd' is always true\. CountSlidingWindowAssigner\.java\(75\) [V6007](https://pvs-studio.ru/ru/docs/warnings/v6007/) Expression 'lastStart <\= currentCount' is always true\. CountSlidingWindowAssigner\.java\(75\) ```cpp @Override public Collection 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 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_ всегда истинны\. И ведь действительно, если посмотреть на условие цикла _while_, то там имеются точно такие же выражения\. Это значит, что внутри цикла эти выражения всегда будут истинны, поэтому в список _windows_ будут добавлены все объекты типа _CountWindow_ созданные в цикле\. Вариантов появления этой бессмысленной проверки множество, и первое, что приходит в голову, либо артефакт рефакторинга, либо перестраховка разработчика\. Но это может быть и ошибка, если хотелось проверить что\-то иное\.\.\. ## Некорректный порядок аргументов [V6029](https://pvs-studio.ru/ru/docs/warnings/v6029/) Possible incorrect order of arguments passed to method: 'hasBufferForReleasedChannel', 'hasBufferForRemovedChannel'\. NettyMessageClientDecoderDelegateTest\.java\(165\), NettyMessageClientDecoderDelegateTest\.java\(166\) ```cpp private void testNettyMessageClientDecoding( boolean hasEmptyBuffer, boolean hasBufferForReleasedChannel, boolean hasBufferForRemovedChannel) throws Exception { .... List messages = createMessageList ( hasEmptyBuffer, hasBufferForReleasedChannel, hasBufferForRemovedChannel); .... } ``` Отсутствие в Java возможности вызова метода с именованными параметрами иногда играет злую шутку с разработчиками\. Именно это и произошло при вызове метода _createMessageList_, на который указал анализатор\. При взгляде на определение этого метода становится ясно, что параметр _hasBufferForRemovedChannel_ должен передаваться в метод перед параметром _hasBufferForReleasedChannel_: ```cpp private List 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](https://pvs-studio.ru/ru/docs/warnings/v6032/) It is odd that the body of method 'seekToFirst' is fully equivalent to the body of another method 'seekToLast'\. RocksIteratorWrapper\.java\(53\), RocksIteratorWrapper\.java\(59\) ```cpp 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_ совпадают\. Причем оба метода используются в коде\. Что\-то здесь нечисто\! И действительно, если посмотреть какие методы имеются у объекта _iterator_, то станет понятно, какую ошибку помог найти анализатор: ```cpp public class RocksIterator extends AbstractRocksIterator { .... } 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_ у _iterator_ на _seekToLast_\. ## Путаница с форматными строками [V6046](https://pvs-studio.ru/ru/docs/warnings/v6046/) Incorrect format\. A different number of format items is expected\. Arguments not used: 1\. UnsignedTypeConversionITCase\.java\(102\) ```cpp 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_ передается строка, которая, скорее всего, была скопирована из другого места, где она использовалась в каком\-нибудь логгере\. В результате, в сообщении исключения _IllegalStateException_ не произойдет подстановки значения поля _INITIALIZE\_DB\_MAX\_RETRY_ вместо '\{\}', и тот, кто поймает или залоггирует это исключение, так и не узнает сколько попыток подключения к БД было произведено\. ## Ненормальное распределение [V6048](https://pvs-studio.ru/ru/docs/warnings/v6048/) This expression can be simplified\. Operand 'index' in the operation equals 0\. CollectionUtil\.java\(76\) ```cpp public static Collection> partition(Collection elements, int numBuckets) { Map> 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(); } ``` Метод _partition_ разделяет элементы из коллекции _elements_ на несколько сегментов, после чего возвращает эти сегменты\. Однако из\-за ошибки, на которую указал анализатор, никакого разделения происходить не будет\. Выражение, при помощи которого определяют номер сегмента _index % numBuckets_, всегда будет равно 0, потому что _index_ всегда равен 0\. Изначально я подумал, что код этого метода был подвергнут рефакторингу, в результате которого забыли добавить увеличение переменной _index_ в цикле _for_\. Но, просмотрев [коммит](https://github.com/apache/flink/commit/f63f7e6be29709b5871abd055ef1b12476952e75), где этот метод был добавлен, выяснилось, что эта ошибка появилась вместе с этим методом\. Исправленный вариант кода: ```cpp public static Collection> partition(Collection elements, int numBuckets) { Map> 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](https://pvs-studio.ru/ru/docs/warnings/v6066/) The type of object passed as argument is incompatible with the type of collection: String, ListStateDescriptor\. FlinkKafkaProducer\.java\(1083\) ```cpp public interface OperatorStateStore { Set getRegisteredStateNames(); } public class FlinkKafkaProducer extends .... { .... private static final ListStateDescriptor NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR = ....; @Override public void initializeState(FunctionInitializationContext context).... { .... if (context.getOperatorStateStore() .getRegisteredStateNames() .contains(NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR)) // <= { migrateNextTransactionalIdHindState(context); } .... } } ``` Выражение, на которое указал анализатор, всегда будет ложно, а значит вызова метода _migrateNextTransactionalIdHindState_ никогда не произойдет\. Как же так случилось, что кто\-то ищет в коллекции типа _Set_ элемент совсем другого типа \- _ListStateDescriptor_? Без помощи анализатора такая ошибка, скорее всего, очень долго бы жила в коде, так как в глаза она не бросается и без тщательной проверки данного метода её просто невозможно найти\. ## Неатомарное изменение переменной [V6074](https://pvs-studio.ru/ru/docs/warnings/v6074/) Non\-atomic modification of volatile variable\. Inspect 'currentNumAcknowledgedSubtasks'\. PendingCheckpointStats\.java\(131\) ```cpp 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 Non\-atomic modification of volatile variable\. Inspect 'currentStateSize'\. PendingCheckpointStats\.java\(134\) * V6074 Non\-atomic modification of volatile variable\. Inspect 'currentProcessedData'\. PendingCheckpointStats\.java\(138\) * V6074 Non\-atomic modification of volatile variable\. Inspect 'currentPersistedData'\. PendingCheckpointStats\.java\(143\) Анализатор подсказал, что аж 4 _volatile_ поля в методе изменяются неатомарно\. И анализатор, как всегда, оказывается прав, потому что операции _\+\+_ и _\+\=,_ на самом деле, \- это последовательность из нескольких операций чтения\-изменения\-записи\. Как известно, значение _volatile_ поля видно всем потокам, а это значит, что из\-за состояния гонки часть изменений поля может быть утеряна\. Более подробную информацию об этом вы можете прочитать в описании диагностики\. ## Заключение В Big Data проектах надежность является одним из ключевых требований, поэтому за качеством кода в них необходимо пристально следить\. В этом разработчикам Apache Flink помогали несколько инструментов, а также они написали значительное количество тестов\. Однако даже в таких условиях анализатор PVS\-Studio смог найти ошибки\. От ошибок невозможно полностью избавиться, но использование различных инструментов статического анализа кода регулярно позволит приблизится к этому идеалу\. Да\-да, именно регулярно\. Только при регулярном использовании статический анализ показывает свою эффективность, о чём более подробно рассказано в [этой статье](https://pvs-studio.ru/ru/blog/posts/0098/)\.