diff --git a/src/common/options.cc b/src/common/options.cc
index 2b02dbfa907d8..f8f32119878c6 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -7304,12 +7304,18 @@ static std::vector<Option> get_rbd_options() {
     .set_default(5)
     .set_description("commit time interval, seconds"),
 
+    Option("rbd_journal_object_writethrough_until_flush", Option::TYPE_BOOL, Option::LEVEL_ADVANCED)
+    .set_default(true)
+    .set_description("when enabled, the rbd_journal_object_flush* configuration "
+                     "options are ignored until the first flush so that batched "
+                     "journal IO is known to be safe for consistency"),
+
     Option("rbd_journal_object_flush_interval", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
     .set_default(0)
     .set_description("maximum number of pending commits per journal object"),
 
     Option("rbd_journal_object_flush_bytes", Option::TYPE_SIZE, Option::LEVEL_ADVANCED)
-    .set_default(0)
+    .set_default(1_M)
     .set_description("maximum number of pending bytes per journal object"),
 
     Option("rbd_journal_object_flush_age", Option::TYPE_FLOAT, Option::LEVEL_ADVANCED)
diff --git a/src/journal/JournalRecorder.cc b/src/journal/JournalRecorder.cc
index d4b023c3d5eb2..aa90660a01fa2 100644
--- a/src/journal/JournalRecorder.cc
+++ b/src/journal/JournalRecorder.cc
@@ -10,7 +10,8 @@
 
 #define dout_subsys ceph_subsys_journaler
 #undef dout_prefix
-#define dout_prefix *_dout << "JournalRecorder: " << this << " "
+#define dout_prefix *_dout << "JournalRecorder: " << this << " " << __func__ \
+                           << ": "
 
 using std::shared_ptr;
 
@@ -49,12 +50,9 @@ struct C_Flush : public Context {
 JournalRecorder::JournalRecorder(librados::IoCtx &ioctx,
                                  const std::string &object_oid_prefix,
                                  const JournalMetadataPtr& journal_metadata,
-                                 uint32_t flush_interval, uint64_t flush_bytes,
-                                 double flush_age,
                                  uint64_t max_in_flight_appends)
   : m_cct(NULL), m_object_oid_prefix(object_oid_prefix),
-    m_journal_metadata(journal_metadata), m_flush_interval(flush_interval),
-    m_flush_bytes(flush_bytes), m_flush_age(flush_age),
+    m_journal_metadata(journal_metadata),
     m_max_in_flight_appends(max_in_flight_appends), m_listener(this),
     m_object_handler(this), m_lock("JournalerRecorder::m_lock"),
     m_current_set(m_journal_metadata->get_active_set()) {
@@ -65,13 +63,14 @@ JournalRecorder::JournalRecorder(librados::IoCtx &ioctx,
 
   uint8_t splay_width = m_journal_metadata->get_splay_width();
   for (uint8_t splay_offset = 0; splay_offset < splay_width; ++splay_offset) {
-    m_object_locks.push_back(shared_ptr<Mutex>(
-                                          new Mutex("ObjectRecorder::m_lock::"+
-                                          std::to_string(splay_offset))));
+    shared_ptr<Mutex> object_lock(new Mutex(
+      "ObjectRecorder::m_lock::" + std::to_string(splay_offset)));
+    m_object_locks.push_back(object_lock);
+
     uint64_t object_number = splay_offset + (m_current_set * splay_width);
+    Mutex::Locker locker(*object_lock);
     m_object_ptrs[splay_offset] = create_object_recorder(
-                                                object_number,
-                                                m_object_locks[splay_offset]);
+      object_number, m_object_locks[splay_offset]);
   }
 
   m_journal_metadata->add_listener(&m_listener);
@@ -108,8 +107,30 @@ void JournalRecorder::shut_down(Context *on_safe) {
   flush(on_safe);
 }
 
+void JournalRecorder::set_append_batch_options(int flush_interval,
+                                               uint64_t flush_bytes,
+                                               double flush_age) {
+  ldout(m_cct, 5) << "flush_interval=" << flush_interval << ", "
+                  << "flush_bytes=" << flush_bytes << ", "
+                  << "flush_age=" << flush_age << dendl;
+
+  Mutex::Locker locker(m_lock);
+  m_flush_interval = flush_interval;
+  m_flush_bytes = flush_bytes;
+  m_flush_age = flush_age;
+
+  uint8_t splay_width = m_journal_metadata->get_splay_width();
+  for (uint8_t splay_offset = 0; splay_offset < splay_width; ++splay_offset) {
+    Mutex::Locker object_locker(*m_object_locks[splay_offset]);
+    auto object_recorder = get_object(splay_offset);
+    object_recorder->set_append_batch_options(flush_interval, flush_bytes,
+                                              flush_age);
+  }
+}
+
 Future JournalRecorder::append(uint64_t tag_tid,
                                const bufferlist &payload_bl) {
+  ldout(m_cct, 20) << "tag_tid=" << tag_tid << dendl;
 
   m_lock.Lock();
 
@@ -132,7 +153,9 @@ Future JournalRecorder::append(uint64_t tag_tid,
 	 entry_bl);
   ceph_assert(entry_bl.length() <= m_journal_metadata->get_object_size());
 
-  bool object_full = object_ptr->append_unlock({{future, entry_bl}});
+  bool object_full = object_ptr->append({{future, entry_bl}});
+  m_object_locks[splay_offset]->Unlock();
+
   if (object_full) {
     ldout(m_cct, 10) << "object " << object_ptr->get_oid() << " now full"
                      << dendl;
@@ -143,6 +166,8 @@ Future JournalRecorder::append(uint64_t tag_tid,
 }
 
 void JournalRecorder::flush(Context *on_safe) {
+  ldout(m_cct, 20) << dendl;
+
   C_Flush *ctx;
   {
     Mutex::Locker locker(m_lock);
@@ -172,7 +197,7 @@ void JournalRecorder::close_and_advance_object_set(uint64_t object_set) {
 
   // entry overflow from open object
   if (m_current_set != object_set) {
-    ldout(m_cct, 20) << __func__ << ": close already in-progress" << dendl;
+    ldout(m_cct, 20) << "close already in-progress" << dendl;
     return;
   }
 
@@ -186,8 +211,7 @@ void JournalRecorder::close_and_advance_object_set(uint64_t object_set) {
   ++m_current_set;
   ++m_in_flight_advance_sets;
 
-  ldout(m_cct, 20) << __func__ << ": closing active object set "
-                   << object_set << dendl;
+  ldout(m_cct, 10) << "closing active object set " << object_set << dendl;
   if (close_object_set(m_current_set)) {
     advance_object_set();
   }
@@ -197,8 +221,7 @@ void JournalRecorder::advance_object_set() {
   ceph_assert(m_lock.is_locked());
 
   ceph_assert(m_in_flight_object_closes == 0);
-  ldout(m_cct, 20) << __func__ << ": advance to object set " << m_current_set
-                   << dendl;
+  ldout(m_cct, 10) << "advance to object set " << m_current_set << dendl;
   m_journal_metadata->set_active_set(m_current_set, new C_AdvanceObjectSet(
     this));
 }
@@ -213,8 +236,8 @@ void JournalRecorder::handle_advance_object_set(int r) {
     --m_in_flight_advance_sets;
 
     if (r < 0 && r != -ESTALE) {
-      lderr(m_cct) << __func__ << ": failed to advance object set: "
-                   << cpp_strerror(r) << dendl;
+      lderr(m_cct) << "failed to advance object set: " << cpp_strerror(r)
+                   << dendl;
     }
 
     if (m_in_flight_advance_sets == 0 && m_in_flight_object_closes == 0) {
@@ -230,8 +253,7 @@ void JournalRecorder::handle_advance_object_set(int r) {
 void JournalRecorder::open_object_set() {
   ceph_assert(m_lock.is_locked());
 
-  ldout(m_cct, 10) << __func__ << ": opening object set " << m_current_set
-                   << dendl;
+  ldout(m_cct, 10) << "opening object set " << m_current_set << dendl;
 
   uint8_t splay_width = m_journal_metadata->get_splay_width();
 
@@ -244,15 +266,14 @@ void JournalRecorder::open_object_set() {
       ceph_assert(object_recorder->is_closed());
 
       // ready to close object and open object in active set
-      create_next_object_recorder_unlock(object_recorder);
-    } else {
-      uint8_t splay_offset = object_number % splay_width;
-      m_object_locks[splay_offset]->Unlock();
+      create_next_object_recorder(object_recorder);
     }
   }
+  unlock_object_recorders();
 }
 
 bool JournalRecorder::close_object_set(uint64_t active_set) {
+  ldout(m_cct, 10) << "active_set=" << active_set << dendl;
   ceph_assert(m_lock.is_locked());
 
   // object recorders will invoke overflow handler as they complete
@@ -263,14 +284,14 @@ bool JournalRecorder::close_object_set(uint64_t active_set) {
        it != m_object_ptrs.end(); ++it) {
     ObjectRecorderPtr object_recorder = it->second;
     if (object_recorder->get_object_number() / splay_width != active_set) {
-      ldout(m_cct, 10) << __func__ << ": closing object "
-                       << object_recorder->get_oid() << dendl;
+      ldout(m_cct, 10) << "closing object " << object_recorder->get_oid()
+                       << dendl;
       // flush out all queued appends and hold future appends
       if (!object_recorder->close()) {
         ++m_in_flight_object_closes;
       } else {
-        ldout(m_cct, 20) << __func__ << ": object "
-                         << object_recorder->get_oid() << " closed" << dendl;
+        ldout(m_cct, 10) << "object " << object_recorder->get_oid() << " closed"
+                         << dendl;
       }
     }
   }
@@ -280,30 +301,32 @@ bool JournalRecorder::close_object_set(uint64_t active_set) {
 
 ObjectRecorderPtr JournalRecorder::create_object_recorder(
     uint64_t object_number, shared_ptr<Mutex> lock) {
+  ldout(m_cct, 10) << "object_number=" << object_number << dendl;
   ObjectRecorderPtr object_recorder(new ObjectRecorder(
     m_ioctx, utils::get_object_name(m_object_oid_prefix, object_number),
     object_number, lock, m_journal_metadata->get_work_queue(),
-    m_journal_metadata->get_timer(), m_journal_metadata->get_timer_lock(),
-    &m_object_handler, m_journal_metadata->get_order(), m_flush_interval,
-    m_flush_bytes, m_flush_age, m_max_in_flight_appends));
+    &m_object_handler, m_journal_metadata->get_order(),
+    m_max_in_flight_appends));
+  object_recorder->set_append_batch_options(m_flush_interval, m_flush_bytes,
+                                            m_flush_age);
   return object_recorder;
 }
 
-void JournalRecorder::create_next_object_recorder_unlock(
+void JournalRecorder::create_next_object_recorder(
     ObjectRecorderPtr object_recorder) {
   ceph_assert(m_lock.is_locked());
 
   uint64_t object_number = object_recorder->get_object_number();
   uint8_t splay_width = m_journal_metadata->get_splay_width();
   uint8_t splay_offset = object_number % splay_width;
+  ldout(m_cct, 10) << "object_number=" << object_number << dendl;
 
   ceph_assert(m_object_locks[splay_offset]->is_locked());
 
   ObjectRecorderPtr new_object_recorder = create_object_recorder(
      (m_current_set * splay_width) + splay_offset, m_object_locks[splay_offset]);
 
-  ldout(m_cct, 10) << __func__ << ": "
-                   << "old oid=" << object_recorder->get_oid() << ", "
+  ldout(m_cct, 10) << "old oid=" << object_recorder->get_oid() << ", "
                    << "new oid=" << new_object_recorder->get_oid() << dendl;
   AppendBuffers append_buffers;
   object_recorder->claim_append_buffers(&append_buffers);
@@ -315,7 +338,7 @@ void JournalRecorder::create_next_object_recorder_unlock(
       new_object_recorder->get_object_number());
   }
 
-  new_object_recorder->append_unlock(std::move(append_buffers));
+  new_object_recorder->append(std::move(append_buffers));
   m_object_ptrs[splay_offset] = new_object_recorder;
 }
 
@@ -325,15 +348,13 @@ void JournalRecorder::handle_update() {
   uint64_t active_set = m_journal_metadata->get_active_set();
   if (m_current_set < active_set) {
     // peer journal client advanced the active set
-    ldout(m_cct, 20) << __func__ << ": "
-                     << "current_set=" << m_current_set << ", "
+    ldout(m_cct, 10) << "current_set=" << m_current_set << ", "
                      << "active_set=" << active_set << dendl;
 
     uint64_t current_set = m_current_set;
     m_current_set = active_set;
     if (m_in_flight_advance_sets == 0 && m_in_flight_object_closes == 0) {
-      ldout(m_cct, 20) << __func__ << ": closing current object set "
-                       << current_set << dendl;
+      ldout(m_cct, 10) << "closing current object set " << current_set << dendl;
       if (close_object_set(active_set)) {
         open_object_set();
       }
@@ -342,7 +363,7 @@ void JournalRecorder::handle_update() {
 }
 
 void JournalRecorder::handle_closed(ObjectRecorder *object_recorder) {
-  ldout(m_cct, 10) << __func__ << ": " << object_recorder->get_oid() << dendl;
+  ldout(m_cct, 10) << object_recorder->get_oid() << dendl;
 
   Mutex::Locker locker(m_lock);
 
@@ -356,8 +377,8 @@ void JournalRecorder::handle_closed(ObjectRecorder *object_recorder) {
   --m_in_flight_object_closes;
 
   // object closed after advance active set committed
-  ldout(m_cct, 20) << __func__ << ": object "
-                   << active_object_recorder->get_oid() << " closed" << dendl;
+  ldout(m_cct, 10) << "object " << active_object_recorder->get_oid()
+                   << " closed" << dendl;
   if (m_in_flight_object_closes == 0) {
     if (m_in_flight_advance_sets == 0) {
       // peer forced closing of object set
@@ -370,7 +391,7 @@ void JournalRecorder::handle_closed(ObjectRecorder *object_recorder) {
 }
 
 void JournalRecorder::handle_overflow(ObjectRecorder *object_recorder) {
-  ldout(m_cct, 10) << __func__ << ": " << object_recorder->get_oid() << dendl;
+  ldout(m_cct, 10) << object_recorder->get_oid() << dendl;
 
   Mutex::Locker locker(m_lock);
 
@@ -380,9 +401,8 @@ void JournalRecorder::handle_overflow(ObjectRecorder *object_recorder) {
   ObjectRecorderPtr active_object_recorder = m_object_ptrs[splay_offset];
   ceph_assert(active_object_recorder->get_object_number() == object_number);
 
-  ldout(m_cct, 20) << __func__ << ": object "
-                   << active_object_recorder->get_oid() << " overflowed"
-                   << dendl;
+  ldout(m_cct, 10) << "object " << active_object_recorder->get_oid()
+                   << " overflowed" << dendl;
   close_and_advance_object_set(object_number / splay_width);
 }
 
diff --git a/src/journal/JournalRecorder.h b/src/journal/JournalRecorder.h
index bdeaab58a81ae..382f75acef9c6 100644
--- a/src/journal/JournalRecorder.h
+++ b/src/journal/JournalRecorder.h
@@ -23,11 +23,14 @@ class JournalRecorder {
 public:
   JournalRecorder(librados::IoCtx &ioctx, const std::string &object_oid_prefix,
                   const JournalMetadataPtr &journal_metadata,
-                  uint32_t flush_interval, uint64_t flush_bytes,
-                  double flush_age, uint64_t max_in_flight_appends);
+                  uint64_t max_in_flight_appends);
   ~JournalRecorder();
 
   void shut_down(Context *on_safe);
+
+  void set_append_batch_options(int flush_interval, uint64_t flush_bytes,
+                                double flush_age);
+
   Future append(uint64_t tag_tid, const bufferlist &bl);
   void flush(Context *on_safe);
 
@@ -79,9 +82,9 @@ class JournalRecorder {
 
   JournalMetadataPtr m_journal_metadata;
 
-  uint32_t m_flush_interval;
-  uint64_t m_flush_bytes;
-  double m_flush_age;
+  uint32_t m_flush_interval = 0;
+  uint64_t m_flush_bytes = 0;
+  double m_flush_age = 0;
   uint64_t m_max_in_flight_appends;
 
   Listener m_listener;
@@ -109,7 +112,7 @@ class JournalRecorder {
 
   ObjectRecorderPtr create_object_recorder(uint64_t object_number,
                                            std::shared_ptr<Mutex> lock);
-  void create_next_object_recorder_unlock(ObjectRecorderPtr object_recorder);
+  void create_next_object_recorder(ObjectRecorderPtr object_recorder);
 
   void handle_update();
 
diff --git a/src/journal/Journaler.cc b/src/journal/Journaler.cc
index 34bd30e290fa1..7d9c140958433 100644
--- a/src/journal/Journaler.cc
+++ b/src/journal/Journaler.cc
@@ -393,15 +393,20 @@ void Journaler::committed(const Future &future) {
   m_trimmer->committed(future_impl->get_commit_tid());
 }
 
-void Journaler::start_append(int flush_interval, uint64_t flush_bytes,
-			     double flush_age, uint64_t max_in_flight_appends) {
+void Journaler::start_append(uint64_t max_in_flight_appends) {
   ceph_assert(m_recorder == nullptr);
 
   // TODO verify active object set >= current replay object set
 
   m_recorder = new JournalRecorder(m_data_ioctx, m_object_oid_prefix,
-				   m_metadata, flush_interval, flush_bytes,
-				   flush_age, max_in_flight_appends);
+				   m_metadata, max_in_flight_appends);
+}
+
+void Journaler::set_append_batch_options(int flush_interval,
+                                         uint64_t flush_bytes,
+                                         double flush_age) {
+  ceph_assert(m_recorder != nullptr);
+  m_recorder->set_append_batch_options(flush_interval, flush_bytes, flush_age);
 }
 
 void Journaler::stop_append(Context *on_safe) {
diff --git a/src/journal/Journaler.h b/src/journal/Journaler.h
index 3f3987105e187..a063a6d43a018 100644
--- a/src/journal/Journaler.h
+++ b/src/journal/Journaler.h
@@ -110,8 +110,9 @@ class Journaler {
   void stop_replay(Context *on_finish);
 
   uint64_t get_max_append_size() const;
-  void start_append(int flush_interval, uint64_t flush_bytes, double flush_age,
-                    uint64_t max_in_flight_appends);
+  void start_append(uint64_t max_in_flight_appends);
+  void set_append_batch_options(int flush_interval, uint64_t flush_bytes,
+                                double flush_age);
   Future append(uint64_t tag_tid, const bufferlist &bl);
   void flush_append(Context *on_safe);
   void stop_append(Context *on_safe);
diff --git a/src/journal/ObjectRecorder.cc b/src/journal/ObjectRecorder.cc
index e4049df65207c..127731e95c322 100644
--- a/src/journal/ObjectRecorder.cc
+++ b/src/journal/ObjectRecorder.cc
@@ -10,7 +10,8 @@
 
 #define dout_subsys ceph_subsys_journaler
 #undef dout_prefix
-#define dout_prefix *_dout << "ObjectRecorder: " << this << " "
+#define dout_prefix *_dout << "ObjectRecorder: " << this << " " \
+                           << __func__ << " (" << m_oid << "): "
 
 using namespace cls::journal;
 using std::shared_ptr;
@@ -19,72 +20,64 @@ namespace journal {
 
 ObjectRecorder::ObjectRecorder(librados::IoCtx &ioctx, const std::string &oid,
                                uint64_t object_number, shared_ptr<Mutex> lock,
-                               ContextWQ *work_queue, SafeTimer &timer,
-                               Mutex &timer_lock, Handler *handler,
-                               uint8_t order, uint32_t flush_interval,
-                               uint64_t flush_bytes, double flush_age,
-                               uint64_t max_in_flight_appends)
+                               ContextWQ *work_queue, Handler *handler,
+                               uint8_t order, int32_t max_in_flight_appends)
   : RefCountedObject(NULL, 0), m_oid(oid), m_object_number(object_number),
-    m_cct(NULL), m_op_work_queue(work_queue), m_timer(timer),
-    m_timer_lock(timer_lock), m_handler(handler), m_order(order),
-    m_soft_max_size(1 << m_order), m_flush_interval(flush_interval),
-    m_flush_bytes(flush_bytes), m_flush_age(flush_age),
+    m_cct(NULL), m_op_work_queue(work_queue), m_handler(handler),
+    m_order(order), m_soft_max_size(1 << m_order),
     m_max_in_flight_appends(max_in_flight_appends), m_flush_handler(this),
-    m_lock(lock), m_append_tid(0), m_pending_bytes(0),
-    m_size(0), m_overflowed(false), m_object_closed(false),
-    m_in_flight_flushes(false), m_aio_scheduled(false) {
+    m_lock(lock), m_last_flush_time(ceph_clock_now()), m_append_tid(0),
+    m_overflowed(false), m_object_closed(false), m_in_flight_flushes(false) {
   m_ioctx.dup(ioctx);
   m_cct = reinterpret_cast<CephContext*>(m_ioctx.cct());
   ceph_assert(m_handler != NULL);
+  ldout(m_cct, 20) << dendl;
 }
 
 ObjectRecorder::~ObjectRecorder() {
-  ceph_assert(m_append_task == NULL);
-  ceph_assert(m_append_buffers.empty());
+  ldout(m_cct, 20) << dendl;
+  ceph_assert(m_pending_buffers.empty());
   ceph_assert(m_in_flight_tids.empty());
   ceph_assert(m_in_flight_appends.empty());
-  ceph_assert(!m_aio_scheduled);
 }
 
-bool ObjectRecorder::append_unlock(AppendBuffers &&append_buffers) {
+void ObjectRecorder::set_append_batch_options(int flush_interval,
+                                              uint64_t flush_bytes,
+                                              double flush_age) {
+  ldout(m_cct, 5) << "flush_interval=" << flush_interval << ", "
+                  << "flush_bytes=" << flush_bytes << ", "
+                  << "flush_age=" << flush_age << dendl;
+
   ceph_assert(m_lock->is_locked());
+  m_flush_interval = flush_interval;
+  m_flush_bytes = flush_bytes;
+  m_flush_age = flush_age;
+}
 
-  FutureImplPtr last_flushed_future;
-  bool schedule_append = false;
+bool ObjectRecorder::append(AppendBuffers &&append_buffers) {
+  ldout(m_cct, 20) << "count=" << append_buffers.size() << dendl;
 
-  if (m_overflowed) {
-    m_append_buffers.insert(m_append_buffers.end(),
-                            append_buffers.begin(), append_buffers.end());
-    m_lock->Unlock();
-    return false;
-  }
+  ceph_assert(m_lock->is_locked());
 
-  for (AppendBuffers::const_iterator iter = append_buffers.begin();
-       iter != append_buffers.end(); ++iter) {
-    if (append(*iter, &schedule_append)) {
-      last_flushed_future = iter->first;
+  FutureImplPtr last_flushed_future;
+  for (auto& append_buffer : append_buffers) {
+    ldout(m_cct, 20) << *append_buffer.first << ", "
+                     << "size=" << append_buffer.second.length() << dendl;
+    bool flush_requested = append_buffer.first->attach(&m_flush_handler);
+    if (flush_requested) {
+      last_flushed_future = append_buffer.first;
     }
-  }
 
-  if (last_flushed_future) {
-    flush(last_flushed_future);
-    m_lock->Unlock();
-  } else {
-    m_lock->Unlock();
-    if (schedule_append) {
-      schedule_append_task();
-    } else {
-      cancel_append_task();
-    }
+    m_pending_buffers.push_back(append_buffer);
+    m_pending_bytes += append_buffer.second.length();
   }
-  return (!m_object_closed && !m_overflowed &&
-          m_size + m_pending_bytes >= m_soft_max_size);
+
+  return send_appends(!!last_flushed_future, last_flushed_future);
 }
 
 void ObjectRecorder::flush(Context *on_safe) {
-  ldout(m_cct, 20) << __func__ << ": " << m_oid << dendl;
+  ldout(m_cct, 20) << dendl;
 
-  cancel_append_task();
   Future future;
   {
     Mutex::Locker locker(*m_lock);
@@ -97,11 +90,7 @@ void ObjectRecorder::flush(Context *on_safe) {
     }
 
     // attach the flush to the most recent append
-    if (!m_append_buffers.empty()) {
-      future = Future(m_append_buffers.rbegin()->first);
-
-      flush_appends(true);
-    } else if (!m_pending_buffers.empty()) {
+    if (!m_pending_buffers.empty()) {
       future = Future(m_pending_buffers.rbegin()->first);
     } else if (!m_in_flight_appends.empty()) {
       AppendBuffers &append_buffers = m_in_flight_appends.rbegin()->second;
@@ -111,143 +100,68 @@ void ObjectRecorder::flush(Context *on_safe) {
   }
 
   if (future.is_valid()) {
-    future.flush(on_safe);
+    // cannot be invoked while the same lock context
+    m_op_work_queue->queue(new FunctionContext(
+      [future, on_safe] (int r) mutable {
+        future.flush(on_safe);
+      }));
   } else {
     on_safe->complete(0);
   }
 }
 
 void ObjectRecorder::flush(const FutureImplPtr &future) {
-  ldout(m_cct, 20) << __func__ << ": " << m_oid << " flushing " << *future
-                   << dendl;
-
-  ceph_assert(m_lock->is_locked());
+  ldout(m_cct, 20) << "flushing " << *future << dendl;
 
+  m_lock->Lock();
   if (future->get_flush_handler().get() != &m_flush_handler) {
     // if we don't own this future, re-issue the flush so that it hits the
     // correct journal object owner
     future->flush();
+    m_lock->Unlock();
     return;
   } else if (future->is_flush_in_progress()) {
+    m_lock->Unlock();
     return;
   }
 
-  if (m_object_closed || m_overflowed) {
-    return;
-  }
-
-  AppendBuffers::reverse_iterator r_it;
-  for (r_it = m_append_buffers.rbegin(); r_it != m_append_buffers.rend();
-       ++r_it) {
-    if (r_it->first == future) {
-      break;
-    }
+  bool overflowed = send_appends(true, future);
+  if (overflowed) {
+    notify_handler_unlock();
+  } else {
+    m_lock->Unlock();
   }
-  ceph_assert(r_it != m_append_buffers.rend());
-
-  auto it = (++r_it).base();
-  ceph_assert(it != m_append_buffers.end());
-  ++it;
-
-  AppendBuffers flush_buffers;
-  flush_buffers.splice(flush_buffers.end(), m_append_buffers,
-                       m_append_buffers.begin(), it);
-  send_appends(&flush_buffers);
 }
 
 void ObjectRecorder::claim_append_buffers(AppendBuffers *append_buffers) {
-  ldout(m_cct, 20) << __func__ << ": " << m_oid << dendl;
+  ldout(m_cct, 20) << dendl;
 
   ceph_assert(m_lock->is_locked());
   ceph_assert(m_in_flight_tids.empty());
   ceph_assert(m_in_flight_appends.empty());
   ceph_assert(m_object_closed || m_overflowed);
-  append_buffers->splice(append_buffers->end(), m_append_buffers,
-                         m_append_buffers.begin(), m_append_buffers.end());
+
+  for (auto& append_buffer : m_pending_buffers) {
+    ldout(m_cct, 20) << "detached " << *append_buffer.first << dendl;
+    append_buffer.first->detach();
+  }
+  append_buffers->splice(append_buffers->end(), m_pending_buffers,
+                         m_pending_buffers.begin(), m_pending_buffers.end());
 }
 
 bool ObjectRecorder::close() {
   ceph_assert(m_lock->is_locked());
 
-  ldout(m_cct, 20) << __func__ << ": " << m_oid << dendl;
-
-  cancel_append_task();
-
-  flush_appends(true);
+  ldout(m_cct, 20) << dendl;
+  send_appends(true, {});
 
   ceph_assert(!m_object_closed);
   m_object_closed = true;
-  return (m_in_flight_tids.empty() && !m_in_flight_flushes && !m_aio_scheduled);
-}
-
-void ObjectRecorder::handle_append_task() {
-  ceph_assert(m_timer_lock.is_locked());
-  m_append_task = NULL;
-
-  Mutex::Locker locker(*m_lock);
-  flush_appends(true);
-}
-
-void ObjectRecorder::cancel_append_task() {
-  Mutex::Locker locker(m_timer_lock);
-  if (m_append_task != NULL) {
-    m_timer.cancel_event(m_append_task);
-    m_append_task = NULL;
-  }
-}
-
-void ObjectRecorder::schedule_append_task() {
-  Mutex::Locker locker(m_timer_lock);
-  if (m_append_task == nullptr && m_flush_age > 0) {
-    m_append_task = m_timer.add_event_after(
-      m_flush_age, new FunctionContext([this](int) {
-	  handle_append_task();
-	}));
-  }
-}
-
-bool ObjectRecorder::append(const AppendBuffer &append_buffer,
-                            bool *schedule_append) {
-  ceph_assert(m_lock->is_locked());
-
-  bool flush_requested = false;
-  if (!m_object_closed && !m_overflowed) {
-    flush_requested = append_buffer.first->attach(&m_flush_handler);
-  }
-
-  m_append_buffers.push_back(append_buffer);
-  m_pending_bytes += append_buffer.second.length();
-
-  if (!flush_appends(false)) {
-    *schedule_append = true;
-  }
-  return flush_requested;
-}
-
-bool ObjectRecorder::flush_appends(bool force) {
-  ceph_assert(m_lock->is_locked());
-  if (m_object_closed || m_overflowed) {
-    return true;
-  }
-
-  if (m_append_buffers.empty() ||
-      (!force &&
-       m_size + m_pending_bytes < m_soft_max_size &&
-       (m_flush_interval > 0 && m_append_buffers.size() < m_flush_interval) &&
-       (m_flush_bytes > 0 && m_pending_bytes < m_flush_bytes))) {
-    return false;
-  }
-
-  m_pending_bytes = 0;
-  AppendBuffers append_buffers;
-  append_buffers.swap(m_append_buffers);
-  send_appends(&append_buffers);
-  return true;
+  return (m_in_flight_tids.empty() && !m_in_flight_flushes);
 }
 
 void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
-  ldout(m_cct, 10) << __func__ << ": " << m_oid << " tid=" << tid
-                   << ", r=" << r << dendl;
+  ldout(m_cct, 20) << "tid=" << tid << ", r=" << r << dendl;
 
   AppendBuffers append_buffers;
   {
@@ -257,17 +171,14 @@ void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
     m_in_flight_tids.erase(tid_iter);
 
     InFlightAppends::iterator iter = m_in_flight_appends.find(tid);
-    if (r == -EOVERFLOW || m_overflowed) {
-      if (iter != m_in_flight_appends.end()) {
-        m_overflowed = true;
-      } else {
-        // must have seen an overflow on a previous append op
-        ceph_assert(r == -EOVERFLOW && m_overflowed);
-      }
+    ceph_assert(iter != m_in_flight_appends.end());
+
+    if (r == -EOVERFLOW) {
+      ldout(m_cct, 10) << "append overflowed" << dendl;
+      m_overflowed = true;
 
       // notify of overflow once all in-flight ops are complete
-      if (m_in_flight_tids.empty() && !m_aio_scheduled) {
-        m_append_buffers.splice(m_append_buffers.begin(), m_pending_buffers);
+      if (m_in_flight_tids.empty()) {
         append_overflowed();
         notify_handler_unlock();
       } else {
@@ -276,21 +187,23 @@ void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
       return;
     }
 
-    ceph_assert(iter != m_in_flight_appends.end());
     append_buffers.swap(iter->second);
     ceph_assert(!append_buffers.empty());
 
+    for (auto& append_buffer : append_buffers) {
+      m_object_bytes += append_buffer.second.length();
+    }
+    ldout(m_cct, 20) << "object_bytes=" << m_object_bytes << dendl;
+
     m_in_flight_appends.erase(iter);
     m_in_flight_flushes = true;
     m_lock->Unlock();
   }
 
   // Flag the associated futures as complete.
-  for (AppendBuffers::iterator buf_it = append_buffers.begin();
-       buf_it != append_buffers.end(); ++buf_it) {
-    ldout(m_cct, 20) << __func__ << ": " << *buf_it->first << " marked safe"
-                     << dendl;
-    buf_it->first->safe(r);
+  for (auto& append_buffer : append_buffers) {
+    ldout(m_cct, 20) << *append_buffer.first << " marked safe" << dendl;
+    append_buffer.first->safe(r);
   }
 
   // wake up any flush requests that raced with a RADOS callback
@@ -298,39 +211,25 @@ void ObjectRecorder::handle_append_flushed(uint64_t tid, int r) {
   m_in_flight_flushes = false;
   m_in_flight_flushes_cond.Signal();
 
-  if (!m_aio_scheduled) {
-    if (m_in_flight_appends.empty() &&
-        (m_object_closed || m_aio_sent_size >= m_soft_max_size)) {
-      if (m_aio_sent_size >= m_soft_max_size) {
-        ldout(m_cct, 20) << __func__ << ": " << m_oid
-                         << " soft max size reached, notifying overflow"
-                         << dendl;
-        m_overflowed = true;
-      }
-      // all remaining unsent appends should be redirected to new object
-      m_append_buffers.splice(m_append_buffers.begin(), m_pending_buffers);
+  if (m_in_flight_appends.empty() && (m_object_closed || m_overflowed)) {
+    // all remaining unsent appends should be redirected to new object
+    notify_handler_unlock();
+  } else {
+    bool overflowed = send_appends(false, {});
+    if (overflowed) {
       notify_handler_unlock();
-    } else if (!m_pending_buffers.empty()) {
-      m_aio_scheduled = true;
-      m_lock->Unlock();
-      send_appends_aio();
     } else {
       m_lock->Unlock();
     }
-  } else {
-    m_lock->Unlock();
   }
 }
 
 void ObjectRecorder::append_overflowed() {
-  ldout(m_cct, 10) << __func__ << ": " << m_oid << " append overflowed"
-                   << dendl;
+  ldout(m_cct, 10) << dendl;
 
   ceph_assert(m_lock->is_locked());
   ceph_assert(!m_in_flight_appends.empty());
 
-  cancel_append_task();
-
   InFlightAppends in_flight_appends;
   in_flight_appends.swap(m_in_flight_appends);
 
@@ -342,94 +241,109 @@ void ObjectRecorder::append_overflowed() {
   }
 
   restart_append_buffers.splice(restart_append_buffers.end(),
-                                m_append_buffers,
-                                m_append_buffers.begin(),
-                                m_append_buffers.end());
-  restart_append_buffers.swap(m_append_buffers);
-
-  for (AppendBuffers::const_iterator it = m_append_buffers.begin();
-       it != m_append_buffers.end(); ++it) {
-    ldout(m_cct, 20) << __func__ << ": overflowed " << *it->first
-                     << dendl;
-    it->first->detach();
-  }
+                                m_pending_buffers,
+                                m_pending_buffers.begin(),
+                                m_pending_buffers.end());
+  restart_append_buffers.swap(m_pending_buffers);
 }
 
-void ObjectRecorder::send_appends(AppendBuffers *append_buffers) {
+bool ObjectRecorder::send_appends(bool force, FutureImplPtr flush_future) {
+  ldout(m_cct, 20) << dendl;
+
   ceph_assert(m_lock->is_locked());
-  ceph_assert(!append_buffers->empty());
-
-  for (AppendBuffers::iterator it = append_buffers->begin();
-       it != append_buffers->end(); ++it) {
-    ldout(m_cct, 20) << __func__ << ": flushing " << *it->first
-                     << dendl;
-    it->first->set_flush_in_progress();
-    m_size += it->second.length();
+  if (m_object_closed || m_overflowed) {
+    ldout(m_cct, 20) << "already closed or overflowed" << dendl;
+    return false;
   }
 
-  m_pending_buffers.splice(m_pending_buffers.end(), *append_buffers,
-                           append_buffers->begin(), append_buffers->end());
-  if (!m_aio_scheduled) {
-    m_op_work_queue->queue(new FunctionContext([this] (int r) {
-        send_appends_aio();
-    }));
-    m_aio_scheduled = true;
+  if (m_pending_buffers.empty()) {
+    ldout(m_cct, 20) << "append buffers empty" << dendl;
+    return false;
   }
-}
 
-void ObjectRecorder::send_appends_aio() {
-  librados::AioCompletion *rados_completion;
-  {
-    Mutex::Locker locker(*m_lock);
-    m_aio_scheduled = false;
+  if (!force &&
+      ((m_flush_interval > 0 && m_pending_buffers.size() >= m_flush_interval) ||
+       (m_flush_bytes > 0 && m_pending_bytes >= m_flush_bytes) ||
+       (m_flush_age > 0 &&
+        m_last_flush_time + m_flush_age >= ceph_clock_now()))) {
+    ldout(m_cct, 20) << "forcing batch flush" << dendl;
+    force = true;
+  }
 
-    if (m_pending_buffers.empty()) {
-      ldout(m_cct, 20) << __func__ << ": " << m_oid << " pending buffers empty"
-                       << dendl;
-      return;
+  auto max_in_flight_appends = m_max_in_flight_appends;
+  if (m_flush_interval > 0 || m_flush_bytes > 0 || m_flush_age > 0) {
+    if (!force && max_in_flight_appends == 0) {
+      ldout(m_cct, 20) << "attempting to batch AIO appends" << dendl;
+      max_in_flight_appends = 1;
     }
+  } else if (max_in_flight_appends < 0) {
+    max_in_flight_appends = 0;
+  }
 
-    if (m_max_in_flight_appends != 0 &&
-        m_in_flight_tids.size() >= m_max_in_flight_appends) {
-      ldout(m_cct, 20) << __func__ << ": " << m_oid
-                       << " max in flight appends reached" << dendl;
-      return;
+  if (!force && max_in_flight_appends != 0 &&
+      static_cast<int32_t>(m_in_flight_tids.size()) >= max_in_flight_appends) {
+    ldout(m_cct, 10) << "max in flight appends reached" << dendl;
+    return false;
+  }
+
+  librados::ObjectWriteOperation op;
+  client::guard_append(&op, m_soft_max_size);
+
+  size_t append_bytes = 0;
+  AppendBuffers append_buffers;
+  for (auto it = m_pending_buffers.begin(); it != m_pending_buffers.end(); ) {
+    auto& future = it->first;
+    auto& bl = it->second;
+    auto size = m_object_bytes + m_in_flight_bytes + append_bytes + bl.length();
+    if (size == m_soft_max_size) {
+      ldout(m_cct, 10) << "object at capacity " << *future << dendl;
+      m_overflowed = true;
+    } else if (size > m_soft_max_size) {
+      ldout(m_cct, 10) << "object beyond capacity " << *future << dendl;
+      m_overflowed = true;
+      break;
     }
 
-    if (m_aio_sent_size >= m_soft_max_size) {
-      ldout(m_cct, 20) << __func__ << ": " << m_oid
-                       << " soft max size reached" << dendl;
-      return;
+    bool flush_break = (force && flush_future && flush_future == future);
+    ldout(m_cct, 20) << "flushing " << *future << dendl;
+    future->set_flush_in_progress();
+
+    op.append(bl);
+    op.set_op_flags2(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
+
+    append_bytes += bl.length();
+    append_buffers.push_back(*it);
+    it = m_pending_buffers.erase(it);
+
+    if (flush_break) {
+      ldout(m_cct, 20) << "stopping at requested flush future" << dendl;
+      break;
     }
+  }
+
+  if (append_bytes > 0) {
+    m_last_flush_time = ceph_clock_now();
 
     uint64_t append_tid = m_append_tid++;
     m_in_flight_tids.insert(append_tid);
+    m_in_flight_appends[append_tid].swap(append_buffers);
+    m_in_flight_bytes += append_bytes;
 
-    ldout(m_cct, 10) << __func__ << ": " << m_oid << " flushing journal tid="
-                     << append_tid << dendl;
-
-    librados::ObjectWriteOperation op;
-    client::guard_append(&op, m_soft_max_size);
-    auto append_buffers = &m_in_flight_appends[append_tid];
-
-    for (auto it = m_pending_buffers.begin(); it != m_pending_buffers.end(); ) {
-      ldout(m_cct, 20) << __func__ << ": flushing " << *it->first << dendl;
-      op.append(it->second);
-      op.set_op_flags2(CEPH_OSD_OP_FLAG_FADVISE_DONTNEED);
-      m_aio_sent_size += it->second.length();
-      append_buffers->push_back(*it);
-      it = m_pending_buffers.erase(it);
-      if (m_aio_sent_size >= m_soft_max_size) {
-        break;
-      }
-    }
-    rados_completion = librados::Rados::aio_create_completion(
-        new C_AppendFlush(this, append_tid), nullptr,
-        utils::rados_ctx_callback);
+    ceph_assert(m_pending_bytes >= append_bytes);
+    m_pending_bytes -= append_bytes;
+
+    auto rados_completion = librados::Rados::aio_create_completion(
+      new C_AppendFlush(this, append_tid), nullptr, utils::rados_ctx_callback);
     int r = m_ioctx.aio_operate(m_oid, rados_completion, &op);
     ceph_assert(r == 0);
+    rados_completion->release();
+    ldout(m_cct, 20) << "flushing journal tid=" << append_tid << ", "
+                     << "append_bytes=" << append_bytes << ", "
+                     << "in_flight_bytes=" << m_in_flight_bytes << ", "
+                     << "pending_bytes=" << m_pending_bytes << dendl;
   }
-  rados_completion->release();
+
+  return m_overflowed;
 }
 
 void ObjectRecorder::notify_handler_unlock() {
diff --git a/src/journal/ObjectRecorder.h b/src/journal/ObjectRecorder.h
index 8d250e5f04230..ff00e0a0a1f03 100644
--- a/src/journal/ObjectRecorder.h
+++ b/src/journal/ObjectRecorder.h
@@ -4,6 +4,7 @@
 #ifndef CEPH_JOURNAL_OBJECT_RECORDER_H
 #define CEPH_JOURNAL_OBJECT_RECORDER_H
 
+#include "include/utime.h"
 #include "include/Context.h"
 #include "include/rados/librados.hpp"
 #include "common/Cond.h"
@@ -39,12 +40,13 @@ class ObjectRecorder : public RefCountedObject, boost::noncopyable {
 
   ObjectRecorder(librados::IoCtx &ioctx, const std::string &oid,
                  uint64_t object_number, std::shared_ptr<Mutex> lock,
-                 ContextWQ *work_queue, SafeTimer &timer, Mutex &timer_lock,
-                 Handler *handler, uint8_t order, uint32_t flush_interval,
-                 uint64_t flush_bytes, double flush_age,
-                 uint64_t max_in_flight_appends);
+                 ContextWQ *work_queue, Handler *handler, uint8_t order,
+                 int32_t max_in_flight_appends);
   ~ObjectRecorder() override;
 
+  void set_append_batch_options(int flush_interval, uint64_t flush_bytes,
+                                double flush_age);
+
   inline uint64_t get_object_number() const {
     return m_object_number;
   }
@@ -52,7 +54,7 @@ class ObjectRecorder : public RefCountedObject, boost::noncopyable {
     return m_oid;
   }
 
-  bool append_unlock(AppendBuffers &&append_buffers);
+  bool append(AppendBuffers &&append_buffers);
   void flush(Context *on_safe);
   void flush(const FutureImplPtr &future);
 
@@ -70,7 +72,7 @@ class ObjectRecorder : public RefCountedObject, boost::noncopyable {
 
   inline size_t get_pending_appends() const {
     Mutex::Locker locker(*m_lock);
-    return m_append_buffers.size();
+    return m_pending_buffers.size();
   }
 
 private:
@@ -87,7 +89,6 @@ class ObjectRecorder : public RefCountedObject, boost::noncopyable {
       object_recorder->put();
     }
     void flush(const FutureImplPtr &future) override {
-      Mutex::Locker locker(*(object_recorder->m_lock));
       object_recorder->flush(future);
     }
   };
@@ -111,31 +112,28 @@ class ObjectRecorder : public RefCountedObject, boost::noncopyable {
 
   ContextWQ *m_op_work_queue;
 
-  SafeTimer &m_timer;
-  Mutex &m_timer_lock;
-
   Handler *m_handler;
 
   uint8_t m_order;
   uint64_t m_soft_max_size;
 
-  uint32_t m_flush_interval;
-  uint64_t m_flush_bytes;
-  double m_flush_age;
-  uint32_t m_max_in_flight_appends;
+  uint32_t m_flush_interval = 0;
+  uint64_t m_flush_bytes = 0;
+  double m_flush_age = 0;
+  int32_t m_max_in_flight_appends;
 
   FlushHandler m_flush_handler;
 
-  Context *m_append_task = nullptr;
-
   mutable std::shared_ptr<Mutex> m_lock;
-  AppendBuffers m_append_buffers;
+  AppendBuffers m_pending_buffers;
+  uint64_t m_pending_bytes = 0;
+  utime_t m_last_flush_time;
+
   uint64_t m_append_tid;
-  uint32_t m_pending_bytes;
 
   InFlightTids m_in_flight_tids;
   InFlightAppends m_in_flight_appends;
-  uint64_t m_size;
+  uint64_t m_object_bytes = 0;
   bool m_overflowed;
   bool m_object_closed;
 
@@ -143,21 +141,11 @@ class ObjectRecorder : public RefCountedObject, boost::noncopyable {
 
   bool m_in_flight_flushes;
   Cond m_in_flight_flushes_cond;
+  uint64_t m_in_flight_bytes = 0;
 
-  AppendBuffers m_pending_buffers;
-  uint64_t m_aio_sent_size = 0;
-  bool m_aio_scheduled;
-
-  void handle_append_task();
-  void cancel_append_task();
-  void schedule_append_task();
-
-  bool append(const AppendBuffer &append_buffer, bool *schedule_append);
-  bool flush_appends(bool force);
+  bool send_appends(bool force, FutureImplPtr flush_sentinal);
   void handle_append_flushed(uint64_t tid, int r);
   void append_overflowed();
-  void send_appends(AppendBuffers *append_buffers);
-  void send_appends_aio();
 
   void notify_handler_unlock();
 };
diff --git a/src/librbd/Journal.cc b/src/librbd/Journal.cc
index 5a648282fe997..66abdf9020733 100644
--- a/src/librbd/Journal.cc
+++ b/src/librbd/Journal.cc
@@ -714,6 +714,26 @@ void Journal<I>::flush_commit_position(Context *on_finish) {
   m_journaler->flush_commit_position(on_finish);
 }
 
+template <typename I>
+void Journal<I>::user_flushed() {
+  if (m_state == STATE_READY && !m_user_flushed.exchange(true) &&
+      m_image_ctx.config.template get_val<bool>("rbd_journal_object_writethrough_until_flush")) {
+    Mutex::Locker locker(m_lock);
+    if (m_state == STATE_READY) {
+      CephContext *cct = m_image_ctx.cct;
+      ldout(cct, 5) << this << " " << __func__ << dendl;
+
+      ceph_assert(m_journaler != nullptr);
+      m_journaler->set_append_batch_options(
+        m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_flush_interval"),
+        m_image_ctx.config.template get_val<Option::size_t>("rbd_journal_object_flush_bytes"),
+        m_image_ctx.config.template get_val<double>("rbd_journal_object_flush_age"));
+    } else {
+      m_user_flushed = false;
+    }
+  }
+}
+
 template <typename I>
 uint64_t Journal<I>::append_write_event(uint64_t offset, size_t length,
                                         const bufferlist &bl,
@@ -1169,11 +1189,16 @@ void Journal<I>::complete_event(typename Events::iterator it, int r) {
 template <typename I>
 void Journal<I>::start_append() {
   ceph_assert(m_lock.is_locked());
+
   m_journaler->start_append(
-    m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_flush_interval"),
-    m_image_ctx.config.template get_val<Option::size_t>("rbd_journal_object_flush_bytes"),
-    m_image_ctx.config.template get_val<double>("rbd_journal_object_flush_age"),
     m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_max_in_flight_appends"));
+  if (!m_image_ctx.config.template get_val<bool>("rbd_journal_object_writethrough_until_flush")) {
+    m_journaler->set_append_batch_options(
+      m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_flush_interval"),
+      m_image_ctx.config.template get_val<Option::size_t>("rbd_journal_object_flush_bytes"),
+      m_image_ctx.config.template get_val<double>("rbd_journal_object_flush_age"));
+  }
+
   transition_state(STATE_READY, 0);
 }
 
diff --git a/src/librbd/Journal.h b/src/librbd/Journal.h
index 88979b6252b21..e63cc4a71377e 100644
--- a/src/librbd/Journal.h
+++ b/src/librbd/Journal.h
@@ -130,6 +130,8 @@ class Journal {
 
   void flush_commit_position(Context *on_finish);
 
+  void user_flushed();
+
   uint64_t append_write_event(uint64_t offset, size_t length,
                               const bufferlist &bl,
                               bool flush_entry);
@@ -291,6 +293,8 @@ class Journal {
   uint64_t m_event_tid;
   Events m_events;
 
+  std::atomic<bool> m_user_flushed = false;
+
   std::atomic<uint64_t> m_op_tid = { 0 };
   TidToFutures m_op_futures;
 
diff --git a/src/librbd/io/ImageRequest.cc b/src/librbd/io/ImageRequest.cc
index ce064f1f07a17..166c8702308b7 100644
--- a/src/librbd/io/ImageRequest.cc
+++ b/src/librbd/io/ImageRequest.cc
@@ -686,6 +686,7 @@ void ImageFlushRequest<I>::send_request() {
     ceph_assert(image_ctx.journal != NULL);
     journal_tid = image_ctx.journal->append_io_event(
       journal::EventEntry(journal::AioFlushEvent()), 0, 0, false, 0);
+    image_ctx.journal->user_flushed();
   }
 
   auto object_dispatch_spec = ObjectDispatchSpec::create_flush(
diff --git a/src/librbd/journal/DemoteRequest.cc b/src/librbd/journal/DemoteRequest.cc
index 5b34942684e39..7656caac80909 100644
--- a/src/librbd/journal/DemoteRequest.cc
+++ b/src/librbd/journal/DemoteRequest.cc
@@ -135,7 +135,7 @@ void DemoteRequest<I>::append_event() {
   bufferlist event_entry_bl;
   encode(event_entry, event_entry_bl);
 
-  m_journaler->start_append(0, 0, 0, 0);
+  m_journaler->start_append(0);
   m_future = m_journaler->append(m_tag_tid, event_entry_bl);
 
   auto ctx = create_context_callback<
diff --git a/src/librbd/journal/PromoteRequest.cc b/src/librbd/journal/PromoteRequest.cc
index 30d9f3bb159f7..22dc83a32d860 100644
--- a/src/librbd/journal/PromoteRequest.cc
+++ b/src/librbd/journal/PromoteRequest.cc
@@ -119,7 +119,7 @@ void PromoteRequest<I>::append_event() {
   bufferlist event_entry_bl;
   encode(event_entry, event_entry_bl);
 
-  m_journaler->start_append(0, 0, 0, 0);
+  m_journaler->start_append(0);
   m_future = m_journaler->append(m_tag_tid, event_entry_bl);
 
   auto ctx = create_context_callback<
diff --git a/src/test/journal/mock/MockJournaler.h b/src/test/journal/mock/MockJournaler.h
index 787d197dd9347..ab424cd6b092d 100644
--- a/src/test/journal/mock/MockJournaler.h
+++ b/src/test/journal/mock/MockJournaler.h
@@ -120,7 +120,8 @@ struct MockJournaler {
   MOCK_METHOD0(stop_replay, void());
   MOCK_METHOD1(stop_replay, void(Context *on_finish));
 
-  MOCK_METHOD4(start_append, void(int, uint64_t, double, uint64_t));
+  MOCK_METHOD1(start_append, void(uint64_t));
+  MOCK_METHOD3(set_append_batch_options, void(int, uint64_t, double));
   MOCK_CONST_METHOD0(get_max_append_size, uint64_t());
   MOCK_METHOD2(append, MockFutureProxy(uint64_t tag_id,
                                        const bufferlist &bl));
@@ -259,11 +260,14 @@ struct MockJournalerProxy {
     MockJournaler::get_instance().stop_replay(on_finish);
   }
 
-  void start_append(int flush_interval, uint64_t flush_bytes, double flush_age,
-                    uint64_t max_in_flight_appends) {
-    MockJournaler::get_instance().start_append(flush_interval, flush_bytes,
-                                               flush_age,
-                                               max_in_flight_appends);
+  void start_append(uint64_t max_in_flight_appends) {
+    MockJournaler::get_instance().start_append(max_in_flight_appends);
+  }
+
+  void set_append_batch_options(int flush_interval, uint64_t flush_bytes,
+                                double flush_age) {
+    MockJournaler::get_instance().set_append_batch_options(
+      flush_interval, flush_bytes, flush_age);
   }
 
   uint64_t get_max_append_size() const {
diff --git a/src/test/journal/test_JournalRecorder.cc b/src/test/journal/test_JournalRecorder.cc
index fb7c06772eccb..7197526a1ce40 100644
--- a/src/test/journal/test_JournalRecorder.cc
+++ b/src/test/journal/test_JournalRecorder.cc
@@ -22,8 +22,9 @@ class TestJournalRecorder : public RadosTestFixture {
   journal::JournalRecorder *create_recorder(
       const std::string &oid, const journal::JournalMetadataPtr &metadata) {
     journal::JournalRecorder *recorder(new journal::JournalRecorder(
-        m_ioctx, oid + ".", metadata, 0, std::numeric_limits<uint32_t>::max(),
-        0, 0));
+        m_ioctx, oid + ".", metadata, 0));
+    recorder->set_append_batch_options(0, std::numeric_limits<uint32_t>::max(),
+                                       0);
     m_recorders.push_back(recorder);
     return recorder;
   }
diff --git a/src/test/journal/test_ObjectRecorder.cc b/src/test/journal/test_ObjectRecorder.cc
index 4b887f1517037..3cc8e893cfe0c 100644
--- a/src/test/journal/test_ObjectRecorder.cc
+++ b/src/test/journal/test_ObjectRecorder.cc
@@ -72,14 +72,12 @@ class TestObjectRecorder : public RadosTestFixture {
     RadosTestFixture::TearDown();
   }
 
-  inline void set_flush_interval(uint32_t i) {
-    m_flush_interval = i;
-  }
-  inline void set_flush_bytes(uint64_t i) {
-    m_flush_bytes = i;
-  }
-  inline void set_flush_age(double i) {
-    m_flush_age = i;
+  inline void set_batch_options(uint32_t flush_interval, uint64_t flush_bytes,
+                                double flush_age, int max_in_flight) {
+    m_flush_interval = flush_interval;
+    m_flush_bytes = flush_bytes;
+    m_flush_age = flush_age;
+    m_max_in_flight_appends = max_in_flight;
   }
 
   journal::AppendBuffer create_append_buffer(uint64_t tag_tid, uint64_t entry_tid,
@@ -96,9 +94,13 @@ class TestObjectRecorder : public RadosTestFixture {
   journal::ObjectRecorderPtr create_object(const std::string &oid,
                                            uint8_t order, shared_ptr<Mutex> lock) {
     journal::ObjectRecorderPtr object(new journal::ObjectRecorder(
-      m_ioctx, oid, 0, lock, m_work_queue, *m_timer, m_timer_lock, &m_handler,
-      order, m_flush_interval, m_flush_bytes, m_flush_age,
+      m_ioctx, oid, 0, lock, m_work_queue, &m_handler, order,
       m_max_in_flight_appends));
+    {
+      Mutex::Locker locker(*lock);
+      object->set_append_batch_options(m_flush_interval, m_flush_bytes,
+                                       m_flush_age);
+    }
     m_object_recorders.push_back(object);
     m_object_recorder_locks.insert(std::make_pair(oid, lock));
     m_handler.object_lock = lock;
@@ -113,6 +115,7 @@ TEST_F(TestObjectRecorder, Append) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
+  set_batch_options(0, 0, 0, 0);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -121,15 +124,17 @@ TEST_F(TestObjectRecorder, Append) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
-  ASSERT_EQ(1U, object->get_pending_appends());
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
+  ASSERT_EQ(0U, object->get_pending_appends());
 
   journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
                                                               "payload");
   append_buffers = {append_buffer2};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
-  ASSERT_EQ(2U, object->get_pending_appends());
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
+  ASSERT_EQ(0U, object->get_pending_appends());
 
   C_SaferCond cond;
   append_buffer2.first->flush(&cond);
@@ -144,7 +149,7 @@ TEST_F(TestObjectRecorder, AppendFlushByCount) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
-  set_flush_interval(2);
+  set_batch_options(2, 0, 0, -1);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -153,14 +158,16 @@ TEST_F(TestObjectRecorder, AppendFlushByCount) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(1U, object->get_pending_appends());
 
   journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
                                                               "payload");
   append_buffers = {append_buffer2};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(0U, object->get_pending_appends());
 
   C_SaferCond cond;
@@ -175,7 +182,7 @@ TEST_F(TestObjectRecorder, AppendFlushByBytes) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
-  set_flush_bytes(10);
+  set_batch_options(0, 10, 0, -1);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -184,14 +191,16 @@ TEST_F(TestObjectRecorder, AppendFlushByBytes) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(1U, object->get_pending_appends());
 
   journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
                                                               "payload");
   append_buffers = {append_buffer2};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(0U, object->get_pending_appends());
 
   C_SaferCond cond;
@@ -206,7 +215,7 @@ TEST_F(TestObjectRecorder, AppendFlushByAge) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
-  set_flush_age(0.1);
+  set_batch_options(0, 0, 0.1, -1);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -215,13 +224,15 @@ TEST_F(TestObjectRecorder, AppendFlushByAge) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
 
   journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
                                                               "payload");
   append_buffers = {append_buffer2};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
 
   C_SaferCond cond;
   append_buffer2.first->wait(&cond);
@@ -245,13 +256,15 @@ TEST_F(TestObjectRecorder, AppendFilledObject) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
 
   journal::AppendBuffer append_buffer2 = create_append_buffer(234, 124,
                                                               payload);
   append_buffers = {append_buffer2};
   lock->Lock();
-  ASSERT_TRUE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_TRUE(object->append(std::move(append_buffers)));
+  lock->Unlock();
 
   C_SaferCond cond;
   append_buffer2.first->wait(&cond);
@@ -266,6 +279,7 @@ TEST_F(TestObjectRecorder, Flush) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
+  set_batch_options(0, 10, 0, -1);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -274,7 +288,8 @@ TEST_F(TestObjectRecorder, Flush) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(1U, object->get_pending_appends());
 
   C_SaferCond cond1;
@@ -294,6 +309,7 @@ TEST_F(TestObjectRecorder, FlushFuture) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
+  set_batch_options(0, 10, 0, -1);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -302,15 +318,13 @@ TEST_F(TestObjectRecorder, FlushFuture) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(1U, object->get_pending_appends());
 
   C_SaferCond cond;
   append_buffer.first->wait(&cond);
-  lock->Lock();
   object->flush(append_buffer.first);
-  ASSERT_TRUE(lock->is_locked());
-  lock->Unlock();
   ASSERT_TRUE(append_buffer.first->is_flush_in_progress() ||
               append_buffer.first->is_complete());
   ASSERT_EQ(0, cond.wait());
@@ -332,13 +346,11 @@ TEST_F(TestObjectRecorder, FlushDetachedFuture) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer};
 
-  lock->Lock();
   object->flush(append_buffer.first);
-  ASSERT_TRUE(lock->is_locked());
-  lock->Unlock();
   ASSERT_FALSE(append_buffer.first->is_flush_in_progress());
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
 
   // should automatically flush once its attached to the object
   C_SaferCond cond;
@@ -353,7 +365,7 @@ TEST_F(TestObjectRecorder, Close) {
   journal::JournalMetadataPtr metadata = create_metadata(oid);
   ASSERT_EQ(0, init_metadata(metadata));
 
-  set_flush_interval(2);
+  set_batch_options(2, 0, 0, -1);
   shared_ptr<Mutex> lock(new Mutex("object_recorder_lock"));
   journal::ObjectRecorderPtr object = create_object(oid, 24, lock);
 
@@ -362,7 +374,8 @@ TEST_F(TestObjectRecorder, Close) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1};
   lock->Lock();
-  ASSERT_FALSE(object->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object->append(std::move(append_buffers)));
+  lock->Unlock();
   ASSERT_EQ(1U, object->get_pending_appends());
 
   lock->Lock();
@@ -402,7 +415,8 @@ TEST_F(TestObjectRecorder, Overflow) {
   journal::AppendBuffers append_buffers;
   append_buffers = {append_buffer1, append_buffer2};
   lock1->Lock();
-  ASSERT_TRUE(object1->append_unlock(std::move(append_buffers)));
+  ASSERT_TRUE(object1->append(std::move(append_buffers)));
+  lock1->Unlock();
 
   C_SaferCond cond;
   append_buffer2.first->wait(&cond);
@@ -433,7 +447,8 @@ TEST_F(TestObjectRecorder, Overflow) {
                                                               payload);
   append_buffers = {append_buffer3};
   lock2->Lock();
-  ASSERT_FALSE(object2->append_unlock(std::move(append_buffers)));
+  ASSERT_FALSE(object2->append(std::move(append_buffers)));
+  lock2->Unlock();
   append_buffer3.first->flush(NULL);
 
   overflowed = false;
diff --git a/src/test/librbd/fsx.cc b/src/test/librbd/fsx.cc
index 1e96b502798f4..25766b5002d37 100644
--- a/src/test/librbd/fsx.cc
+++ b/src/test/librbd/fsx.cc
@@ -435,7 +435,7 @@ int replay_journal(rados_ioctx_t ioctx, const char *image_name,
                 return r;
         }
 
-        replay_journaler.start_append(0, 0, 0, 0);
+        replay_journaler.start_append(0);
 
         C_SaferCond replay_ctx;
         ReplayHandler replay_handler(&journaler, &replay_journaler,
diff --git a/src/test/librbd/journal/test_mock_PromoteRequest.cc b/src/test/librbd/journal/test_mock_PromoteRequest.cc
index 0e61a88890d38..68a627a79a8d0 100644
--- a/src/test/librbd/journal/test_mock_PromoteRequest.cc
+++ b/src/test/librbd/journal/test_mock_PromoteRequest.cc
@@ -120,7 +120,7 @@ class TestMockJournalPromoteRequest : public TestMockFixture {
   }
 
   void expect_start_append(::journal::MockJournaler &mock_journaler) {
-    EXPECT_CALL(mock_journaler, start_append(_, _, _, _));
+    EXPECT_CALL(mock_journaler, start_append(_));
   }
 
   void expect_stop_append(::journal::MockJournaler &mock_journaler, int r) {
diff --git a/src/test/librbd/mock/MockJournal.h b/src/test/librbd/mock/MockJournal.h
index dd932830707cf..31806217af165 100644
--- a/src/test/librbd/mock/MockJournal.h
+++ b/src/test/librbd/mock/MockJournal.h
@@ -65,6 +65,8 @@ struct MockJournal {
 
   MOCK_METHOD0(allocate_op_tid, uint64_t());
 
+  MOCK_METHOD0(user_flushed, void());
+
   MOCK_METHOD3(append_op_event_mock, void(uint64_t, const journal::EventEntry&,
                                           Context *));
   void append_op_event(uint64_t op_tid, journal::EventEntry &&event_entry,
diff --git a/src/test/librbd/test_mock_Journal.cc b/src/test/librbd/test_mock_Journal.cc
index 727d0db838193..86e31f905d778 100644
--- a/src/test/librbd/test_mock_Journal.cc
+++ b/src/test/librbd/test_mock_Journal.cc
@@ -396,7 +396,16 @@ class TestMockJournal : public TestMockFixture {
   }
 
   void expect_start_append(::journal::MockJournaler &mock_journaler) {
-    EXPECT_CALL(mock_journaler, start_append(_, _, _, _));
+    EXPECT_CALL(mock_journaler, start_append(_));
+  }
+
+  void expect_set_append_batch_options(MockJournalImageCtx &mock_image_ctx,
+                                       ::journal::MockJournaler &mock_journaler,
+                                       bool user_flushed) {
+    if (mock_image_ctx.image_ctx->config.get_val<bool>("rbd_journal_object_writethrough_until_flush") ==
+          user_flushed) {
+      EXPECT_CALL(mock_journaler, set_append_batch_options(_, _, _));
+    }
   }
 
   void expect_stop_append(::journal::MockJournaler &mock_journaler, int r) {
@@ -518,6 +527,7 @@ class TestMockJournal : public TestMockFixture {
     expect_committed(mock_journaler, 0);
     expect_flush_commit_position(mock_journaler);
     expect_start_append(mock_journaler);
+    expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
     ASSERT_EQ(0, when_open(mock_journal));
   }
 
@@ -585,6 +595,7 @@ TEST_F(TestMockJournal, StateTransitions) {
   expect_flush_commit_position(mock_journaler);
 
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
 
   ASSERT_EQ(0, when_open(mock_journal));
 
@@ -662,6 +673,7 @@ TEST_F(TestMockJournal, ReplayCompleteError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, 0);
@@ -719,6 +731,7 @@ TEST_F(TestMockJournal, FlushReplayError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, 0);
@@ -773,6 +786,7 @@ TEST_F(TestMockJournal, CorruptEntry) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, -EINVAL);
@@ -811,6 +825,7 @@ TEST_F(TestMockJournal, StopError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, -EINVAL);
@@ -876,6 +891,7 @@ TEST_F(TestMockJournal, ReplayOnDiskPreFlushError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
 
   C_SaferCond ctx;
   mock_journal.open(&ctx);
@@ -958,6 +974,7 @@ TEST_F(TestMockJournal, ReplayOnDiskPostFlushError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
 
   C_SaferCond ctx;
   mock_journal.open(&ctx);
@@ -1272,6 +1289,7 @@ TEST_F(TestMockJournal, ExternalReplay) {
   InSequence seq;
   expect_stop_append(mock_journaler, 0);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   expect_shut_down_journaler(mock_journaler);
 
   C_SaferCond start_ctx;
@@ -1303,6 +1321,7 @@ TEST_F(TestMockJournal, ExternalReplayFailure) {
   InSequence seq;
   expect_stop_append(mock_journaler, -EINVAL);
   expect_start_append(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   expect_shut_down_journaler(mock_journaler);
 
   C_SaferCond start_ctx;
@@ -1486,4 +1505,27 @@ TEST_F(TestMockJournal, ForcePromoted) {
   ASSERT_EQ(0, listener.ctx.wait());
 }
 
+TEST_F(TestMockJournal, UserFlushed) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockJournalImageCtx mock_image_ctx(*ictx);
+  MockJournal mock_journal(mock_image_ctx);
+  MockObjectDispatch mock_object_dispatch;
+  ::journal::MockJournaler mock_journaler;
+  MockJournalOpenRequest mock_open_request;
+  open_journal(mock_image_ctx, mock_journal, mock_object_dispatch,
+               mock_journaler, mock_open_request);
+  BOOST_SCOPE_EXIT_ALL(&) {
+    close_journal(mock_image_ctx, mock_journal, mock_journaler);
+  };
+
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, true);
+  mock_journal.user_flushed();
+
+  expect_shut_down_journaler(mock_journaler);
+}
+
 } // namespace librbd
diff --git a/src/tools/rbd/action/Journal.cc b/src/tools/rbd/action/Journal.cc
index 76f76b1c5ebc7..db506207c4b0c 100644
--- a/src/tools/rbd/action/Journal.cc
+++ b/src/tools/rbd/action/Journal.cc
@@ -832,7 +832,7 @@ class JournalImporter {
     if (r < 0) {
       return r;
     }
-    m_journaler.start_append(0, 0, 0, 0);
+    m_journaler.start_append(0);
 
     int r1 = 0;
     bufferlist bl;