Conversation
| **Max bytes per batch:** Maximum number of records, which plugin puts into a Salesforce batch. | ||
| This value cannot be greater than 10,000,000. Which is enforced by Salesforce. | ||
|
|
||
| **Error Handling:** Strategy used to handle erroneous records.<br> |
There was a problem hiding this comment.
I think
is not park of markdown. Just use list below
There was a problem hiding this comment.
Not sure if it's part of markdown or not, but it works. Do you want me remove line breaks here?
| for (int i = 0; i < resultCols; i++) { | ||
| resultInfo.put(resultHeader.get(i), row.get(i)); | ||
| } | ||
| boolean success = Boolean.valueOf(resultInfo.get("Success")); |
| try { | ||
| return new SalesforceRecordWriter(taskAttemptContext); | ||
| } catch (AsyncApiException e) { | ||
| throw new RuntimeException("There was issue communicating with Salesforce", e); |
There was a problem hiding this comment.
is there a strong reason to wrap this exception? Same question for all similar cases
There was a problem hiding this comment.
This is checked exception. I think I wrap only those which are in overriden methods which does not have "throws" and also the one which is in config validation.
| * @param config Salesforce batch sink configuration | ||
| */ | ||
| public SalesforceOutputFormatProvider(SalesforceSinkConfig config) { | ||
| ImmutableMap.Builder<String, String> builder = new ImmutableMap.Builder<String, String>() |
There was a problem hiding this comment.
builder variable is redundant
| submitCurrentBatch(true); | ||
| } | ||
|
|
||
| csvBufferSizeCheck.write(csvRecord); |
There was a problem hiding this comment.
shouldn't it write to csvBuffer?
| public void write(NullWritable key, CSVRecord csvRecord) throws IOException { | ||
| csvBufferSizeCheck.write(csvRecord); | ||
|
|
||
| if(csvBufferSizeCheck.size() > maxBytesPerBatch || |
There was a problem hiding this comment.
why do we need additional buffer just to check sizes? It's a bit confusing
| try { | ||
| creatableSObjectFields = getCreatableSObjectFields(); | ||
| } catch (ConnectionException e) { | ||
| throw new RuntimeException("There was issue communicating with Salesforce", e); |
There was a problem hiding this comment.
should be InvalidStageException
| inputFields.removeAll(creatableSObjectFields); | ||
|
|
||
| if (!inputFields.isEmpty()) { | ||
| throw new IllegalArgumentException(String.format("Following schema fields: '%s' are not present " + |
There was a problem hiding this comment.
should be InvalidStageException
aonischuk
force-pushed
the
feature/CDAP-15313-salesforce-batch-sink
branch
3 times, most recently
from
9c2c07d to
f49b34d
Compare
aonischuk
force-pushed
the
feature/CDAP-15313-salesforce-batch-sink
branch
from
f49b34d to
400f09d
Compare
2 participants
Insert records into Salesforce sObjects using Salesforce Bulk API. Diving the data into separate Salesforce batches and submit them from separate mappers.
How Insert to Salesforce Batch API works:
For more info:
https://wiki.cask.co/display/CE/Salesforce+Batch+Sink