class TopicDeleteInjector implements Injector {

  @SuppressWarnings("unchecked")
  @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) {
      checkTopicRefs(source);
      try {
        ExecutorUtil.executeWithRetries(
            () -> topicClient.deleteTopics(ImmutableList.of(source.getKafkaTopicName())),
            ExecutorUtil.RetryBehaviour.ALWAYS);
      } catch (final Exception e) {
        throw new RuntimeException("Could not delete the corresponding kafka topic: "
                + source.getKafkaTopicName(), e);
      }

      try {
        final Format valueFormat = FormatFactory
            .fromName(source.getKsqlTopic().getValueFormat().getFormat());

        if (valueFormat.supportsFeature(SerdeFeature.SCHEMA_INFERENCE)) {
          SchemaRegistryUtil.deleteSubjectWithRetries(
                  schemaRegistryClient,
                  source.getKafkaTopicName() + KsqlConstants.SCHEMA_REGISTRY_VALUE_SUFFIX);
        }
      } catch (final Exception e) {
        checkSchemaError(e, source.getKafkaTopicName());
      }
    } 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);
  }

  public  TopicDeleteInjector(
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  );
  @VisibleForTesting  TopicDeleteInjector(
      final MetaStore metastore,
      final KafkaTopicClient serviceContext,
      final SchemaRegistryClient schemaRegistryClient
  );

}