@Override
    public void upgrade() {
        if (clusterConfigService.get(MigrationCompleted.class) != null) {
            LOG.debug("Migration already completed.");
            return;
        }

        var result = collection.find(Filters.and(
                Filters.eq("config.type", "aggregation-v1"),
                Filters.type(SERIES_PATH_STRING, "array"),
                Filters.exists(SERIES_PATH_STRING + ".0")
        ));

        var bulkOperations = new ArrayList<WriteModel<? extends Document>>();
        for (Document doc : result) {
            processDoc(doc, bulkOperations);
        }
        if (bulkOperations.size() > 0) {
            collection.bulkWrite(bulkOperations);
        }
        this.clusterConfigService.write(new MigrationCompleted());
    }
@Test
    public void doesNotRunAgainIfMigrationHadCompletedBefore() {
        when(clusterConfigService.get(V20230629140000_RenameFieldTypeOfEventDefinitionSeries.MigrationCompleted.class))
                .thenReturn(new V20230629140000_RenameFieldTypeOfEventDefinitionSeries.MigrationCompleted());

        this.migration.upgrade();

        verify(clusterConfigService, never()).write(any());
    }