private static long recordSizeInBytes(final long keyBytes,
                                          final long valueBytes,
                                          final String topic,
                                          final Headers headers) {
        long headerSizeInBytes = 0L;

        if (headers != null) {
            for (final Header header : headers.toArray()) {
                headerSizeInBytes += Utils.utf8(header.key()).length;
                if (header.value() != null) {
                    headerSizeInBytes += header.value().length;
                }
            }
        }

        return keyBytes +
            valueBytes +
            8L + // timestamp
            8L + // offset
            Utils.utf8(topic).length +
            4L + // partition
            headerSizeInBytes;
    }
@throws StreamsException if the admin client request throws an exception
streams/src/main/java/org/apache/kafka/streams/processor/internals/ClientUtils.java