### Question:
WriteToGCSText extends PTransform<PCollection<KV<String, String>>, PDone> { public static WriteToGCSBuilder newBuilder() { return new AutoValue_WriteToGCSText.Builder(); }  static WriteToGCSBuilder newBuilder(); abstract String outputDirectory(); abstract String outputFilenamePrefix(); abstract String tempLocation(); abstract Integer numShards(); @Override PDone expand(PCollection<KV<String, String>> kafkaRecords);  }### Answer:
@Test public void testWriteWithoutOutputDirectory() { expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("withOutputDirectory(outputDirectory) called with null input."); pipeline .apply( "CreateInput", Create.of(message).withCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))) .apply( "WriteTextFile(s)", WriteToGCSText.newBuilder() .withOutputDirectory(null) .withOutputFilenamePrefix(TEXT_FILENAME_PREFIX) .setNumShards(NUM_SHARDS) .withTempLocation(FAKE_TEMP_LOCATION) .build()); pipeline.run(); }
@Test public void testWriteWithoutTempLocation() { expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("withTempLocation(tempLocation) called with null input. "); pipeline .apply( "CreateInput", Create.of(message).withCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))) .apply( "WriteTextFile(s)", WriteToGCSText.newBuilder() .withOutputDirectory(FAKE_DIR) .withOutputFilenamePrefix(TEXT_FILENAME_PREFIX) .setNumShards(NUM_SHARDS) .withTempLocation(null) .build()); pipeline.run(); }