Можно ли при помощи Кафка стримов модифицировать сообщения внутри одного

топика? Нашел примеры где на входе input топик, а выходе output, но при таком кейсе объем данных в Кафке возрастает в два раза

Сообщения в топике это append-only log, поэтому нет. Можешь брать consumer и закидывать обратно в цикле, но сообщений всё равно будет на каждую итерацию больше

У Кафки знаю есть технология, которая позволяет для ключа хранить только последнее сообщение, если в консьюмере для всех сообщений использовать уникальный ключ , потом читать сообщение, преобразовывать и записывать с этим же ключом

Нет, хранит всегда все, а это скорее всего материализованное представлени

retentionPolicy: compact в настройках топика

Это поможет уменьшить обьем занимаемых данных в Кафке? Или хранится все равно будут все сообщения , а отображаться только последнее?

именно храниться будет последнее

Это всё равно тот же самый append-only лог, просто старые сообщения спустя некоторое время вычищаются. Но далеко не моментально.

Вот, при чтение с начала партиции мы всё равно их получаем же?

Нет, он удаляет старые сегменты и копирует то, что надо сохранить, в новые.

если кафка успеет уже сделать этот compaction, то не прочитаете уже

В активном сегменте compact не работает, поэтому для одного ключа возможна ситуация, что есть запись и в активном, и в сжатом сегментах. То есть, кафка будет хранить не строго одно последнее.