cheeeee commented on code in PR #23832:
URL: https://github.com/apache/kafka/pull/23832#discussion_r4320285787
##########
clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java:
##########
@@ -1283,6 +1290,106 @@ public void testInvalidSslEngineFactory(Args args) {
assertThrows(KafkaException.class, () ->
createSelector(args.sslClientConfigs));
}
+ /**
+ * A failed read of the file being transferred is rethrown, even after
part of the transfer was sent, and must not
+ * leave the previous read in the buffer to be sent as the contents of the
failed position.
+ */
+ @ParameterizedTest
+ @ArgumentsSource(SslTransportLayerArgumentsProvider.class)
+ public void testTransferFromDoesNotSendStaleBytesAfterFailedRead(Args
args) throws Exception {
+ ByteArrayOutputStream sent = new ByteArrayOutputStream();
+ SslTransportLayer layer = connectRecordingWrites(args, sent);
+ FileChannel fileChannel = fileChannelReadableUpTo(TRANSFER_SIZE, () ->
{
+ throw new IOException("Input/output error");
+ });
+
+ // The readable part is sent, then the read of the next part fails
+ assertThrows(IOException.class, () -> layer.transferFrom(fileChannel,
0, 3 * TRANSFER_SIZE));
+ // A transfer from the unreadable position must fail as well instead
of sending what was read before
+ assertThrows(IOException.class, () -> layer.transferFrom(fileChannel,
TRANSFER_SIZE, 2 * TRANSFER_SIZE));
+ assertArrayEquals(fileContents(TRANSFER_SIZE), sent.toByteArray());
Review Comment:
Done, in both tests.
##########
clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java:
##########
@@ -1031,33 +1031,33 @@ public long transferFrom(FileChannel fileChannel, long
position, long count) thr
int totalBytesWritten = 0;
long pos = position;
- try {
- while (totalBytesWritten < totalBytesToWrite) {
- if (!fileChannelBuffer.hasRemaining()) {
- fileChannelBuffer.clear();
- int bytesRemaining = totalBytesToWrite - totalBytesWritten;
- if (bytesRemaining < fileChannelBuffer.limit())
- fileChannelBuffer.limit(bytesRemaining);
- int bytesRead = fileChannel.read(fileChannelBuffer, pos);
- if (bytesRead <= 0)
- break;
+ while (totalBytesWritten < totalBytesToWrite) {
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
