[Backport to the 2.0.5 branch] [#3665] Evict log cache at follower af…

…ter operation was appended to log

Summary:
After 38ef51a, we ended up not calling ResponseFromPeer
for followers. At least one of the side effects of this was that the LogCache would be filled up
when getting requests, but no longer cleared after writing them to the Log. This quickly lead to
memory issues on lower memory machines, as we assign up to 1GB of memory across all LogCache
instances.

This diff addresses the issue, by cleaning LogCache after the operation is appended to the Log.

Test Plan:
ybd --gtest_filter QLStressTest.LongRemoteBootstrap
Jenkins: auto rebase: no

Reviewers: timur, mikhail, bogdan

Reviewed By: bogdan

Subscribers: ybase

Differential Revision: https://phabricator.dev.yugabyte.com/D8312

src/yb/client/ql-stress-test.cc

@@ -930,6 +930,21 @@ TEST_F_EX(QLStressTest, LongRemoteBootstrap, QLStressTestLongRemoteBootstrap) {
   ASSERT_OK(WaitAllReplicasHaveIndex(cluster_.get(), key.load(std::memory_order_acquire), 40s));
   LOG(INFO) << "All replicas ready";
 
+  ASSERT_OK(WaitFor([this] {
+    bool result = true;
+    auto followers = ListTabletPeers(cluster_.get(), ListPeersFilter::kNonLeaders);
+    LOG(INFO) << "Num followers: " << followers.size();
+    for (const auto& peer : followers) {
+      auto log_cache_size = peer->raft_consensus()->LogCacheSize();
+      LOG(INFO) << "T " << peer->tablet_id() << " P " << peer->permanent_uuid()
+                << ", log cache size: " << log_cache_size;
+      if (log_cache_size != 0) {
+        result = false;
+      }
+    }
+    return result;
+  }, 5s, "All followers cleanup cache"));
+
   // Write some more values and check that replica still in touch.
   std::this_thread::sleep_for(5s);
   ASSERT_OK(WaitAllReplicasHaveIndex(cluster_.get(), key.load(std::memory_order_acquire), 1s));

src/yb/consensus/consensus_queue.cc

@@ -324,6 +324,9 @@ void PeerMessageQueue::LocalPeerAppendFinished(const OpId& id,
     fake_response.mutable_status()->set_last_committed_idx(queue_state_.committed_index.index());
 
     if (queue_state_.mode != Mode::LEADER) {
+      log_cache_.EvictThroughOp(id.index());
+
+      UpdateMetrics();
       return;
     }
   }