@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
    @MongoDBFixtures("V20170110150100_FixAlertConditionsMigration.json")
    public void upgrade() throws Exception {
        // First check all types of the existing documents
        AlertConditionAssertions.assertThat(getAlertCondition("2fa6a415-ce0c-4a36-accc-dd9519eb06d9"))
                .hasParameter("backlog", 2)
                .hasParameter("grace", 1)
                .hasParameter("threshold_type", "MORE")
                .hasParameter("threshold", "5")
                .hasParameter("time", "1");

        AlertConditionAssertions.assertThat(getAlertCondition("393fd8b2-9b17-42d3-86b0-6e55d0f5343a"))
                .hasParameter("backlog", 0)
                .hasParameter("field", "bar")
                .hasParameter("grace", "0")
                .hasParameter("value", "baz");

        AlertConditionAssertions.assertThat(getAlertCondition("0e75404f-c0ee-40b0-8872-b1aec441ba1c"))
                .hasParameter("backlog", "0")
                .hasParameter("field", "foo")
                .hasParameter("grace", "0")
                .hasParameter("threshold_type", "HIGHER")
                .hasParameter("threshold", "0")
                .hasParameter("time", "5")
                .hasParameter("type", "MAX");

        // Run the migration that should convert all affected fields to integers
        migration.upgrade();

        // Check all types again
        AlertConditionAssertions.assertThat(getAlertCondition("2fa6a415-ce0c-4a36-accc-dd9519eb06d9"))
                .hasParameter("backlog", 2)
                .hasParameter("grace", 1)
                .hasParameter("threshold_type", "MORE")
                .hasParameter("threshold", 5)
                .hasParameter("time", 1);

        AlertConditionAssertions.assertThat(getAlertCondition("393fd8b2-9b17-42d3-86b0-6e55d0f5343a"))
                .hasParameter("backlog", 0)
                .hasParameter("field", "bar")
                .hasParameter("grace", 0)
                .hasParameter("value", "baz");

        AlertConditionAssertions.assertThat(getAlertCondition("0e75404f-c0ee-40b0-8872-b1aec441ba1c"))
                .hasParameter("backlog", 0)
                .hasParameter("field", "foo")
                .hasParameter("grace", 0)
                .hasParameter("threshold_type", "HIGHER")
                .hasParameter("threshold", 0)
                .hasParameter("time", 5)
                .hasParameter("type", "MAX");

        final MigrationCompleted migrationCompleted = clusterConfigService.get(MigrationCompleted.class);

        assertThat(migrationCompleted).isNotNull();

        assertThat(migrationCompleted.streamIds()).containsOnly("58458e442f857c314491344e", "58458e442f857c314491345e");
        assertThat(migrationCompleted.alertConditionIds()).containsOnly("2fa6a415-ce0c-4a36-accc-dd9519eb06d9", "393fd8b2-9b17-42d3-86b0-6e55d0f5343a", "0e75404f-c0ee-40b0-8872-b1aec441ba1c");
    }