### Question:
DataIngestionAPIWriter implements Writer { @SuppressWarnings("unchecked") @Override public int write(List<SDKField> sdkFields, List<List<Object>> dataTable) throws ConnectorSDKException { if(dataTable!=null){ if(writeAttributes.isFlushStrategyRequired()){ FlushHandler flushHandler = writeAttributes.getFlushHandler(); flushHandler.incrementProcessedRows((long) dataTable.size()); flushHandler.addRecords(dataTable); if(isBufferEligibleForFlush(flushHandler)){ dataTable = (List<List<Object>>) flushHandler.getDataTable(); return flushRecords(sdkFields, dataTable, flushHandler); } else return 0; } else{ return flushRecords(sdkFields, dataTable, null); } } else{ if(writeAttributes.isFlushStrategyRequired()){ FlushHandler flushHandler = writeAttributes.getFlushHandler(); if(isBufferEligibleForFlush(flushHandler)){ dataTable = (List<List<Object>>) flushHandler.getDataTable(); return flushRecords(sdkFields, dataTable, flushHandler); } else return 0; } else{ return -1; } } } DataIngestionAPIWriter(DataIngestionService dis,
                                  DataWiringParam param,
                                  FileFormat outputFileFormat,
                                  Formatter formatter,
                                  WriteAttributes writeAttributes,
                                  CatalogService catalogService); @SuppressWarnings("unchecked") @Override /** * Write function invoked incase of flat connector * where hierarchy building needs to be done by sdk itself * return types are 0 for success & -1 for failure. */ int write(List<SDKField> sdkFields, List<List<Object>> dataTable); @SuppressWarnings("unchecked") @Override /** * Write function invoked incase of procedural connector * where hierarchy is already constructed by vendor * return types are 0 for success & -1 for failure. */ int write(List<Object> dataTable); @Override int markBatchCompletion(Boolean isSuccess); @Override int markBatchCompletion(Boolean isSuccess, Boolean shouldPollForBatchStatus);  }### Answer:
@Test public void testWriteAPIForProcedural() throws ConnectorSDKException { when(dis.writeToBatch(any(),any(),any(),any(),any(),any())).thenReturn(0); DataIngestionAPIWriter writer = new DataIngestionAPIWriter(dis, param, CSV_FILE_FORMAT, formatter, writeAttributes, catService); List<Object> data = new ArrayList<>(); assert (writer.write(data) == 0); }
@SuppressWarnings("unchecked") @Test public void testWriteAPIForProceduralWithFlushStrategy() throws ConnectorSDKException, ParseException { DataSet datset = getDataSetFromString(datasetInnerSample1); DataWiringParam param = new DataWiringParam("imsOrg", datset); Formatter platFormatter = new PlatformDataFormatterFactory(writer, param, registryFactory).getFormatter(JSON_FILE_FORMAT); writeAttributes = new WriteAttributes.WriteAttributesBuilder().withFlushStrategy(true).withSizeOfRecord(20).build(); DataIngestionAPIWriter disWriter = new DataIngestionAPIWriter(dis, param, CSV_FILE_FORMAT, platFormatter, writeAttributes, catService); ArrayList<Object> records = new ArrayList<Object>(); JSONObject record1 = new JSONObject(); record1.put("col1", "val1"); record1.put("col2", 45); records.add(record1); JSONObject record2 = new JSONObject(); record2.put("col1", "val2"); record2.put("col2", -45); records.add(record2); try { disWriter.write(records); } catch (Exception e) { e.printStackTrace(); } writeAttributes.setEOF(true); try{ disWriter.write(null); } catch(ConnectorSDKException e){ e.getMessage(); } }