@Override
  public void deleteSource(final SourceName sourceName, final boolean restoreInProgress) {
    synchronized (metaStoreLock) {
      dataSources.compute(sourceName, (ignored, sourceInfo) -> {
        if (sourceInfo == null) {
          throw new KsqlException(String.format("No data source with name %s exists.",
              sourceName.text()));
        }

        if (dropConstraints.containsKey(sourceName)) {
          final String references = dropConstraints.get(sourceName).stream().map(SourceName::text)
              .sorted().collect(Collectors.joining(", "));

          // If this request is part of the metastore restoration process, then ignore any
          // constraints this source may have. This logic fixes a compatibility issue caused by
          // https://github.com/confluentinc/ksql/pull/6545, which makes the restoration to fail if
          // this source has another source referencing to it.
          if (restoreInProgress) {
            LOG.warn("The following streams and/or tables read from the '{}' source: [{}].\n"
                    + "Ignoring DROP constraints when restoring the metastore. \n"
                    + "Future CREATE statements that recreate this '{}' source may not have "
                    + "DROP constraints if existing source references exist.",
                sourceName.text(), references);

            dropConstraints.remove(sourceName);
          } else {
            throw new KsqlReferentialIntegrityException(String.format(
                "Cannot drop %s.%n"
                    + "The following streams and/or tables read from this source: [%s].%n"
                    + "You need to drop them before dropping %s.",
                sourceName.text(),
                references,
                sourceName.text()
            ));
          }
        }

        // Remove drop constraints from the referenced sources
        sourceInfo.references.stream().forEach(ref -> dropConstraint(ref, sourceName));

        LOG.info("Source {} deleted from the metastore", sourceName.text());
        return null;
      });
    }
  }
@Test
  public void shouldThrowOnRemoveUnknownSource() {
    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> metaStore.deleteSource(of("bob"))
    );

    // Then:
    assertThat(e.getMessage(), containsString("No data source with name bob exists"));
  }