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

        final ImmutableSet.Builder<String> modifiedStreams = ImmutableSet.builder();
        final ImmutableSet.Builder<String> modifiedAlertConditions = ImmutableSet.builder();

        for (Document document : collection.find().sort(ascending(FIELD_CREATED_AT))) {

            final String streamId = document.getObjectId(FIELD_ID).toHexString();

            if (!document.containsKey(FIELD_ALERT_CONDITIONS)) {
                continue;
            }

            final List<Document> alertConditions = (List<Document>) document.get(FIELD_ALERT_CONDITIONS);

            // Need to check if the following fields are integers:
            //
            // FieldContentValue: grace, backlog
            // FieldValue:        grace, backlog, time, threshold
            // MessageCount:      grace, backlog, time, threshold
            final Set<String> intFields = ImmutableSet.of("grace", "backlog", "time", "threshold");

            for (Document alertCondition : alertConditions) {
                final String alertConditionId = alertCondition.get("id", String.class);
                final String alertConditionTitle = alertCondition.get("title", String.class);
                final Document parameters = alertCondition.get("parameters", Document.class);

                for (String field : intFields) {
                    final Object fieldValue = parameters.get(field);

                    // No need to convert anything if the field does not exist or is already an integer
                    if (fieldValue == null || fieldValue instanceof Integer) {
                        continue;
                    }

                    if (!(fieldValue instanceof String)) {
                        LOG.warn("Field <{}> in alert condition <{}> ({}) of stream <{}> is not a string but a <{}>, not trying to convert it!",
                                field, alertConditionId, alertConditionTitle, streamId,
                                fieldValue.getClass().getCanonicalName());
                        continue;
                    }

                    final String stringValue = parameters.get(field, String.class);
                    final Integer intValue = Ints.tryParse(stringValue);

                    LOG.info("Converting value for field <{}> from string to integer in alert condition <{}> ({}) of stream <{}>",
                            field, alertConditionId, alertConditionTitle, streamId);

                    if (intValue == null) {
                        LOG.error("Unable to parse \"{}\" into integer!", fieldValue);
                    }

                    final UpdateResult result = collection.updateOne(eq(FIELD_ALERT_CONDITIONS_ID, alertConditionId), set(ALERT_CONDITIONS_PARAMETERS_PREFIX + field, intValue));

                    // Use UpdateResult#getMatchedCount() instead of #getModifiedCount() to make it work on MongoDB 2.4
                    if (result.getMatchedCount() > 0) {
                        modifiedStreams.add(streamId);
                        modifiedAlertConditions.add(alertConditionId);
                    } else {
                        LOG.warn("No document modified for alert condition <{}> ({})", alertConditionId, alertConditionTitle);
                    }
                }
            }
        }

        clusterConfigService.write(MigrationCompleted.create(modifiedStreams.build(), modifiedAlertConditions.build()));
    }
@Test
    public void upgradeWhenMigrationCompleted() throws Exception {
        clusterConfigService.write(MigrationCompleted.create(Collections.emptySet(), Collections.emptySet()));

        // Reset the spy to be able to verify that there wasn't a write
        reset(clusterConfigService);

        migration.upgrade();

        verify(collection, never()).updateOne(any(), any(Bson.class));
        verify(clusterConfigService, never()).write(any(MigrationCompleted.class));
    }