@Override
    public SchemaVersion versionFromBytes(byte[] version) {
        // The schema storage converts the schema from bytes to long
        // so it handles both cases 1) version is 64 bytes long pre 2.4.0;
        // 2) version is 8 bytes long post 2.4.0
        //
        // NOTE: if you are planning to change the logic here. you should consider
        //       both 64 bytes and 8 bytes cases.
        ByteBuffer bb = ByteBuffer.wrap(version);
        return new LongSchemaVersion(bb.getLong());
    }
@Test
    public void testVersionFromBytes() {
        long version = System.currentTimeMillis();

        ByteBuffer bbPre240 = ByteBuffer.allocate(Long.SIZE);
        bbPre240.putLong(version);
        byte[] versionBytesPre240 = bbPre240.array();

        ByteBuffer bbPost240 = ByteBuffer.allocate(Long.BYTES);
        bbPost240.putLong(version);
        byte[] versionBytesPost240 = bbPost240.array();

        PulsarService mockPulsarService = mock(PulsarService.class);
        when(mockPulsarService.getLocalMetadataStore()).thenReturn(mock(MetadataStoreExtended.class));
        BookkeeperSchemaStorage schemaStorage = new BookkeeperSchemaStorage(mockPulsarService);
        assertEquals(new LongSchemaVersion(version), schemaStorage.versionFromBytes(versionBytesPre240));
        assertEquals(new LongSchemaVersion(version), schemaStorage.versionFromBytes(versionBytesPost240));
    }