@Override
  public Optional<String> canUpgradeTo(final DataSource other) {
    final List<String> issues = PROPERTIES.stream()
        .filter(prop -> !prop.isCompatible(this, other))
        .map(prop -> getCompatMessage(other, prop))
        .collect(Collectors.toList());

    checkSchemas(getSchema(), other.getSchema())
        .map(s -> getCompatMessage(other, SCHEMA_PROP) + ". (" + s + ")")
        .ifPresent(issues::add);

    final String err = String.join("\n\tAND ", issues);
    return err.isEmpty() ? Optional.empty() : Optional.of(err);
  }
@Test
  public void shouldEnforceSameNameCompatbility() {
    // Given:
    final KsqlStream<String> streamA = new KsqlStream<>(
        "sql",
        SourceName.of("A"),
        SOME_SCHEMA,
        Optional.empty(),
        true,
        topic,
        false
    );
    final KsqlStream<String> streamB = new KsqlStream<>(
        "sql",
        SourceName.of("B"),
        SOME_SCHEMA,
        Optional.empty(),
        true,
        topic,
        false
    );

    // When:
    final Optional<String> err = streamA.canUpgradeTo(streamB);

    // Then:
    assertThat(err.isPresent(), is(true));
    assertThat(err.get(), containsString("has name = `A` which is not upgradeable to `B`"));
  }