public static <T> IntermediateCompatibilityResult<T> constructIntermediateCompatibilityResult(
            TypeSerializerSnapshot<?>[] newNestedSerializerSnapshots,
            TypeSerializerSnapshot<?>[] oldNestedSerializerSnapshots) {

        Preconditions.checkArgument(
                newNestedSerializerSnapshots.length == oldNestedSerializerSnapshots.length,
                "Different number of new serializer snapshots and existing serializer snapshots.");

        TypeSerializer<?>[] nestedSerializers =
                new TypeSerializer[newNestedSerializerSnapshots.length];

        // check nested serializers for compatibility
        boolean nestedSerializerRequiresMigration = false;
        boolean hasReconfiguredNestedSerializers = false;
        for (int i = 0; i < oldNestedSerializerSnapshots.length; i++) {
            TypeSerializerSchemaCompatibility<?> compatibility =
                    resolveCompatibility(
                            newNestedSerializerSnapshots[i], oldNestedSerializerSnapshots[i]);

            // if any one of the new nested serializers is incompatible, we can just short circuit
            // the result
            if (compatibility.isIncompatible()) {
                return IntermediateCompatibilityResult.definedIncompatibleResult();
            }

            if (compatibility.isCompatibleAfterMigration()) {
                nestedSerializerRequiresMigration = true;
            } else if (compatibility.isCompatibleWithReconfiguredSerializer()) {
                hasReconfiguredNestedSerializers = true;
                nestedSerializers[i] = compatibility.getReconfiguredSerializer();
            } else if (compatibility.isCompatibleAsIs()) {
                nestedSerializers[i] = newNestedSerializerSnapshots[i].restoreSerializer();
            } else {
                throw new IllegalStateException("Undefined compatibility type.");
            }
        }

        if (nestedSerializerRequiresMigration) {
            return IntermediateCompatibilityResult.definedCompatibleAfterMigrationResult();
        }

        if (hasReconfiguredNestedSerializers) {
            return IntermediateCompatibilityResult.undefinedReconfigureResult(nestedSerializers);
        }

        // ends up here if everything is compatible as is
        return IntermediateCompatibilityResult.definedCompatibleAsIsResult(nestedSerializers);
    }
@Test
    void testCompatibleAfterMigrationIntermediateCompatibilityResult() {
        final TypeSerializerSnapshot<?>[] previousSerializerSnapshots =
                new TypeSerializerSnapshot<?>[] {
                    new SchemaCompatibilityTestingSerializer("a").snapshotConfiguration(),
                    new SchemaCompatibilityTestingSerializer("b").snapshotConfiguration(),
                    new SchemaCompatibilityTestingSerializer("c").snapshotConfiguration()
                };

        final TypeSerializerSnapshot<?>[] newSerializerSnapshots =
                new TypeSerializerSnapshot<?>[] {
                    SchemaCompatibilityTestingSnapshot
                            .thatIsCompatibleWithLastSerializerAfterReconfiguration("a"),
                    SchemaCompatibilityTestingSnapshot
                            .thatIsCompatibleWithLastSerializerAfterMigration("b"),
                    SchemaCompatibilityTestingSnapshot.thatIsCompatibleWithLastSerializer("c"),
                };

        IntermediateCompatibilityResult<?> intermediateCompatibilityResult =
                CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
                        newSerializerSnapshots, previousSerializerSnapshots);

        assertThat(intermediateCompatibilityResult.isCompatibleAfterMigration()).isTrue();
        assertThat(intermediateCompatibilityResult.getFinalResult().isCompatibleAfterMigration())
                .isTrue();
    }