@Override
  public GenericRecord parse(ByteBuffer bytes)
  {
    if (bytes.remaining() < 5) {
      throw new ParseException("record must have at least 5 bytes carrying version and schemaId");
    }

    byte version = bytes.get();
    if (version != V1) {
      throw new ParseException("found record of arbitrary version [%s]", version);
    }

    int schemaId = bytes.getInt();
    Schema schemaObj = schemaObjs.get(schemaId);
    if (schemaObj == null) {
      throw new ParseException("Failed to find schema for id [%s]", schemaId);
    }

    DatumReader<GenericRecord> reader = new GenericDatumReader<>(schemaObj);
    try (ByteBufferInputStream inputStream = new ByteBufferInputStream(Collections.singletonList(bytes))) {
      return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));
    }
    catch (EOFException eof) {
      // waiting for avro v1.9.0 (#AVRO-813)
      throw new ParseException(
          eof,
          "Avro's unnecessary EOFException, detail: [%s]",
          "https://issues.apache.org/jira/browse/AVRO-813"
      );
    }
    catch (Exception e) {
      throw new ParseException(e, "Fail to decode avro message with schemaId [%s].", schemaId);
    }
  }
@ Override public GenericRecord parse ( ByteBuffer bytes ) { if ( bytes . remaining ( ) < 5 ) { throw new ParseException ( "record must have at least 5 bytes carrying version and schemaId" ) ; } byte version = bytes . get ( ) ; if ( version != V1 ) { throw new ParseException ( "found record of arbitrary version [%s]" , version ) ; } int schemaId = bytes . getInt ( ) ; Schema schemaObj = schemaObjs . get ( schemaId ) ; if ( schemaObj == null ) { throw new ParseException ( "Failed to find schema for id [%s]" , schemaId ) ; } DatumReader < GenericRecord > reader = new GenericDatumReader < > ( schemaObj ) ; try ( ByteBufferInputStream inputStream = new ByteBufferInputStream ( Collections . singletonList ( bytes ) ) ) { return reader . read ( null , DecoderFactory . get ( ) . binaryDecoder ( inputStream , null ) ) ; } catch ( EOFException eof ) { throw new ParseException ( eof , "Avro's unnecessary EOFException, detail: [%s]" , "https://issues.apache.org/jira/browse/AVRO-813" ) ; } catch ( Exception e ) { throw new ParseException ( e , "Fail to decode avro message with schemaId [%s]." , schemaId ) ; } }
remaining bytes would have avro data