junrao commented on code in PR #23832:
URL: https://github.com/apache/kafka/pull/23832#discussion_r4316998171
##########
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:
Should we add this check after the first transferFrom() too? Ditto in the
test below.
##########
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:
Could we add a javadoc that transferFrom() throws IOException, after which,
the channel shouldn't be used?
--
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]
