class SubscriptionInfo {

    public static SubscriptionInfo decode(ByteBuffer data) {
        // ensure we are at the beginning of the ByteBuffer
        data.rewind();

        // Decode version
        int version = data.getInt();
        if (version == CURRENT_VERSION || version == 1) {
            // Decode client UUID
            UUID processId = new UUID(data.getLong(), data.getLong());
            // Decode previously active tasks
            Set<TaskId> prevTasks = new HashSet<>();
            int numPrevs = data.getInt();
            for (int i = 0; i < numPrevs; i++) {
                TaskId id = TaskId.readFrom(data);
                prevTasks.add(id);
            }
            // Decode previously cached tasks
            Set<TaskId> standbyTasks = new HashSet<>();
            int numCached = data.getInt();
            for (int i = 0; i < numCached; i++) {
                standbyTasks.add(TaskId.readFrom(data));
            }

            String userEndPoint = null;
            if (version == CURRENT_VERSION) {
                int bytesLength = data.getInt();
                if (bytesLength != 0) {
                    byte[] bytes = new byte[bytesLength];
                    data.get(bytes);
                    userEndPoint = new String(bytes, Charset.forName("UTF-8"));
                }

            }
            return new SubscriptionInfo(version, processId, prevTasks, standbyTasks, userEndPoint);

        } else {
            TaskAssignmentException ex = new TaskAssignmentException("unable to decode subscription data: version=" + version);
            log.error(ex.getMessage(), ex);
            throw ex;
        }
    }

    public  SubscriptionInfo(UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint);
    private  SubscriptionInfo(int version, UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint);

}