@Override
    public void upgrade() {
        final MigrationStatus migrationStatus = clusterConfigService.getOrDefault(MigrationStatus.class, MigrationStatus.createEmpty());

        final Stopwatch started = Stopwatch.createStarted();

        final MongoCollection<Document> jobTriggers = mongoConnection.getMongoDatabase().getCollection(DBJobTriggerService.COLLECTION_NAME);
        final MongoCollection<Document> jobDefinitions = mongoConnection.getMongoDatabase().getCollection(DBJobDefinitionService.COLLECTION_NAME);


        // Duplicate the type field from the JobDefinition config to their JobTriggers

        // We cannot use aggregations because Mongo 3.6 does not support $toString or $toObjectId
        final Map<String, String> typeMap = Streams.stream(jobDefinitions.find())
                .collect(Collectors.toMap(d -> d.getObjectId("_id").toString(), d -> d.getEmbedded(ImmutableList.of("config", "type"), String.class)));

        final FindIterable<Document> query = jobTriggers.find(Filters.exists(JobTriggerDto.FIELD_JOB_DEFINITION_TYPE, false));

        List<UpdateOneModel<Document>> typeUpdate = Streams.stream(query).map(d -> new UpdateOneModel<Document>(
                        Filters.eq("_id", d.getObjectId("_id")),
                        Updates.set(JobTriggerDto.FIELD_JOB_DEFINITION_TYPE, typeMap.get(d.getString(JobTriggerDto.FIELD_JOB_DEFINITION_ID)))))
                .collect(Collectors.toList());

        long modifiedCount = 0;
        if (!typeUpdate.isEmpty()) {
            modifiedCount = jobTriggers.bulkWrite(typeUpdate).getModifiedCount();

            LOG.info("Added type field to <{}> JobTriggers", modifiedCount);
        }
        final long elapsed = started.stop().elapsed(TimeUnit.SECONDS);
        if (elapsed > 60) {
            LOG.warn("Migration ran longer than expected! Took <{}> seconds", elapsed);
        }

        clusterConfigService.write(MigrationStatus.
                create(migrationStatus.convertedTriggersCount() + modifiedCount,
                        migrationStatus.executionCount() + 1));
    }
@Test
    @MongoDBFixtures("V20220623125450_AddJobTypeToJobTriggerTest.json")
    void upgrade() {
        long before = collection.countDocuments(Filters.exists(JobTriggerDto.FIELD_JOB_DEFINITION_TYPE, false));
        Assertions.assertThat(before).isEqualTo(3);

        migration.upgrade();

        long after = collection.countDocuments(Filters.exists(JobTriggerDto.FIELD_JOB_DEFINITION_TYPE, true));
        Assertions.assertThat(after).isEqualTo(3);

        final long eventTypeCount = collection.countDocuments(Filters.eq(JobTriggerDto.FIELD_JOB_DEFINITION_TYPE, "event-processor-execution-v1"));
        Assertions.assertThat(eventTypeCount).isEqualTo(2);
        final long notificationTypeCount = collection.countDocuments(Filters.eq(JobTriggerDto.FIELD_JOB_DEFINITION_TYPE, "notification-execution-v1"));
        Assertions.assertThat(notificationTypeCount).isEqualTo(1);
    }