jt2594838 commented on code in PR #18051:
URL: https://github.com/apache/iotdb/pull/18051#discussion_r3526075948
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
Review Comment:
i18n
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java:
##########
@@ -612,73 +612,114 @@ protected void transferFilePieces(
final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
final byte[] readBuffer = new byte[readFileBufferSize];
long position = 0;
- while (true) {
- mayLimitRateAndRecordIO(readFileBufferSize);
- final int readLength = reader.read(readBuffer);
- if (readLength == -1) {
- break;
- }
+ int readLength;
+ while ((readLength = readNextFilePiece(reader, readBuffer,
readFileBufferSize)) != -1) {
+ position =
+ transferFilePiece(
+ pipe2WeightMap,
+ file,
+ clientAndStatus,
+ isMultiFile,
+ readBuffer,
+ position,
+ readLength,
+ reader);
+ }
+ }
+ }
- final byte[] payLoad =
- readLength == readFileBufferSize
- ? readBuffer
- : Arrays.copyOfRange(readBuffer, 0, readLength);
- final PipeTransferFilePieceResp resp;
- try {
- final TPipeTransferReq req =
- compressIfNeeded(
- isMultiFile
- ? getTransferMultiFilePieceReq(file.getName(), position,
payLoad)
- : getTransferSingleFilePieceReq(file.getName(),
position, payLoad));
- pipe2WeightMap.forEach(
- (namePair, weight) ->
- rateLimitIfNeeded(
- namePair.getLeft(),
- namePair.getRight(),
- clientAndStatus.getLeft().getEndPoint(),
- (long) (req.getBody().length * weight)));
- resp =
- PipeTransferFilePieceResp.fromTPipeTransferResp(
- clientAndStatus.getLeft().pipeTransfer(req));
- } catch (final Exception e) {
- clientAndStatus.setRight(false);
- throw new PipeConnectionException(
- String.format(
- "Network error when transfer file %s, because %s.", file,
e.getMessage()),
- e);
- }
+ private int readNextFilePiece(
+ final RandomAccessFile reader, final byte[] readBuffer, final int
readFileBufferSize)
+ throws IOException {
+ mayLimitRateAndRecordIO(readFileBufferSize);
+ return reader.read(readBuffer);
+ }
- position += readLength;
+ private long transferFilePiece(
+ final Map<Pair<String, Long>, Double> pipe2WeightMap,
+ final File file,
+ final Pair<IoTDBSyncClient, Boolean> clientAndStatus,
+ final boolean isMultiFile,
+ final byte[] readBuffer,
+ final long position,
+ final int readLength,
+ final RandomAccessFile reader)
+ throws PipeException, IOException {
+ final byte[] payLoad = buildFilePiecePayload(readBuffer, readLength);
+ final PipeTransferFilePieceResp resp =
+ doTransferFilePiece(pipe2WeightMap, file, clientAndStatus,
isMultiFile, payLoad, position);
+ return handleTransferFilePieceResp(file, clientAndStatus, reader, position
+ readLength, resp);
+ }
- final TSStatus status = resp.getStatus();
- if (status.getCode() ==
TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET.getStatusCode()) {
- position = resp.getEndWritingOffset();
- reader.seek(position);
- LOGGER.info(DataNodePipeMessages.REDIRECT_FILE_POSITION_TO,
position);
- continue;
- }
+ private byte[] buildFilePiecePayload(final byte[] readBuffer, final int
readLength) {
+ return readLength == readBuffer.length
+ ? readBuffer
+ : Arrays.copyOfRange(readBuffer, 0, readLength);
+ }
- if (status.getCode()
- ==
TSStatusCode.PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED.getStatusCode()) {
- getClientManager().sendHandshakeReq(clientAndStatus);
- }
+ private PipeTransferFilePieceResp doTransferFilePiece(
+ final Map<Pair<String, Long>, Double> pipe2WeightMap,
+ final File file,
+ final Pair<IoTDBSyncClient, Boolean> clientAndStatus,
+ final boolean isMultiFile,
+ final byte[] payLoad,
+ final long position)
+ throws PipeException, IOException {
+ try {
+ final TPipeTransferReq req =
+ compressIfNeeded(
+ isMultiFile
+ ? getTransferMultiFilePieceReq(file.getName(), position,
payLoad)
+ : getTransferSingleFilePieceReq(file.getName(), position,
payLoad));
+ pipe2WeightMap.forEach(
+ (namePair, weight) ->
+ rateLimitIfNeeded(
+ namePair.getLeft(),
+ namePair.getRight(),
+ clientAndStatus.getLeft().getEndPoint(),
+ (long) (req.getBody().length * weight)));
+ return PipeTransferFilePieceResp.fromTPipeTransferResp(
+ clientAndStatus.getLeft().pipeTransfer(req));
+ } catch (final Exception e) {
+ clientAndStatus.setRight(false);
+ throw new PipeConnectionException(
+ String.format("Network error when transfer file %s, because %s.",
file, e.getMessage()),
+ e);
+ }
Review Comment:
i18n
--
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]
