@SuppressWarnings({"unchecked", "UnstableApiUsage"})
  @Override
  public <T extends Statement> ConfiguredStatement<T> inject(
      final ConfiguredStatement<T> statement) {
    if (!(statement.getStatement() instanceof DropStatement)) {
      return statement;
    }

    final DropStatement dropStatement = (DropStatement) statement.getStatement();
    if (!dropStatement.isDeleteTopic()) {
      return statement;
    }

    final SourceName sourceName = dropStatement.getName();
    final DataSource source = metastore.getSource(sourceName);

    if (source != null) {
      if (source.isSource()) {
        throw new KsqlException("Cannot delete topic for read-only source: " + sourceName.text());
      }
      checkTopicRefs(source);

      deleteTopic(source);

      final Closer closer = Closer.create();
      closer.register(() -> deleteKeySubject(source));
      closer.register(() -> deleteValueSubject(source));
      try {
        closer.close();
      } catch (final KsqlException e) {
        throw e;
      } catch (final Exception e) {
        throw new KsqlException(e);
      }
    } else if (!dropStatement.getIfExists()) {
      throw new KsqlException("Could not find source to delete topic for: " + statement);
    }

    final T withoutDelete = (T) dropStatement.withoutDeleteClause();
    final String withoutDeleteText = SqlFormatter.formatSql(withoutDelete) + ";";

    return statement.withStatement(withoutDeleteText, withoutDelete);
  }
@Test
  public void shouldDeleteProtoSchemaInSR() throws IOException, RestClientException {
    // Given:
    when(topic.getValueFormat()).thenReturn(ValueFormat.of(FormatInfo.of(FormatFactory.PROTOBUF.name()),
        SerdeFeatures.of()));

    // When:
    deleteInjector.inject(DROP_WITH_DELETE_TOPIC);

    // Then:
    verify(registryClient).deleteSubject(KsqlConstants.getSRSubject("something", false));
  }