add fault injection

kaijchen · Mar 19, 2024 · 7548dcb · 7548dcb
1 parent a4c2718
commit 7548dcb
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 3 deletions.
diff --git a/be/src/vec/sink/load_stream_stub.h b/be/src/vec/sink/load_stream_stub.h
@@ -194,16 +194,17 @@ class LoadStreamStub {
 
  friend std::ostream& operator<<(std::ostream& ostr, const LoadStreamStub& stub);
 
-#ifdef BE_TEST
+ // for tests only
  inline void add_success_tablet(int64_t tablet_id) {
  std::lock_guard<bthread::Mutex> lock(_success_tablets_mutex);
  _success_tablets.push_back(tablet_id);
  }
+
+ // for tests only
  inline void add_failed_tablet(int64_t tablet_id, Status reason) {
  std::lock_guard<bthread::Mutex> lock(_failed_tablets_mutex);
  _failed_tablets[tablet_id] = reason;
  }
-#endif
 
 private:
  Status _encode_and_send(PStreamHeader& header, std::span<const Slice> data = {});

diff --git a/be/src/vec/sink/writer/vtablet_writer_v2.cpp b/be/src/vec/sink/writer/vtablet_writer_v2.cpp
@@ -566,6 +566,18 @@ Status VTabletWriterV2::close(Status exec_status) {
  }
 
  std::vector<TTabletCommitInfo> tablet_commit_infos;
+ DBUG_EXECUTE_IF("VTabletWriterV2.close.add_failed_tablet", {
+ auto node = _streams_for_node.begin()->second;
+ int64_t tablet_id;
+ for (auto& stream : node->streams()) {
+ const auto& tablets = stream->success_tablets();
+ if (tablets.size() > 0) {
+ tablet_id = tablets[0];
+ break;
+ }
+ }
+ node->streams()[0]->add_failed_tablet(tablet_id, Status::InternalError("fault injection"));
+ });
  RETURN_IF_ERROR(_create_commit_info(tablet_commit_infos, _streams_for_node, _num_replicas));
  _state->tablet_commit_infos().insert(_state->tablet_commit_infos().end(),
  std::make_move_iterator(tablet_commit_infos.begin()),
@@ -607,6 +619,8 @@ Status VTabletWriterV2::_create_commit_info(
  known_tablets.insert(tablet_id);
  failed_tablets[tablet_id]++;
  failed_reason[tablet_id] = reason;
+ LOG(INFO) << "tablet " << tablet_id << " failed on backend " << node_id << ": "
+ << reason;
  }
  for (auto tablet_id : stream->success_tablets()) {
  if (known_tablets.contains(tablet_id)) {
@@ -622,7 +636,8 @@ Status VTabletWriterV2::_create_commit_info(
  }
  for (auto [tablet_id, replicas] : failed_tablets) {
  if (replicas > (num_replicas - 1) / 2) {
- LOG(INFO) << "tablet " << tablet_id << " failed, reason: " << failed_reason[tablet_id];
+ LOG(INFO) << "tablet " << tablet_id
+ << " failed on majority backends: " << failed_reason[tablet_id];
  return failed_reason[tablet_id];
  }
  }