class SchemaRegistryTopicSchemaSupplier implements TopicSchemaSupplier {

  @Override
  public SchemaResult getValueSchema(
      final String topicName,
      final Optional<Integer> schemaId,
      final FormatInfo expectedFormat
  ) {
    try {
      final String subject = topicName + KsqlConstants.SCHEMA_REGISTRY_VALUE_SUFFIX;

      final int id;
      if (schemaId.isPresent()) {
        id = schemaId.get();
      } else {
        id = srClient.getLatestSchemaMetadata(subject).getId();
      }

      final ParsedSchema schema = srClient.getSchemaBySubjectAndId(subject, id);
      return fromParsedSchema(topicName, id, schema, expectedFormat);
    } catch (final RestClientException e) {
      switch (e.getStatus()) {
        case HttpStatus.SC_NOT_FOUND:
        case HttpStatus.SC_UNAUTHORIZED:
        case HttpStatus.SC_FORBIDDEN:
          return notFound(topicName);
        default:
          throw new KsqlException("Schema registry fetch for topic "
              + topicName + " request failed.", e);
      }
    } catch (final Exception e) {
      throw new KsqlException("Schema registry fetch for topic "
          + topicName + " request failed.", e);
    }
  }

  public  SchemaRegistryTopicSchemaSupplier(final SchemaRegistryClient srClient);
  @VisibleForTesting  SchemaRegistryTopicSchemaSupplier(
      final SchemaRegistryClient srClient,
      final Function<FormatInfo, Format> formatFactory
  );

}