fix: fully consume index when transforming

[jeqo]

As it is currently validated, there is a chance indexes are not fully consumed when transforming, as the validation is assuming the whole stream is consumed, when it is not.

final var inputStream = transformFinisher.nextElement();
segmentIndexBuilder.add(indexType, singleChunk(transformFinisher.chunkIndex()).range().size());
return inputStream;

By introducing an output stream to hold the processed content, and passing the buffer to the storage layer we guarantee that the validation is correctly validating that there is only a single chunk processed.

[ivanyu]

Do I understand correctly that the problem is that the index may be longer than nextElement()? Or something else?

[jeqo]

Yes, that was my understanding; however I may have rushed a bit here and actually it's working correctly (though not obvious):

• Chunk size is about source, not target content. This means that even if the transformation changes the size (e.g. encryption) if the chunk size is equal to the original size, that's enough to get a single chunk as a result.
• The chunk index is fully built by next element if there are no more elements, validating the conditions to fully read the original content.

I got confused by the mixture of a outdated past memories. I will create a quick PR to document these less obvious conditions so the future me doesn't get confused again 😅

[jeqo]

Added doc comments as part of https://github.com/Aiven-Open/tiered-storage-for-apache-kafka/pull/535