public static int registerSchema(
      final SchemaRegistryClient srClient,
      final ParsedSchema parsedSchema,
      final String topic,
      final String subject,
      final boolean isKey
  ) throws KsqlSchemaAuthorizationException, KsqlException {
    try {
      if (parsedSchema instanceof ProtobufSchema) {
        final ProtobufSchema resolved = AbstractKafkaProtobufSerializer.resolveDependencies(
            srClient,
            true,
            false,
            true,
            null,
            new DefaultReferenceSubjectNameStrategy(),
            topic,
            isKey,
            (ProtobufSchema) parsedSchema
        );
        return srClient.register(subject, resolved);
      } else {
        return srClient.register(subject, parsedSchema);
      }
    } catch (IOException | RestClientException e) {
      if (SchemaRegistryUtil.isAuthErrorCode(e)) {
        final AclOperation deniedOperation = SchemaRegistryUtil.getDeniedOperation(e.getMessage());

        if (deniedOperation != AclOperation.UNKNOWN) {
          throw new KsqlSchemaAuthorizationException(
              deniedOperation,
              subject);
        }
      }

      throw new KsqlException("Could not register schema for topic: " + e.getMessage(), e);
    }
  }
@Test
  public void shouldThrowKsqlSchemaAuthorizationException() throws Exception {
    // Given:
    when(schemaRegistryClient.register(anyString(), any(ParsedSchema.class)))
        .thenThrow(new RestClientException("User is denied operation Write on Subject", 403, 40301));

    // When:
    final Exception e = assertThrows(
        KsqlSchemaAuthorizationException.class,
        () -> SchemaRegistryUtil.registerSchema(schemaRegistryClient, AVRO_SCHEMA, "topic", "subject", false)
    );

    // Then:
    assertThat(e.getMessage(), equalTo(
        "Authorization denied to Write on Schema Registry subject: [subject]"));
  }