[python] - Expand KuduError capabilities

The Python client currently doesn't support the full error reporting
capabilities of the KuduError class. This patch exposes most of them
and includes tests.

Change-Id: I4947a663ae09ec2fc9580876521252670d081aa1
Reviewed-on: http://gerrit.cloudera.org:8080/4885
Reviewed-by: Jean-Daniel Cryans <[email protected]>
Tested-by: Kudu Jenkins

python/kudu/client.pyx

@@ -2308,7 +2308,29 @@ cdef class KuduError:
             del self.error
 
     def failed_op(self):
-        raise NotImplementedError
+        """
+        Get debug string representation of the failed operation.
+
+        Returns
+        -------
+        op : str
+        """
+        return frombytes(self.error.failed_op().ToString())
+
+    def was_possibly_successful(self):
+        """
+        Check if there is a chance that the requested operation was successful.
+
+        In some cases, it is possible that the server did receive and
+        successfully perform the requested operation, but the client can't
+        tell whether or not it was successful. For example, if the call times
+        out, the server may still succeed in processing at a later time.
+
+        Returns
+        -------
+        result : bool
+        """
+        return self.error.was_possibly_successful()
 
     def __repr__(self):
         return "KuduError('%s')" % (self.error.status().ToString())

python/kudu/tests/test_client.py

@@ -206,6 +206,31 @@ def test_insert_and_mutate_rows(self):
         scanner = table.scanner().open()
         assert len(scanner.read_all_tuples()) == 0
 
+    def test_failed_write_op(self):
+        # Insert row
+        table = self.client.table(self.ex_table)
+        session = self.client.new_session()
+        session.apply(table.new_insert({'key': 1}))
+        session.flush()
+
+        # Attempt to insert row again
+        session.apply(table.new_insert({'key': 1}))
+        self.assertRaises(kudu.KuduBadStatus, session.flush)
+
+        # Check errors
+        errors, overflowed = session.get_pending_errors()
+        self.assertFalse(overflowed)
+        self.assertEqual(len(errors), 1)
+        error = errors[0]
+        # This test passes because we currently always return
+        # True.
+        self.assertTrue(error.was_possibly_successful())
+        self.assertEqual(error.failed_op(), 'INSERT int32 key=1')
+
+        # Delete inserted row
+        session.apply(table.new_delete({'key': 1}))
+        session.flush()
+
     def test_session_auto_open(self):
         table = self.client.table(self.ex_table)
         scanner = table.scanner()