Fix up IcebergIO Write path

 - remove Read path (will propose separately)
 - re-enable checking, fix type errors
 - some style adjustments

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -1151,7 +1151,7 @@ class BeamModulePlugin implements Plugin<Project> {
         options.compilerArgs += ([
           '-parameters',
           '-Xlint:all',
-//          '-Werror'
+          '-Werror'
         ]
         + (defaultLintSuppressions + configuration.disableLintWarnings).collect { "-Xlint:-${it}" })
       }

sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/PrepareWrite.java → sdks/java/io/iceberg/src/main/java/org/apache/beam/io/iceberg/AssignDestinations.java

@@ -31,52 +31,51 @@
 import org.apache.beam.sdk.values.ValueInSingleWindow;
 import org.joda.time.Instant;
 
-@SuppressWarnings("all") // TODO: Remove this once development is stable.
-public class PrepareWrite<InputT, DestinationT, OutputT>
-    extends PTransform<PCollection<InputT>, PCollection<KV<DestinationT, OutputT>>> {
+class AssignDestinations<ElementT, DestinationT, OutputT>
+    extends PTransform<PCollection<ElementT>, PCollection<KV<DestinationT, OutputT>>> {
 
-  private DynamicDestinations<InputT, DestinationT> dynamicDestinations;
-  private SerializableFunction<InputT, OutputT> formatFunction;
-  private Coder outputCoder;
+  private DynamicDestinations<ElementT, DestinationT> dynamicDestinations;
+  private SerializableFunction<ElementT, OutputT> formatFunction;
+  private Coder<OutputT> outputCoder;
 
-  public PrepareWrite(
-      DynamicDestinations<InputT, DestinationT> dynamicDestinations,
-      SerializableFunction<InputT, OutputT> formatFunction,
-      Coder outputCoder) {
+  public AssignDestinations(
+      DynamicDestinations<ElementT, DestinationT> dynamicDestinations,
+      SerializableFunction<ElementT, OutputT> formatFunction,
+      Coder<OutputT> outputCoder) {
     this.dynamicDestinations = dynamicDestinations;
     this.formatFunction = formatFunction;
     this.outputCoder = outputCoder;
   }
 
   @Override
-  public PCollection<KV<DestinationT, OutputT>> expand(PCollection<InputT> input) {
+  public PCollection<KV<DestinationT, OutputT>> expand(PCollection<ElementT> input) {
 
-    final Coder destCoder;
+    final Coder<KV<DestinationT, OutputT>> destCoder;
     try {
       destCoder =
           KvCoder.of(
               dynamicDestinations.getDestinationCoderWithDefault(
                   input.getPipeline().getCoderRegistry()),
               outputCoder);
     } catch (Exception e) {
-      RuntimeException e1 = new RuntimeException("Unable to expand PrepareWrite");
+      RuntimeException e1 = new RuntimeException("Unable to expand AssignDestinations");
       e1.addSuppressed(e);
       throw e1;
     }
     return input
         .apply(
             ParDo.of(
-                    new DoFn<InputT, KV<DestinationT, OutputT>>() {
+                    new DoFn<ElementT, KV<DestinationT, OutputT>>() {
 
                       @ProcessElement
                       public void processElement(
                           ProcessContext c,
-                          @Element InputT element,
+                          @Element ElementT element,
                           @Timestamp Instant timestamp,
                           BoundedWindow window,
                           PaneInfo pane)
                           throws IOException {
-                        ValueInSingleWindow<InputT> windowedElement =
+                        ValueInSingleWindow<ElementT> windowedElement =
                             ValueInSingleWindow.of(element, timestamp, window, pane);
                         dynamicDestinations.setSideInputProcessContext(c);
                         DestinationT tableDestination =