Reduced the number of calls we need to make when creating records using
the sink by better using the create/update recordset api.
This also fixes a bug where the sink could trigger a race condition in
the worker causing it to throw a BadAction exception.
Partial-Bug: #1768618
Change-Id: Iaf21ec59755375d3c3bc043b16a1b14aa991475e
(cherry picked from commit 4869913519)