Should flush and sync WAL when writing it in DB::Open() (#6417)

Summary:
A recent fix related to 2pc https://github.com/facebook/rocksdb/pull/6313/ writes something to WAL, but does not flush or sync. This causes assertion failure "impl->TEST_WALBufferIsEmpty()" if manual_wal_flush = true. We should fsync the entry to make sure a second power reset can recover.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6417

Test Plan: Add manual_wal_flush=true case in TransactionTest.DoubleCrashInRecovery and fix a bug in the test so that the bug can be reproduced. It passes with the fix.

Differential Revision: D19894537

fbshipit-source-id: f1e84e49e2269f583c6019743118292cd8b6598e

db/db_impl/db_impl_open.cc

@@ -1494,6 +1494,13 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
         uint64_t log_used, log_size;
         log::Writer* log_writer = impl->logs_.back().writer;
         s = impl->WriteToWAL(empty_batch, log_writer, &log_used, &log_size);
+        if (s.ok()) {
+          // Need to fsync, otherwise it might get lost after a power reset.
+          s = impl->FlushWAL(false);
+          if (s.ok()) {
+            s = log_writer->file()->Sync(impl->immutable_db_options_.use_fsync);
+          }
+        }
       }
     }
   }

utilities/transactions/transaction_test.cc

@@ -6120,8 +6120,10 @@ TEST_P(TransactionTest, ReseekOptimization) {
 // there. The new log files should be still read succesfully during recovery of
 // the 2nd crash.
 TEST_P(TransactionTest, DoubleCrashInRecovery) {
+  for (const bool manual_wal_flush : {false, true}) {
     for (const bool write_after_recovery : {false, true}) {
       options.wal_recovery_mode = WALRecoveryMode::kPointInTimeRecovery;
+      options.manual_wal_flush = manual_wal_flush;
       ReOpen();
       std::string cf_name = "two";
       ColumnFamilyOptions cf_options;
@@ -6171,7 +6173,7 @@ TEST_P(TransactionTest, DoubleCrashInRecovery) {
       file_content[400] = 'h';
       file_content[401] = 'a';
       ASSERT_OK(env->DeleteFile(fname));
-    ASSERT_OK(WriteStringToFile(env, file_content, fname));
+      ASSERT_OK(WriteStringToFile(env, file_content, fname, true));
 
       // Recover from corruption
       std::vector<ColumnFamilyHandle*> handles;
@@ -6200,6 +6202,7 @@ TEST_P(TransactionTest, DoubleCrashInRecovery) {
         delete handle;
       }
     }
+  }
 }
 
 }  // namespace rocksdb