BUG(duplication): When "duplicate_log_batch_bytes" is greater than 0 , there is a probability that some data will be lost.

[lengyuexuexuan]

Bug Report

1. What did you do?
   

   incubator-pegasus/src/server/pegasus_mutation_duplicator.cpp

   Lines 251 to 270 in e42f1a6

   	for (auto mut : muts) {
   	// mut: 0=timestamp, 1=rpc_code, 2=raw_message
   	batch_count++;
   	dsn::task_code rpc_code = std::get<1>(mut);
   	dsn::blob raw_message = std::get<2>(mut);
   	auto dreq = std::make_unique<dsn::apps::duplicate_request>();
   	if (gutil::ContainsKey(ingnored_rpc_code, rpc_code)) {
   	// It it do not recommend to use bulkload and normal writing in the same app,
   	// it may also cause inconsistency between actual data and expected data
   	// And duplication will not dup the data of bulkload to backup clusters,
   	// if you want to force use it, you can permit this risk in you own way on the clusters
   	// you maintenance. For example, you can do bulkload both on master-clusters and
   	// backup-cluster (with duplication enable) at the same time, but this will inevitably
   	// cause data inconsistency problems.
   	if (rpc_code == dsn::apps::RPC_RRDB_RRDB_BULK_LOAD) {
   	LOG_DEBUG_PREFIX("Ignore sending bulkload rpc when doing duplication");
   	}
   	continue;
   	}

• The "muts" contain multiple mutations.
• When the RPC code of the last mutation is "RPC_RRDB_RRDB_DUPLICATE" or "RPC_RRDB_RRDB_BULK_LOAD", will execute "continue", not push duplicate_rpc into "_inflights".
• But the last_decree has updated, the next time the logs are loaded, these mutations will not be read again.

[acelyc111]

@empiredan Please take a look