public ByteBuffer encode() {
        if (usedVersion == 3 || usedVersion == 4 || usedVersion == 5 || usedVersion == 6) {
            final byte[] endPointBytes = prepareUserEndPoint(this.userEndPoint);

            final ByteBuffer buf = ByteBuffer.allocate(
                4 + // used version
                    4 + // latest supported version version
                    16 + // client ID
                    4 + prevTasks.size() * 8 + // length + prev tasks
                    4 + standbyTasks.size() * 8 + // length + standby tasks
                    4 + endPointBytes.length
            );

            buf.putInt(usedVersion); // used version
            buf.putInt(LATEST_SUPPORTED_VERSION); // supported version
            encodeClientUUID(buf, processId().id());
            encodeTasks(buf, prevTasks, usedVersion);
            encodeTasks(buf, standbyTasks, usedVersion);
            encodeUserEndPoint(buf, endPointBytes);

            buf.rewind();

            return buf;
        } else if (usedVersion == 2) {
            final byte[] endPointBytes = prepareUserEndPoint(this.userEndPoint);

            final ByteBuffer buf = ByteBuffer.allocate(
                4 + // version
                    16 + // client ID
                    4 + prevTasks.size() * 8 + // length + prev tasks
                    4 + standbyTasks.size() * 8 + // length + standby tasks
                    4 + endPointBytes.length
            );

            buf.putInt(2); // version
            encodeClientUUID(buf, processId().id());
            encodeTasks(buf, prevTasks, usedVersion);
            encodeTasks(buf, standbyTasks, usedVersion);
            encodeUserEndPoint(buf, endPointBytes);

            buf.rewind();

            return buf;
        } else if (usedVersion == 1) {
            final ByteBuffer buf1 = ByteBuffer.allocate(
                4 + // version
                    16 + // client ID
                    4 + prevTasks.size() * 8 + // length + prev tasks
                    4 + standbyTasks.size() * 8
            );

            buf1.putInt(1); // version
            encodeClientUUID(buf1, processId().id());
            encodeTasks(buf1, prevTasks, usedVersion);
            encodeTasks(buf1, standbyTasks, usedVersion);
            buf1.rewind();
            return buf1;
        } else {
            throw new IllegalStateException("Unknown metadata version: " + usedVersion
                                                + "; latest supported version: " + LATEST_SUPPORTED_VERSION);
        }
    }
@throws TaskAssignmentException if method fails to encode the data
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/LegacySubscriptionInfoSerde.java