Fixed a bug in EventListener::OnCompactionCompleted().

Summary:
Fixed a bug in EventListener::OnCompactionCompleted() that returns
incorrect list of input / output file names.

Test Plan: Extend existing test in listener_test.cc

Reviewers: sdong, rven, igor

Reviewed By: igor

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D38349

db/column_family.cc

@@ -630,18 +630,21 @@ void ColumnFamilyData::NotifyOnCompactionCompleted(
     DB* db, Compaction* c, const Status& status) {
 #ifndef ROCKSDB_LITE
   auto listeners = ioptions()->listeners;
+  ASSERT_GT(listeners.size(), 0U);
   CompactionJobInfo info;
   info.cf_name = c->column_family_data()->GetName();
   info.status = status;
   info.output_level = c->output_level();
-  for (const auto fmd : *c->inputs(c->level())) {
-    info.input_files.push_back(
-        TableFileName(options_.db_paths,
-                      fmd->fd.GetNumber(),
-                      fmd->fd.GetPathId()));
+  for (size_t i = 0; i < c->num_input_levels(); ++i) {
+    for (const auto fmd : *c->inputs(i)) {
+      info.input_files.push_back(
+          TableFileName(options_.db_paths,
+                        fmd->fd.GetNumber(),
+                        fmd->fd.GetPathId()));
+    }
   }
   for (const auto newf : c->edit()->GetNewFiles()) {
-    info.input_files.push_back(
+    info.output_files.push_back(
         TableFileName(options_.db_paths,
                       newf.second.fd.GetNumber(),
                       newf.second.fd.GetPathId()));

db/listener_test.cc

@@ -152,10 +152,14 @@ class EventListenerTest : public testing::Test {
 class TestCompactionListener : public EventListener {
  public:
   void OnCompactionCompleted(DB *db, const CompactionJobInfo& ci) override {
+    std::lock_guard<std::mutex> lock(mutex_);
     compacted_dbs_.push_back(db);
+    ASSERT_GT(ci.input_files.size(), 0U);
+    ASSERT_GT(ci.output_files.size(), 0U);
   }
 
   std::vector<DB*> compacted_dbs_;
+  std::mutex mutex_;
 };
 
 TEST_F(EventListenerTest, OnSingleDBCompactionTest) {

include/rocksdb/listener.h

@@ -45,6 +45,18 @@ struct CompactionJobInfo {
 // actual thread that involves in that specific event.   For example, it
 // is the RocksDB background flush thread that does the actual flush to
 // call EventListener::OnFlushCompleted().
+//
+// [Locking] All EventListener callbacks are designed to be called without
+// the current thread holding any DB mutex. This is to prevent potential
+// deadlock and performance issue when using EventListener callback
+// in a complex way. However, all EventListener call-back functions
+// should not run for an extended period of time before the function
+// returns, otherwise RocksDB may be blocked. For example, it is not
+// suggested to do DB::CompactFiles() (as it may run for a long while)
+// or issue many of DB::Put() (as Put may be blocked in certain cases)
+// in the same thread in the EventListener callback. However, doing
+// DB::CompactFiles() and DB::Put() in a thread other than the
+// EventListener callback thread is considered safe.
 class EventListener {
  public:
   // A call-back function to RocksDB which will be called whenever a