@Override
  public void onDeserializationFailure(
      final String source,
      final String changelog,
      final byte[] data
  ) {
    // NOTE: this only happens for values, we should never auto-register key schemas
    final String sourceSubject = KsqlConstants.getSRSubject(source, false);
    final String changelogSubject = KsqlConstants.getSRSubject(changelog, false);

    // all schema registry events start with a magic byte 0x0 and then four bytes
    // indicating the schema id - we extract that schema id from the data that failed
    // to deserialize and then register it into the changelog subject
    final int id = ByteBuffer.wrap(data, 1, Integer.BYTES).getInt();
    final SchemaRegisterEvent event = new SchemaRegisterEvent(id, sourceSubject, changelogSubject);

    try {
      if (!failedAttempts.contains(event)) {
        LOG.info("Trying to fetch & register schema id {} under subject {}", id, changelogSubject);
        final ParsedSchema schema = srClient.getSchemaBySubjectAndId(sourceSubject, id);
        srClient.register(changelogSubject, schema);
      }
    } catch (Exception e) {
      LOG.warn("Failed during deserialization callback for topic {}. "
          + "Will not try again to register id {} under subject {}.",
          source,
          id,
          changelogSubject,
          e
      );

      failedAttempts.add(event);
    }
  }
@Test
  public void shouldRegisterOtherSchemaIdIfFirstFails() throws IOException, RestClientException {
    // Given:
    when(srClient.getSchemaBySubjectAndId(KsqlConstants.getSRSubject(SOURCE, false), ID2)).thenReturn(schema2);
    when(srClient.getSchemaBySubjectAndId(KsqlConstants.getSRSubject(SOURCE, false), ID)).thenReturn(schema);
    when(srClient.register(KsqlConstants.getSRSubject(CHANGELOG, false), schema)).thenThrow(new KsqlException(""));
    final RegisterSchemaCallback call = new RegisterSchemaCallback(srClient);

    // When:
    call.onDeserializationFailure(SOURCE, CHANGELOG, SOME_DATA);
    call.onDeserializationFailure(SOURCE, CHANGELOG, OTHER_DATA);

    // Then:
    verify(srClient, times(1)).getSchemaBySubjectAndId(KsqlConstants.getSRSubject(SOURCE, false), ID2);
    verify(srClient).register(KsqlConstants.getSRSubject(CHANGELOG, false), schema2);
  }