ByteBuffer serialize(final int endPadding) {

        final int sizeOfValueLength = Integer.BYTES;

        final int sizeOfPriorValue = priorValue == null ? 0 : priorValue.length;
        final int sizeOfOldValue = oldValue == null || priorValue == oldValue ? 0 : oldValue.length;
        final int sizeOfNewValue = newValue == null ? 0 : newValue.length;

        final byte[] serializedContext = recordContext.serialize();

        final ByteBuffer buffer = ByteBuffer.allocate(
            serializedContext.length
                + sizeOfValueLength + sizeOfPriorValue
                + sizeOfValueLength + sizeOfOldValue
                + sizeOfValueLength + sizeOfNewValue
                + endPadding
        );

        buffer.put(serializedContext);

        addValue(buffer, priorValue);

        if (oldValue == null) {
            buffer.putInt(NULL_VALUE_SENTINEL);
        } else if (Arrays.equals(priorValue, oldValue)) {
            buffer.putInt(OLD_PREV_DUPLICATE_VALUE_SENTINEL);
        } else {
            buffer.putInt(sizeOfOldValue);
            buffer.put(oldValue);
        }

        addValue(buffer, newValue);

        return buffer;
    }
@Test
    public void shouldSerializeNew() {
        final ProcessorRecordContext context = new ProcessorRecordContext(0L, 0L, 0, "topic", new RecordHeaders());
        final byte[] serializedContext = context.serialize();
        final byte[] newValue = {(byte) 5};
        final byte[] bytes = new BufferValue(null, null, newValue, context).serialize(0).array();
        final byte[] withoutContext = Arrays.copyOfRange(bytes, serializedContext.length, bytes.length);

        assertThat(withoutContext, is(ByteBuffer.allocate(Integer.BYTES * 3 + 1).putInt(-1).putInt(-1).putInt(1).put(newValue).array()));
    }