public static <T> T readJsonSR(
      @Nonnull final byte[] jsonWithMagic,
      final ObjectMapper mapper,
      final Class<? extends T> clazz
  ) throws IOException {
    if (!hasMagicByte(jsonWithMagic)) {
      // don't log contents of jsonWithMagic to avoid leaking data into the logs
      throw new KsqlException(
          "Got unexpected JSON serialization format that did not start with the magic byte. If "
              + "this stream was not serialized using the JsonSchemaConverter, then make sure "
              + "the stream is declared with JSON format (not JSON_SR).");
    }

    return mapper.readValue(
        jsonWithMagic,
        SIZE_OF_SR_PREFIX,
        jsonWithMagic.length - SIZE_OF_SR_PREFIX,
        clazz
    );
  }
@Test()
  public void shouldThrowOnStandardJsonConversion() {
    // Given:
    byte[] json = new byte[]{/* data */ 0x01};

    // When:
    final Exception e = assertThrows(
        Exception.class,
        () -> JsonSerdeUtils.readJsonSR(json, mapper, Object.class)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Got unexpected JSON serialization format that did not start with the magic byte"));
  }