rgw: handle stripe transition when flushing final pending_data_bl

Fixes: http://tracker.ceph.com/issues/15745

When complete_writing_data() is called, if pending_data_bl is not empty
we still need to handle stripe transition correctly. If pending_data_bl
has more data that we can allow in current stripe, move to the next one.

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
(cherry picked from commit b7a7d41)

src/rgw/rgw_rados.cc

@@ -2409,18 +2409,33 @@ int RGWPutObjProcessor_Atomic::complete_writing_data()
     }
     obj_len = (uint64_t)first_chunk.length();
   }
-  if (pending_data_bl.length()) {
+  while (pending_data_bl.length()) {
     void *handle;
-    int r = write_data(pending_data_bl, data_ofs, &handle, false);
+    uint64_t max_write_size = MIN(max_chunk_size, (uint64_t)next_part_ofs - data_ofs);
+    if (max_write_size > pending_data_bl.length()) {
+      max_write_size = pending_data_bl.length();
+    }
+    bufferlist bl;
+    pending_data_bl.splice(0, max_write_size, &bl);
+    int r = write_data(bl, data_ofs, &handle, false);
     if (r < 0) {
       ldout(store->ctx(), 0) << "ERROR: write_data() returned " << r << dendl;
       return r;
     }
+    data_ofs += bl.length();
     r = throttle_data(handle, false);
     if (r < 0) {
       ldout(store->ctx(), 0) << "ERROR: throttle_data() returned " << r << dendl;
       return r;
     }
+
+    if (data_ofs >= next_part_ofs) {
+      r = prepare_next_part(data_ofs);
+      if (r < 0) {
+        ldout(store->ctx(), 0) << "ERROR: prepare_next_part() returned " << r << dendl;
+        return r;
+      }
+    }
   }
   int r = complete_parts();
   if (r < 0) {
@@ -3013,6 +3028,9 @@ int RGWRados::get_required_alignment(rgw_bucket& bucket, uint64_t *alignment)
       << r << dendl;
     return r;
   }
+  if (align != 0) {
+    ldout(cct, 20) << "required alignment=" << align << dendl;
+  }
   *alignment = align;
   return 0;
 }
@@ -3039,6 +3057,8 @@ int RGWRados::get_max_chunk_size(rgw_bucket& bucket, uint64_t *max_chunk_size)
 
   *max_chunk_size = config_chunk_size - (config_chunk_size % alignment);
 
+  ldout(cct, 20) << "max_chunk_size=" << *max_chunk_size << dendl;
+
   return 0;
 }