### Question:
KuduToBigQuery { public static PipelineResult run(KuduToBigQueryOptions options) { Pipeline pipeline = Pipeline.create(options); KuduIO.Read<TableRow> readKuduRecords = KuduIO.<TableRow>read() .withMasterAddresses(options.getKuduMasterAddresses()) .withTable(options.getkuduTableSource()); if (!options.getIncludeColumns().equals("")) { String includedColumns = options.getIncludeColumns().replaceAll("\\s+", ""); List<String> includedColumnsList = Splitter.on(',').splitToList(includedColumns); readKuduRecords.withProjectedColumns(includedColumnsList); } PCollection<TableRow> tableRows = pipeline.apply( readKuduRecords.withParseFn(KuduConverters.of()).withCoder(TableRowJsonCoder.of())); tableRows.apply( "Write to BigQuery", BigQueryIO.writeTableRows() .withoutValidation() .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER) .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND) .to(options.getBigQueryTableSpec())); return pipeline.run(); }  static void main(String[] args); static PipelineResult run(KuduToBigQueryOptions options);  }### Answer:
@Test public void kuduRowResultShouldConvertedToBigQueryTableRowAsExpected() { TableRow expectedTableRow = new TableRow(); expectedTableRow.set("INT32", 1); expectedTableRow.set("BOOL", true); expectedTableRow.set("DOUBLE", 1.2); expectedTableRow.set("FLOAT", 1.3); expectedTableRow.set("INT8", 8); expectedTableRow.set("INT16", 16); expectedTableRow.set("INT64", 64); expectedTableRow.set("STRING", "string"); expectedTableRow.set("UNIXTIME_MICROS", 1586422251000L); RowResult kuduResult = Mockito.mock(RowResult.class); Mockito.doReturn(createTestKuduSchema()).when(kuduResult).getSchema(); Mockito.doReturn(1).when(kuduResult).getObject(0); Mockito.doReturn(true).when(kuduResult).getObject(1); Mockito.doReturn(1.2).when(kuduResult).getObject(2); Mockito.doReturn(1.3).when(kuduResult).getObject(3); Mockito.doReturn(8).when(kuduResult).getObject(4); Mockito.doReturn(16).when(kuduResult).getObject(5); Mockito.doReturn(64).when(kuduResult).getObject(6); Mockito.doReturn("string").when(kuduResult).getObject(7); Mockito.doReturn(1586422251000000000L).when(kuduResult).getLong(8); KuduToBigQuery.KuduConverters kuduConverters = new KuduToBigQuery.KuduConverters(); TableRow convertedTableRow = kuduConverters.apply(kuduResult); List<TableRow> rows = ImmutableList.of(convertedTableRow); PCollection<TableRow> resultRows = pipeline.apply("CreateInput", Create.of(rows)); assertThat(convertedTableRow, equalTo(expectedTableRow)); PAssert.that(resultRows) .satisfies( collection -> { TableRow result = collection.iterator().next(); assertThat(result.get("INT32"), equalTo(1)); assertThat(result.get("BOOL"), equalTo(true)); assertThat(result.get("DOUBLE"), equalTo(1.2)); assertThat(result.get("FLOAT"), equalTo(1.3)); assertThat(result.get("INT8"), equalTo(8)); assertThat(result.get("INT16"), equalTo(16)); assertThat(result.get("INT64"), equalTo(64)); assertThat(result.get("STRING"), equalTo("string")); assertThat(result.get("UNIXTIME_MICROS"), equalTo(1586422251000L)); return null; }); pipeline.run(); }