Caideyipi commented on code in PR #18051:
URL: https://github.com/apache/iotdb/pull/18051#discussion_r3526260240
##########
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:
Applied in 1370353caf5: reused the existing transfer-file network error i18n
constant and added en/zh i18n for the transfer file status message.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
Review Comment:
Applied in 1370353caf5: moved the pipe meta report logs to
DataNodePipeMessages with en/zh i18n constants.
--
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]
