#9647

pending completion

Build # #9647

Build Type

push

travis_ci

Committed by

web-flow

Commit Message

[IOTDB-6075] Pipe: File resource races when different tsfile load operations concurrently modify the same tsfile at receiver (#10629)

* Add a parameter in iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties to control the connection timeout.
* Set the pipe connection timeout between sender and receiver to 15 mins to allow long time-cost load operation.
* Redesign the pipe receiver's dir to avoid file resource races when different tsfile load operations concurrently modify the same tsfile.

Run Details

184 of 184 new or added lines in 9 files covered. (100.0%)

79058 of 165585 relevant lines covered (47.74%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

14.72

/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/v1/IoTDBThriftReceiverV1.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.iotdb.db.pipe.connector.v1;

import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.receiver.IoTDBThriftReceiver;
import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorRequestVersion;
import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFilePieceReq;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFileSealReq;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferTabletReq;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.file.Files;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public class IoTDBThriftReceiverV1 implements IoTDBThriftReceiver {

  private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftReceiverV1.class);

  private static final IoTDBConfig IOTDB_CONFIG = IoTDBDescriptor.getInstance().getConfig();
  private static final String RECEIVER_FILE_BASE_DIR = IOTDB_CONFIG.getPipeReceiverFileDir();
  private final AtomicReference<File> receiverFileDirWithIdSuffix = new AtomicReference<>();

  // Used to generate transfer id, which is used to identify a receiver thread.
  private static final AtomicLong RECEIVER_ID_GENERATOR = new AtomicLong(0);
  private final AtomicLong receiverId = new AtomicLong(0);

  private File writingFile;
  private RandomAccessFile writingFileWriter;

  @Override
  public synchronized TPipeTransferResp receive(
      TPipeTransferReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
    final short rawRequestType = req.getType();
    if (PipeRequestType.isValidatedRequestType(rawRequestType)) {
      switch (PipeRequestType.valueOf(rawRequestType)) {
        case HANDSHAKE:
          return handleTransferHandshake(PipeTransferHandshakeReq.fromTPipeTransferReq(req));
        case TRANSFER_INSERT_NODE:
          return handleTransferInsertNode(
              PipeTransferInsertNodeReq.fromTPipeTransferReq(req), partitionFetcher, schemaFetcher);
        case TRANSFER_TABLET:
          return handleTransferTablet(
              PipeTransferTabletReq.fromTPipeTransferReq(req), partitionFetcher, schemaFetcher);
        case TRANSFER_FILE_PIECE:
          return handleTransferFilePiece(PipeTransferFilePieceReq.fromTPipeTransferReq(req));
        case TRANSFER_FILE_SEAL:
          return handleTransferFileSeal(
              PipeTransferFileSealReq.fromTPipeTransferReq(req), partitionFetcher, schemaFetcher);
        default:
          break;
      }
    }

    // unknown request type, which means the request can not be handled by this receiver,
    // maybe the version of the receiver is not compatible with the sender
    final TSStatus status =
        RpcUtils.getStatus(
            TSStatusCode.PIPE_TYPE_ERROR,
            String.format("Unknown PipeRequestType %s.", rawRequestType));
    LOGGER.warn("Unknown PipeRequestType, response status = {}.", status);
    return new TPipeTransferResp(status);
  }

  private TPipeTransferResp handleTransferHandshake(PipeTransferHandshakeReq req) {
    if (!CommonDescriptor.getInstance()
        .getConfig()
        .getTimestampPrecision()
        .equals(req.getTimestampPrecision())) {
      final TSStatus status =
          RpcUtils.getStatus(
              TSStatusCode.PIPE_HANDSHAKE_ERROR,
              String.format(
                  "IoTDB receiver's timestamp precision %s, "
                      + "connector's timestamp precision %s. Validation fails.",
                  CommonDescriptor.getInstance().getConfig().getTimestampPrecision(),
                  req.getTimestampPrecision()));
      LOGGER.warn("Handshake failed, response status = {}.", status);
      return new TPipeTransferResp(status);
    }

    receiverId.set(RECEIVER_ID_GENERATOR.incrementAndGet());

    // clear the original receiver file dir if exists
    if (receiverFileDirWithIdSuffix.get() != null) {
      if (receiverFileDirWithIdSuffix.get().exists()) {
        try {
          Files.delete(receiverFileDirWithIdSuffix.get().toPath());
          LOGGER.info(
              "Original receiver file dir {} was deleted.",
              receiverFileDirWithIdSuffix.get().getPath());
        } catch (IOException e) {
          LOGGER.warn(
              "Failed to delete original receiver file dir {}, because {}.",
              receiverFileDirWithIdSuffix.get().getPath(),
              e.getMessage());
        }
      } else {
        LOGGER.info(
            "Original receiver file dir {} is not existed. No need to delete.",
            receiverFileDirWithIdSuffix.get().getPath());
      }
      receiverFileDirWithIdSuffix.set(null);
    } else {
      LOGGER.info("Current receiver file dir is null. No need to delete.");
    }

    // create a new receiver file dir
    final File newReceiverDir = new File(RECEIVER_FILE_BASE_DIR, Long.toString(receiverId.get()));
    if (!newReceiverDir.exists()) {
      if (newReceiverDir.mkdirs()) {
        LOGGER.info("Receiver file dir {} was created.", newReceiverDir.getPath());
      } else {
        LOGGER.error("Failed to create receiver file dir {}.", newReceiverDir.getPath());
      }
    }
    receiverFileDirWithIdSuffix.set(newReceiverDir);

    LOGGER.info(
        "Handshake successfully, receiver id = {}, receiver file dir = {}.",
        receiverId.get(),
        newReceiverDir.getPath());
    return new TPipeTransferResp(RpcUtils.SUCCESS_STATUS);
  }

  private TPipeTransferResp handleTransferInsertNode(
      PipeTransferInsertNodeReq req,
      IPartitionFetcher partitionFetcher,
      ISchemaFetcher schemaFetcher) {
    return new TPipeTransferResp(
        executeStatement(req.constructStatement(), partitionFetcher, schemaFetcher));
  }

  private TPipeTransferResp handleTransferTablet(
      PipeTransferTabletReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
    InsertTabletStatement statement = req.constructStatement();
    return new TPipeTransferResp(
        statement.isEmpty()
            ? RpcUtils.SUCCESS_STATUS
            : executeStatement(statement, partitionFetcher, schemaFetcher));
  }

  private TPipeTransferResp handleTransferFilePiece(PipeTransferFilePieceReq req) {
    try {
      updateWritingFileIfNeeded(req.getFileName());

      if (!isWritingFileOffsetCorrect(req.getStartWritingOffset())) {
        final TSStatus status =
            RpcUtils.getStatus(
                TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET,
                String.format(
                    "Request sender to reset file reader's offset from %s to %s.",
                    req.getStartWritingOffset(), writingFileWriter.length()));
        LOGGER.warn("File offset reset requested by receiver, response status = {}.", status);
        return PipeTransferFilePieceResp.toTPipeTransferResp(status, writingFileWriter.length());
      }

      writingFileWriter.write(req.getFilePiece());
      return PipeTransferFilePieceResp.toTPipeTransferResp(
          RpcUtils.SUCCESS_STATUS, writingFileWriter.length());
    } catch (Exception e) {
      LOGGER.warn(String.format("Failed to write file piece from req %s.", req), e);
      final TSStatus status =
          RpcUtils.getStatus(
              TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
              String.format("Failed to write file piece, because %s", e.getMessage()));
      try {
        return PipeTransferFilePieceResp.toTPipeTransferResp(
            status, PipeTransferFilePieceResp.ERROR_END_OFFSET);
      } catch (IOException ex) {
        return PipeTransferFilePieceResp.toTPipeTransferResp(status);
      }
    }
  }

  private void updateWritingFileIfNeeded(String fileName) throws IOException {
    if (isFileExistedAndNameCorrect(fileName)) {
      return;
    }

    LOGGER.info(
        "Writing file {} is not existed or name is not correct, try to create it. "
            + "Current writing file is {}.",
        fileName,
        writingFile == null ? "null" : writingFile.getPath());

    closeCurrentWritingFileWriter();
    deleteCurrentWritingFile();

    // make sure receiver file dir exists
    // this may be useless, because receiver file dir is created when handshake. just in case.
    if (!receiverFileDirWithIdSuffix.get().exists()) {
      if (receiverFileDirWithIdSuffix.get().mkdirs()) {
        LOGGER.info(
            "Receiver file dir {} was created.", receiverFileDirWithIdSuffix.get().getPath());
      } else {
        LOGGER.error(
            "Failed to create receiver file dir {}.", receiverFileDirWithIdSuffix.get().getPath());
      }
    }

    writingFile = new File(receiverFileDirWithIdSuffix.get(), fileName);
    writingFileWriter = new RandomAccessFile(writingFile, "rw");
    LOGGER.info("Writing file {} was created. Ready to write file pieces.", writingFile.getPath());
  }

  private boolean isFileExistedAndNameCorrect(String fileName) {
    return writingFile != null && writingFile.getName().equals(fileName);
  }

  private void closeCurrentWritingFileWriter() {
    if (writingFileWriter != null) {
      try {
        writingFileWriter.close();
        LOGGER.info(
            "Current writing file writer {} was closed.",
            writingFile == null ? "null" : writingFile.getPath());
      } catch (IOException e) {
        LOGGER.warn(
            "Failed to close current writing file writer {}, because {}.",
            writingFile == null ? "null" : writingFile.getPath(),
            e.getMessage());
      }
      writingFileWriter = null;
    } else {
      LOGGER.info("Current writing file writer is null. No need to close.");
    }
  }

  private void deleteCurrentWritingFile() {
    if (writingFile != null) {
      if (writingFile.exists()) {
        try {
          Files.delete(writingFile.toPath());
          LOGGER.info("Original writing file {} was deleted.", writingFile.getPath());
        } catch (IOException e) {
          LOGGER.warn(
              "Failed to delete original writing file {}, because {}.",
              writingFile.getPath(),
              e.getMessage());
        }
      } else {
        LOGGER.info("Original file {} is not existed. No need to delete.", writingFile.getPath());
      }
      writingFile = null;
    } else {
      LOGGER.info("Current writing file is null. No need to delete.");
    }
  }

  private boolean isWritingFileOffsetCorrect(long offset) throws IOException {
    final boolean offsetCorrect = writingFileWriter.length() == offset;
    if (!offsetCorrect) {
      LOGGER.warn(
          "Writing file {}'s offset is {}, but request sender's offset is {}.",
          writingFile.getPath(),
          writingFileWriter.length(),
          offset);
    }
    return offsetCorrect;
  }

  private TPipeTransferResp handleTransferFileSeal(
      PipeTransferFileSealReq req,
      IPartitionFetcher partitionFetcher,
      ISchemaFetcher schemaFetcher) {
    try {
      if (!isWritingFileAvailable()) {
        final TSStatus status =
            RpcUtils.getStatus(
                TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
                String.format(
                    "Failed to seal file, because writing file %s is not available.",
                    req.getFileName()));
        LOGGER.warn(status.getMessage());
        return new TPipeTransferResp(status);
      }

      if (!isFileExistedAndNameCorrect(req.getFileName())) {
        final TSStatus status =
            RpcUtils.getStatus(
                TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
                String.format(
                    "Failed to seal file %s, but writing file is %s.",
                    req.getFileName(), writingFile));
        LOGGER.warn(status.getMessage());
        return new TPipeTransferResp(status);
      }

      if (!isWritingFileOffsetCorrect(req.getFileLength())) {
        final TSStatus status =
            RpcUtils.getStatus(
                TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
                String.format(
                    "Failed to seal file %s, because the length of file is not correct. "
                        + "The original file has length %s, but receiver file has length %s.",
                    req.getFileName(), req.getFileLength(), writingFileWriter.length()));
        LOGGER.warn(status.getMessage());
        return new TPipeTransferResp(status);
      }

      final String fileAbsolutePath = writingFile.getAbsolutePath();
      final LoadTsFileStatement statement = new LoadTsFileStatement(fileAbsolutePath);

      // 1. The writing file writer must be closed, otherwise it may cause concurrent errors during
      // the process of loading tsfile when parsing tsfile.
      //
      // 2. The writing file must be set to null, otherwise if the next passed tsfile has the same
      // name as the current tsfile, it will bypass the judgment logic of
      // updateWritingFileIfNeeded#isFileExistedAndNameCorrect, and continue to write to the already
      // loaded file. Since the writing file writer has already been closed, it will throw a Stream
      // Close exception.
      writingFileWriter.close();
      writingFileWriter = null;

      // writingFile will be deleted after load if no exception occurs
      writingFile = null;

      statement.setDeleteAfterLoad(true);
      statement.setVerifySchema(true);
      statement.setAutoCreateDatabase(false);

      final TSStatus status = executeStatement(statement, partitionFetcher, schemaFetcher);
      if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        LOGGER.info(
            "Seal file {} successfully. Receiver id is {}.", fileAbsolutePath, receiverId.get());
      } else {
        LOGGER.warn(
            "Failed to seal file {}, because {}. Receiver id is {}.",
            fileAbsolutePath,
            status.getMessage(),
            receiverId.get());
      }
      return new TPipeTransferResp(status);
    } catch (IOException e) {
      LOGGER.warn(
          String.format(
              "Failed to seal file %s from req %s. Receiver id is %d.",
              writingFile, req, receiverId.get()),
          e);
      return new TPipeTransferResp(
          RpcUtils.getStatus(
              TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
              String.format("Failed to seal file %s because %s", writingFile, e.getMessage())));
    } finally {
      // If the writing file is not sealed successfully, the writing file will be deleted.
      // All pieces of the writing file should be retransmitted by the sender.
      closeCurrentWritingFileWriter();
      deleteCurrentWritingFile();
    }
  }

  private boolean isWritingFileAvailable() {
    final boolean isWritingFileAvailable =
        writingFile != null && writingFile.exists() && writingFileWriter != null;
    if (!isWritingFileAvailable) {
      LOGGER.info(
          "Writing file {} is not available. Writing file is null: {}, writing file exists: {}, writing file writer is null: {}.",
          writingFile,
          writingFile == null,
          writingFile != null && writingFile.exists(),
          writingFileWriter == null);
    }
    return isWritingFileAvailable;
  }

  private TSStatus executeStatement(
      Statement statement, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
    if (statement == null) {
      return RpcUtils.getStatus(
          TSStatusCode.PIPE_TRANSFER_EXECUTE_STATEMENT_ERROR, "Execute null statement.");
    }

    final long queryId = SessionManager.getInstance().requestQueryId();
    final ExecutionResult result =
        Coordinator.getInstance()
            .execute(
                statement,
                queryId,
                null,
                "",
                partitionFetcher,
                schemaFetcher,
                IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold());
    if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
      LOGGER.warn(
          "failed to execute statement, statement: {}, result status is: {}",
          statement,
          result.status);
    }
    return result.status;
  }

  @Override
  public synchronized void handleExit() {
    if (writingFileWriter != null) {
      try {
        writingFileWriter.close();
        LOGGER.info("IoTDBThriftReceiverV1#handleExit: writing file writer was closed.");
      } catch (Exception e) {
        LOGGER.warn("IoTDBThriftReceiverV1#handleExit: close writing file writer error.", e);
      }
      writingFileWriter = null;
    } else {
      LOGGER.info(
          "IoTDBThriftReceiverV1#handleExit: writing file writer is null. No need to close.");
    }

    if (writingFile != null) {
      try {
        Files.delete(writingFile.toPath());
        LOGGER.info(
            "IoTDBThriftReceiverV1#handleExit: writing file {} was deleted.",
            writingFile.getPath());
      } catch (Exception e) {
        LOGGER.warn(
            "IoTDBThriftReceiverV1#handleExit: delete file {} error.", writingFile.getPath());
      }
      writingFile = null;
    } else {
      LOGGER.info("IoTDBThriftReceiverV1#handleExit: writing file is null. No need to delete.");
    }

    // clear the original receiver file dir if exists
    if (receiverFileDirWithIdSuffix.get() != null) {
      if (receiverFileDirWithIdSuffix.get().exists()) {
        try {
          Files.delete(receiverFileDirWithIdSuffix.get().toPath());
          LOGGER.info(
              "IoTDBThriftReceiverV1#handleExit: original receiver file dir {} was deleted.",
              receiverFileDirWithIdSuffix.get().getPath());
        } catch (IOException e) {
          LOGGER.warn(
              "IoTDBThriftReceiverV1#handleExit: delete original receiver file dir {} error.",
              receiverFileDirWithIdSuffix.get().getPath());
        }
      } else {
        LOGGER.info(
            "IoTDBThriftReceiverV1#handleExit: original receiver file dir {} does not exist. No need to delete.",
            receiverFileDirWithIdSuffix.get().getPath());
      }
      receiverFileDirWithIdSuffix.set(null);
    } else {
      LOGGER.info(
          "IoTDBThriftReceiverV1#handleExit: original receiver file dir is null. No need to delete.");
    }

    LOGGER.info("IoTDBThriftReceiverV1#handleExit: receiver exited.");
  }

  @Override
  public IoTDBThriftConnectorRequestVersion getVersion() {
    return IoTDBThriftConnectorRequestVersion.VERSION_1;
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one
3	* or more contributor license agreements. See the NOTICE file
4	* distributed with this work for additional information
5	* regarding copyright ownership. The ASF licenses this file
6	* to you under the Apache License, Version 2.0 (the
7	* "License"); you may not use this file except in compliance
8	* with the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing,
13	* software distributed under the License is distributed on an
14	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15	* KIND, either express or implied. See the License for the
16	* specific language governing permissions and limitations
17	* under the License.
18	*/
19
20	package org.apache.iotdb.db.pipe.connector.v1;
21
22	import org.apache.iotdb.common.rpc.thrift.TSStatus;
23	import org.apache.iotdb.commons.conf.CommonDescriptor;
24	import org.apache.iotdb.db.conf.IoTDBConfig;
25	import org.apache.iotdb.db.conf.IoTDBDescriptor;
26	import org.apache.iotdb.db.pipe.agent.receiver.IoTDBThriftReceiver;
27	import org.apache.iotdb.db.pipe.connector.IoTDBThriftConnectorRequestVersion;
28	import org.apache.iotdb.db.pipe.connector.v1.reponse.PipeTransferFilePieceResp;
29	import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFilePieceReq;
30	import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferFileSealReq;
31	import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferHandshakeReq;
32	import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferInsertNodeReq;
33	import org.apache.iotdb.db.pipe.connector.v1.request.PipeTransferTabletReq;
34	import org.apache.iotdb.db.protocol.session.SessionManager;
35	import org.apache.iotdb.db.queryengine.plan.Coordinator;
36	import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
37	import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
38	import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
39	import org.apache.iotdb.db.queryengine.plan.statement.Statement;
40	import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
41	import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
42	import org.apache.iotdb.rpc.RpcUtils;
43	import org.apache.iotdb.rpc.TSStatusCode;
44	import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
45	import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
46
47	import org.slf4j.Logger;
48	import org.slf4j.LoggerFactory;
49
50	import java.io.File;
51	import java.io.IOException;
52	import java.io.RandomAccessFile;
53	import java.nio.file.Files;
54	import java.util.concurrent.atomic.AtomicLong;
55	import java.util.concurrent.atomic.AtomicReference;
56
57	public class IoTDBThriftReceiverV1 implements IoTDBThriftReceiver {	1✔
58
59	private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBThriftReceiverV1.class);	1✔
60
61	private static final IoTDBConfig IOTDB_CONFIG = IoTDBDescriptor.getInstance().getConfig();	1✔
62	private static final String RECEIVER_FILE_BASE_DIR = IOTDB_CONFIG.getPipeReceiverFileDir();	1✔
63	private final AtomicReference<File> receiverFileDirWithIdSuffix = new AtomicReference<>();	1✔
64
65	// Used to generate transfer id, which is used to identify a receiver thread.
66	private static final AtomicLong RECEIVER_ID_GENERATOR = new AtomicLong(0);	1✔
67	private final AtomicLong receiverId = new AtomicLong(0);	1✔
68
69	private File writingFile;
70	private RandomAccessFile writingFileWriter;
71
72	@Override
73	public synchronized TPipeTransferResp receive(
74	TPipeTransferReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
75	final short rawRequestType = req.getType();	1✔
76	if (PipeRequestType.isValidatedRequestType(rawRequestType)) {	1✔
77	switch (PipeRequestType.valueOf(rawRequestType)) {	1✔
78	case HANDSHAKE:
79	return handleTransferHandshake(PipeTransferHandshakeReq.fromTPipeTransferReq(req));	1✔
80	case TRANSFER_INSERT_NODE:
81	return handleTransferInsertNode(	×
82	PipeTransferInsertNodeReq.fromTPipeTransferReq(req), partitionFetcher, schemaFetcher);	×
83	case TRANSFER_TABLET:
84	return handleTransferTablet(	1✔
85	PipeTransferTabletReq.fromTPipeTransferReq(req), partitionFetcher, schemaFetcher);	1✔
86	case TRANSFER_FILE_PIECE:
87	return handleTransferFilePiece(PipeTransferFilePieceReq.fromTPipeTransferReq(req));	×
88	case TRANSFER_FILE_SEAL:
89	return handleTransferFileSeal(	×
90	PipeTransferFileSealReq.fromTPipeTransferReq(req), partitionFetcher, schemaFetcher);	×
91	default:
92	break;
93	}
94	}
95
96	// unknown request type, which means the request can not be handled by this receiver,
97	// maybe the version of the receiver is not compatible with the sender
98	final TSStatus status =	×
99	RpcUtils.getStatus(	×
100	TSStatusCode.PIPE_TYPE_ERROR,
101	String.format("Unknown PipeRequestType %s.", rawRequestType));	×
102	LOGGER.warn("Unknown PipeRequestType, response status = {}.", status);	×
103	return new TPipeTransferResp(status);	×
104	}
105
106	private TPipeTransferResp handleTransferHandshake(PipeTransferHandshakeReq req) {
107	if (!CommonDescriptor.getInstance()	1✔
108	.getConfig()	1✔
109	.getTimestampPrecision()	1✔
110	.equals(req.getTimestampPrecision())) {	1✔
111	final TSStatus status =	×
112	RpcUtils.getStatus(	×
113	TSStatusCode.PIPE_HANDSHAKE_ERROR,
114	String.format(	×
115	"IoTDB receiver's timestamp precision %s, "
116	+ "connector's timestamp precision %s. Validation fails.",
117	CommonDescriptor.getInstance().getConfig().getTimestampPrecision(),	×
118	req.getTimestampPrecision()));	×
119	LOGGER.warn("Handshake failed, response status = {}.", status);	×
120	return new TPipeTransferResp(status);	×
121	}
122
123	receiverId.set(RECEIVER_ID_GENERATOR.incrementAndGet());	1✔
124
125	// clear the original receiver file dir if exists
126	if (receiverFileDirWithIdSuffix.get() != null) {	1✔
127	if (receiverFileDirWithIdSuffix.get().exists()) {	×
128	try {
129	Files.delete(receiverFileDirWithIdSuffix.get().toPath());	×
130	LOGGER.info(	×
131	"Original receiver file dir {} was deleted.",
132	receiverFileDirWithIdSuffix.get().getPath());	×
133	} catch (IOException e) {	×
134	LOGGER.warn(	×
135	"Failed to delete original receiver file dir {}, because {}.",
136	receiverFileDirWithIdSuffix.get().getPath(),	×
137	e.getMessage());	×
138	}	×
139	} else {
140	LOGGER.info(	×
141	"Original receiver file dir {} is not existed. No need to delete.",
142	receiverFileDirWithIdSuffix.get().getPath());	×
143	}
144	receiverFileDirWithIdSuffix.set(null);	×
145	} else {
146	LOGGER.info("Current receiver file dir is null. No need to delete.");	1✔
147	}
148
149	// create a new receiver file dir
150	final File newReceiverDir = new File(RECEIVER_FILE_BASE_DIR, Long.toString(receiverId.get()));	1✔
151	if (!newReceiverDir.exists()) {	1✔
152	if (newReceiverDir.mkdirs()) {	1✔
153	LOGGER.info("Receiver file dir {} was created.", newReceiverDir.getPath());	1✔
154	} else {
155	LOGGER.error("Failed to create receiver file dir {}.", newReceiverDir.getPath());	×
156	}
157	}
158	receiverFileDirWithIdSuffix.set(newReceiverDir);	1✔
159
160	LOGGER.info(	1✔
161	"Handshake successfully, receiver id = {}, receiver file dir = {}.",
162	receiverId.get(),	1✔
163	newReceiverDir.getPath());	1✔
164	return new TPipeTransferResp(RpcUtils.SUCCESS_STATUS);	1✔
165	}
166
167	private TPipeTransferResp handleTransferInsertNode(
168	PipeTransferInsertNodeReq req,
169	IPartitionFetcher partitionFetcher,
170	ISchemaFetcher schemaFetcher) {
171	return new TPipeTransferResp(	×
172	executeStatement(req.constructStatement(), partitionFetcher, schemaFetcher));	×
173	}
174
175	private TPipeTransferResp handleTransferTablet(
176	PipeTransferTabletReq req, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
177	InsertTabletStatement statement = req.constructStatement();	1✔
178	return new TPipeTransferResp(	1✔
179	statement.isEmpty()	1✔
180	? RpcUtils.SUCCESS_STATUS	1✔
181	: executeStatement(statement, partitionFetcher, schemaFetcher));	1✔
182	}
183
184	private TPipeTransferResp handleTransferFilePiece(PipeTransferFilePieceReq req) {
185	try {
186	updateWritingFileIfNeeded(req.getFileName());	×
187
188	if (!isWritingFileOffsetCorrect(req.getStartWritingOffset())) {	×
189	final TSStatus status =	×
190	RpcUtils.getStatus(	×
191	TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET,
192	String.format(	×
193	"Request sender to reset file reader's offset from %s to %s.",
194	req.getStartWritingOffset(), writingFileWriter.length()));	×
195	LOGGER.warn("File offset reset requested by receiver, response status = {}.", status);	×
196	return PipeTransferFilePieceResp.toTPipeTransferResp(status, writingFileWriter.length());	×
197	}
198
199	writingFileWriter.write(req.getFilePiece());	×
200	return PipeTransferFilePieceResp.toTPipeTransferResp(	×
201	RpcUtils.SUCCESS_STATUS, writingFileWriter.length());	×
202	} catch (Exception e) {	×
203	LOGGER.warn(String.format("Failed to write file piece from req %s.", req), e);	×
204	final TSStatus status =	×
205	RpcUtils.getStatus(	×
206	TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
207	String.format("Failed to write file piece, because %s", e.getMessage()));	×
208	try {
209	return PipeTransferFilePieceResp.toTPipeTransferResp(	×
210	status, PipeTransferFilePieceResp.ERROR_END_OFFSET);
211	} catch (IOException ex) {	×
212	return PipeTransferFilePieceResp.toTPipeTransferResp(status);	×
213	}
214	}
215	}
216
217	private void updateWritingFileIfNeeded(String fileName) throws IOException {
218	if (isFileExistedAndNameCorrect(fileName)) {	×
219	return;	×
220	}
221
222	LOGGER.info(	×
223	"Writing file {} is not existed or name is not correct, try to create it. "
224	+ "Current writing file is {}.",
225	fileName,
226	writingFile == null ? "null" : writingFile.getPath());	×
227
228	closeCurrentWritingFileWriter();	×
229	deleteCurrentWritingFile();	×
230
231	// make sure receiver file dir exists
232	// this may be useless, because receiver file dir is created when handshake. just in case.
233	if (!receiverFileDirWithIdSuffix.get().exists()) {	×
234	if (receiverFileDirWithIdSuffix.get().mkdirs()) {	×
235	LOGGER.info(	×
236	"Receiver file dir {} was created.", receiverFileDirWithIdSuffix.get().getPath());	×
237	} else {
238	LOGGER.error(	×
239	"Failed to create receiver file dir {}.", receiverFileDirWithIdSuffix.get().getPath());	×
240	}
241	}
242
243	writingFile = new File(receiverFileDirWithIdSuffix.get(), fileName);	×
244	writingFileWriter = new RandomAccessFile(writingFile, "rw");	×
245	LOGGER.info("Writing file {} was created. Ready to write file pieces.", writingFile.getPath());	×
246	}	×
247
248	private boolean isFileExistedAndNameCorrect(String fileName) {
249	return writingFile != null && writingFile.getName().equals(fileName);	×
250	}
251
252	private void closeCurrentWritingFileWriter() {
253	if (writingFileWriter != null) {	×
254	try {
255	writingFileWriter.close();	×
256	LOGGER.info(	×
257	"Current writing file writer {} was closed.",
258	writingFile == null ? "null" : writingFile.getPath());	×
259	} catch (IOException e) {	×
260	LOGGER.warn(	×
261	"Failed to close current writing file writer {}, because {}.",
262	writingFile == null ? "null" : writingFile.getPath(),	×
263	e.getMessage());	×
264	}	×
265	writingFileWriter = null;	×
266	} else {
267	LOGGER.info("Current writing file writer is null. No need to close.");	×
268	}
269	}	×
270
271	private void deleteCurrentWritingFile() {
272	if (writingFile != null) {	×
273	if (writingFile.exists()) {	×
274	try {
275	Files.delete(writingFile.toPath());	×
276	LOGGER.info("Original writing file {} was deleted.", writingFile.getPath());	×
277	} catch (IOException e) {	×
278	LOGGER.warn(	×
279	"Failed to delete original writing file {}, because {}.",
280	writingFile.getPath(),	×
281	e.getMessage());	×
282	}	×
283	} else {
284	LOGGER.info("Original file {} is not existed. No need to delete.", writingFile.getPath());	×
285	}
286	writingFile = null;	×
287	} else {
288	LOGGER.info("Current writing file is null. No need to delete.");	×
289	}
290	}	×
291
292	private boolean isWritingFileOffsetCorrect(long offset) throws IOException {
293	final boolean offsetCorrect = writingFileWriter.length() == offset;	×
294	if (!offsetCorrect) {	×
295	LOGGER.warn(	×
296	"Writing file {}'s offset is {}, but request sender's offset is {}.",
297	writingFile.getPath(),	×
298	writingFileWriter.length(),	×
299	offset);	×
300	}
301	return offsetCorrect;	×
302	}
303
304	private TPipeTransferResp handleTransferFileSeal(
305	PipeTransferFileSealReq req,
306	IPartitionFetcher partitionFetcher,
307	ISchemaFetcher schemaFetcher) {
308	try {
309	if (!isWritingFileAvailable()) {	×
310	final TSStatus status =	×
311	RpcUtils.getStatus(	×
312	TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
313	String.format(	×
314	"Failed to seal file, because writing file %s is not available.",
315	req.getFileName()));	×
316	LOGGER.warn(status.getMessage());	×
317	return new TPipeTransferResp(status);	×
318	}
319
320	if (!isFileExistedAndNameCorrect(req.getFileName())) {	×
321	final TSStatus status =	×
322	RpcUtils.getStatus(	×
323	TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
324	String.format(	×
325	"Failed to seal file %s, but writing file is %s.",
326	req.getFileName(), writingFile));	×
327	LOGGER.warn(status.getMessage());	×
328	return new TPipeTransferResp(status);	×
329	}
330
331	if (!isWritingFileOffsetCorrect(req.getFileLength())) {	×
332	final TSStatus status =	×
333	RpcUtils.getStatus(	×
334	TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
335	String.format(	×
336	"Failed to seal file %s, because the length of file is not correct. "
337	+ "The original file has length %s, but receiver file has length %s.",
338	req.getFileName(), req.getFileLength(), writingFileWriter.length()));	×
339	LOGGER.warn(status.getMessage());	×
340	return new TPipeTransferResp(status);	×
341	}
342
343	final String fileAbsolutePath = writingFile.getAbsolutePath();	×
344	final LoadTsFileStatement statement = new LoadTsFileStatement(fileAbsolutePath);	×
345
346	// 1. The writing file writer must be closed, otherwise it may cause concurrent errors during
347	// the process of loading tsfile when parsing tsfile.
348	//
349	// 2. The writing file must be set to null, otherwise if the next passed tsfile has the same
350	// name as the current tsfile, it will bypass the judgment logic of
351	// updateWritingFileIfNeeded#isFileExistedAndNameCorrect, and continue to write to the already
352	// loaded file. Since the writing file writer has already been closed, it will throw a Stream
353	// Close exception.
354	writingFileWriter.close();	×
355	writingFileWriter = null;	×
356
357	// writingFile will be deleted after load if no exception occurs
358	writingFile = null;	×
359
360	statement.setDeleteAfterLoad(true);	×
361	statement.setVerifySchema(true);	×
362	statement.setAutoCreateDatabase(false);	×
363
364	final TSStatus status = executeStatement(statement, partitionFetcher, schemaFetcher);	×
365	if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {	×
366	LOGGER.info(	×
367	"Seal file {} successfully. Receiver id is {}.", fileAbsolutePath, receiverId.get());	×
368	} else {
369	LOGGER.warn(	×
370	"Failed to seal file {}, because {}. Receiver id is {}.",
371	fileAbsolutePath,
372	status.getMessage(),	×
373	receiverId.get());	×
374	}
375	return new TPipeTransferResp(status);	×
376	} catch (IOException e) {	×
377	LOGGER.warn(	×
378	String.format(	×
379	"Failed to seal file %s from req %s. Receiver id is %d.",
380	writingFile, req, receiverId.get()),	×
381	e);
382	return new TPipeTransferResp(	×
383	RpcUtils.getStatus(	×
384	TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
385	String.format("Failed to seal file %s because %s", writingFile, e.getMessage())));	×
386	} finally {
387	// If the writing file is not sealed successfully, the writing file will be deleted.
388	// All pieces of the writing file should be retransmitted by the sender.
389	closeCurrentWritingFileWriter();	×
390	deleteCurrentWritingFile();	×
391	}
392	}
393
394	private boolean isWritingFileAvailable() {
395	final boolean isWritingFileAvailable =	×
396	writingFile != null && writingFile.exists() && writingFileWriter != null;	×
397	if (!isWritingFileAvailable) {	×
398	LOGGER.info(	×
399	"Writing file {} is not available. Writing file is null: {}, writing file exists: {}, writing file writer is null: {}.",
400	writingFile,
401	writingFile == null,	×
402	writingFile != null && writingFile.exists(),	×
403	writingFileWriter == null);	×
404	}
405	return isWritingFileAvailable;	×
406	}
407
408	private TSStatus executeStatement(
409	Statement statement, IPartitionFetcher partitionFetcher, ISchemaFetcher schemaFetcher) {
410	if (statement == null) {	×
411	return RpcUtils.getStatus(	×
412	TSStatusCode.PIPE_TRANSFER_EXECUTE_STATEMENT_ERROR, "Execute null statement.");
413	}
414
415	final long queryId = SessionManager.getInstance().requestQueryId();	×
416	final ExecutionResult result =
417	Coordinator.getInstance()	×
418	.execute(	×
419	statement,
420	queryId,
421	null,
422	"",
423	partitionFetcher,
424	schemaFetcher,
425	IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold());	×
426	if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {	×
427	LOGGER.warn(	×
428	"failed to execute statement, statement: {}, result status is: {}",
429	statement,
430	result.status);
431	}
432	return result.status;	×
433	}
434
435	@Override
436	public synchronized void handleExit() {
437	if (writingFileWriter != null) {	×
438	try {
439	writingFileWriter.close();	×
440	LOGGER.info("IoTDBThriftReceiverV1#handleExit: writing file writer was closed.");	×
441	} catch (Exception e) {	×
442	LOGGER.warn("IoTDBThriftReceiverV1#handleExit: close writing file writer error.", e);	×
443	}	×
444	writingFileWriter = null;	×
445	} else {
446	LOGGER.info(	×
447	"IoTDBThriftReceiverV1#handleExit: writing file writer is null. No need to close.");
448	}
449
450	if (writingFile != null) {	×
451	try {
452	Files.delete(writingFile.toPath());	×
453	LOGGER.info(	×
454	"IoTDBThriftReceiverV1#handleExit: writing file {} was deleted.",
455	writingFile.getPath());	×
456	} catch (Exception e) {	×
457	LOGGER.warn(	×
458	"IoTDBThriftReceiverV1#handleExit: delete file {} error.", writingFile.getPath());	×
459	}	×
460	writingFile = null;	×
461	} else {
462	LOGGER.info("IoTDBThriftReceiverV1#handleExit: writing file is null. No need to delete.");	×
463	}
464
465	// clear the original receiver file dir if exists
466	if (receiverFileDirWithIdSuffix.get() != null) {	×
467	if (receiverFileDirWithIdSuffix.get().exists()) {	×
468	try {
469	Files.delete(receiverFileDirWithIdSuffix.get().toPath());	×
470	LOGGER.info(	×
471	"IoTDBThriftReceiverV1#handleExit: original receiver file dir {} was deleted.",
472	receiverFileDirWithIdSuffix.get().getPath());	×
473	} catch (IOException e) {	×
474	LOGGER.warn(	×
475	"IoTDBThriftReceiverV1#handleExit: delete original receiver file dir {} error.",
476	receiverFileDirWithIdSuffix.get().getPath());	×
477	}	×
478	} else {
479	LOGGER.info(	×
480	"IoTDBThriftReceiverV1#handleExit: original receiver file dir {} does not exist. No need to delete.",
481	receiverFileDirWithIdSuffix.get().getPath());	×
482	}
483	receiverFileDirWithIdSuffix.set(null);	×
484	} else {
485	LOGGER.info(	×
486	"IoTDBThriftReceiverV1#handleExit: original receiver file dir is null. No need to delete.");
487	}
488
489	LOGGER.info("IoTDBThriftReceiverV1#handleExit: receiver exited.");	×
490	}	×
491
492	@Override
493	public IoTDBThriftConnectorRequestVersion getVersion() {
494	return IoTDBThriftConnectorRequestVersion.VERSION_1;	×
495	}
496	}

apache / iotdb / #9647

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous