### Question:
WriteToGCSParquet extends PTransform<PCollection<KV<String, String>>, WriteFilesResult<Void>> { public static WriteToGCSBuilder newBuilder() { return new AutoValue_WriteToGCSParquet.Builder(); }  static WriteToGCSBuilder newBuilder(); abstract String outputDirectory(); abstract String outputFilenamePrefix(); abstract Integer numShards(); @Override WriteFilesResult<Void> 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)", WriteToGCSParquet.newBuilder() .withOutputDirectory(null) .withOutputFilenamePrefix(PARQUET_FILENAME_PREFIX) .setNumShards(NUM_SHARDS) .build()); pipeline.run(); }