public static Optional<SchemaAndId> getLatestSchemaAndId(
      final SchemaRegistryClient srClient,
      final String topic,
      final boolean isKey
  ) {
    final String subject = KsqlConstants.getSRSubject(topic, isKey);

    return getLatestSchemaId(srClient, topic, isKey)
        .map(id -> {
          try {
            return new SchemaAndId(srClient.getSchemaById(id), id);
          } catch (final Exception e) {
            throwOnAuthError(e, subject);
            throw new KsqlException(
                "Could not get schema for subject " + subject + " and id " + id, e);
          }
        });
  }
@Test
  public void shouldReturnParsedSchemaFromSubjectKey() throws Exception {
    // Given:
    when(schemaMetadata.getId()).thenReturn(123);
    when(schemaRegistryClient.getLatestSchemaMetadata("bar-key"))
        .thenReturn(schemaMetadata);
    when(schemaRegistryClient.getSchemaById(123))
        .thenReturn(AVRO_SCHEMA);

    // When:
    final Optional<SchemaAndId> schemaAndId =
        SchemaRegistryUtil.getLatestSchemaAndId(schemaRegistryClient, "bar", true);

    // Then:
    assertThat(schemaAndId.get().getSchema(), equalTo(AVRO_SCHEMA));
  }