### Question:
BigQueryToElasticsearch { private static PipelineResult run(BigQueryToElasticsearchReadOptions options) { Pipeline pipeline = Pipeline.create(options); pipeline .apply( "ReadFromBigQuery", ReadBigQuery.newBuilder() .setOptions(options.as(BigQueryToElasticsearchReadOptions.class)) .build()) .apply("TableRowsToJsonDocument", ParDo.of(new TableRowToJsonFn())) .apply( "WriteToElasticsearch", WriteToElasticsearch.newBuilder() .setOptions(options.as(WriteToElasticsearchOptions.class)) .build()); return pipeline.run(); }  static void main(String[] args);  }### Answer:
@Test public void testBigQueryToElasticsearchE2E() { BigQueryToElasticsearch.BigQueryToElasticsearchReadOptions options = PipelineOptionsFactory.create() .as(BigQueryToElasticsearch.BigQueryToElasticsearchReadOptions.class); options.setNodeAddresses("http: options.setIndex("test"); options.setDocumentType("_doc"); options.setInputTableSpec("my-project:my-dataset.my-table"); options.setQuery(null); PCollection<String> testStrings = pipeline .apply("CreateInput", Create.of(rows)) .apply("TestTableRowToJson", ParDo.of(new BigQueryConverters.TableRowToJsonFn())); PAssert.that(testStrings) .satisfies( collection -> { String result = collection.iterator().next(); assertThat(result, is(equalTo(jsonifiedTableRow))); return null; }); pipeline.run(); }
@Test public void testNoQueryOrInputTableSpec() { exceptionRule.expect(IllegalArgumentException.class); BigQueryToElasticsearch.BigQueryToElasticsearchReadOptions options = PipelineOptionsFactory.create() .as(BigQueryToElasticsearch.BigQueryToElasticsearchReadOptions.class); options.setNodeAddresses("http: options.setIndex("test"); options.setDocumentType("_doc"); options.setInputTableSpec(null); options.setQuery(null); pipeline.apply("CreateInput", Create.of(tableRow)); pipeline.run(); }