public static void writeTaskIdTo(final TaskId taskId, final ByteBuffer buf, final int version) {
        buf.putInt(taskId.subtopology());
        buf.putInt(taskId.partition());
        if (version >= MIN_NAMED_TOPOLOGY_VERSION) {
            if (taskId.topologyName() != null) {
                buf.putInt(taskId.topologyName().length());
                for (final char c : taskId.topologyName().toCharArray()) {
                    buf.putChar(c);
                }
            } else {
                buf.putInt(0);
            }
        } else if (taskId.topologyName() != null) {
            throw new TaskAssignmentException("Named topologies are not compatible with protocol version " + version);
        }
    }
@throws IOException if cannot read from input stream
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java