Report per-record errors in batch_operate instead of aborting the batch

lib/aerospike/command/batch_operate_command.rb

@@ -125,6 +125,14 @@ def write_buffer
       mark_compressed(@policy)
     end
 
+    # Batch operate reports per-record status on each BatchRecord, so a non-OK
+    # result code for a single record must not abort the whole batch. Capture
+    # every code in parse_row instead of raising (matches respond_all_keys
+    # semantics of the Java/C/Go clients).
+    def handle_result_code(result_code)
+      # NOOP
+    end
+
     # Parse all results in the batch.  Add records to shared list.
     # If the record was not found, the bins will be nil.
     def parse_row(result_code)

lib/aerospike/command/multi_command.rb

@@ -98,15 +98,7 @@ def parse_group(receive_size)
         read_bytes(MSG_REMAINING_HEADER_SIZE)
         result_code = @data_buffer.read(5).ord & 0xFF
 
-        # The only valid server return codes are "ok", "not found" and "filtered out".
-        # If other return codes are received, then abort the batch.
-        if result_code != 0
-            if [Aerospike::ResultCode::KEY_NOT_FOUND_ERROR, Aerospike::ResultCode::FILTERED_OUT].include?(result_code)
-              # NOOP
-            else
-              raise Aerospike::Exceptions::Aerospike.new(result_code, nil, [@node])
-            end
-        end
+        handle_result_code(result_code)
 
         # If cmd is the end marker of the response, do not proceed further
         info3 = @data_buffer.read(3).ord
@@ -118,6 +110,21 @@ def parse_group(receive_size)
       true
     end
 
+    # Decide what to do with a per-record result code before the row is parsed.
+    #
+    # The only valid server return codes are "ok", "not found" and "filtered out".
+    # If other return codes are received, then abort the batch by raising.
+    #
+    # Commands that report per-record status (e.g. batch operate) override this
+    # to treat non-OK codes as data captured on each record instead of aborting
+    # the whole batch.
+    def handle_result_code(result_code)
+      return if result_code == 0
+      return if [Aerospike::ResultCode::KEY_NOT_FOUND_ERROR, Aerospike::ResultCode::FILTERED_OUT].include?(result_code)
+
+      raise Aerospike::Exceptions::Aerospike.new(result_code, nil, [@node])
+    end
+
     def parse_key(field_count)
       # in Stream queries, there are no keys
       return unless field_count > 0

spec/aerospike/batch_operate_spec.rb

@@ -203,6 +203,30 @@
         expect(exists).to eql false
 
       end
+
+      it 'reports per-record failures without aborting the whole batch' do
+        good_ops = [Aerospike::Operation.put(Aerospike::Bin.new("new_bin", "value"))]
+        # `add` on the pre-existing string bin "key" is a type mismatch, so the
+        # server rejects this single record with BIN_TYPE_ERROR.
+        bad_ops = [Aerospike::Operation.add(Aerospike::Bin.new("key", 1))]
+
+        records = [
+          Aerospike::BatchWrite.new(keys.first, good_ops),
+          Aerospike::BatchWrite.new(keys.last, bad_ops)
+        ]
+
+        expect { client.batch_operate(records, batch_policy) }.not_to raise_error
+
+        # The healthy record is applied and reported as OK...
+        expect(records[0].result_code).to eql Aerospike::ResultCode::OK
+        expect(records[0].record).not_to be_nil
+
+        # ...while the failing record carries its own error code.
+        expect(records[1].result_code).to eql Aerospike::ResultCode::BIN_TYPE_ERROR
+
+        # The successful write is durable and readable.
+        expect(client.get(keys.first).bins["new_bin"]).to eql "value"
+      end
     end
 
     context '#BatchDelete' do