diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
index f8cc1a4074c1..f630e7413876 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/coders/AvroCoder.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.extensions.avro.coders;
 
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
 import com.google.errorprone.annotations.FormatMethod;
 import com.google.errorprone.annotations.FormatString;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -109,9 +111,6 @@
  *
  * @param <T> the type of elements handled by this coder
  */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class AvroCoder<T> extends CustomCoder<T> {
 
   private static final Cache<AvroCoderCacheKey, AvroCoder<?>> AVRO_CODER_CACHE =
@@ -137,7 +136,12 @@ public static <T> AvroCoder<T> specific(TypeDescriptor<T> type) {
    * suite for encoding and decoding.
    */
   public static <T> AvroCoder<T> specific(Class<T> type) {
-    return specific(type, new SpecificData(type.getClassLoader()).getSchema(type));
+    return specific(type, specificSchemaOf(type));
+  }
+
+  @SuppressWarnings("nullness") // SpecificData tolerates a null class loader but is unannotated
+  private static Schema specificSchemaOf(Class<?> type) {
+    return new SpecificData(type.getClassLoader()).getSchema(type);
   }
 
   /**
@@ -167,7 +171,12 @@ public static <T> AvroCoder<T> reflect(TypeDescriptor<T> type) {
    * suite for encoding and decoding.
    */
   public static <T> AvroCoder<T> reflect(Class<T> type) {
-    return reflect(type, new ReflectData(type.getClassLoader()).getSchema(type));
+    return reflect(type, reflectSchemaOf(type));
+  }
+
+  @SuppressWarnings("nullness") // ReflectData tolerates a null class loader but is unannotated
+  private static Schema reflectSchemaOf(Class<?> type) {
+    return new ReflectData(type.getClassLoader()).getSchema(type);
   }
 
   /**
@@ -395,10 +404,10 @@ public Schema get() {
 
   // writer and reader are unused but kept for serialization update compatibility.
   @SuppressWarnings("unused")
-  private final EmptyOnDeserializationThreadLocal<DatumWriter<T>> writer = null;
+  private final @Nullable EmptyOnDeserializationThreadLocal<DatumWriter<T>> writer = null;
 
   @SuppressWarnings("unused")
-  private final EmptyOnDeserializationThreadLocal<DatumReader<T>> reader = null;
+  private final @Nullable EmptyOnDeserializationThreadLocal<DatumReader<T>> reader = null;
 
   // datumReader and datumWriter are initialized in the constructor and
   // on deserialization (see readObject).
@@ -424,7 +433,8 @@ protected AvroCoder(AvroDatumFactory<T> datumFactory, Schema schema) {
     this.decoder = new EmptyOnDeserializationThreadLocal<>();
     this.encoder = new EmptyOnDeserializationThreadLocal<>();
 
-    initializeAvroDatumReaderAndWriter();
+    this.datumReader = datumFactory.apply(schema, schema);
+    this.datumWriter = datumFactory.apply(schema);
   }
 
   /** Returns the type this coder encodes/decodes. */
@@ -473,6 +483,11 @@ public T decode(InputStream inStream) throws IOException {
     BinaryDecoder decoderInstance = DECODER_FACTORY.directBinaryDecoder(inStream, decoder.get());
     // Save the potentially-new instance for later.
     decoder.set(decoderInstance);
+    return readWithoutReuse(decoderInstance);
+  }
+
+  @SuppressWarnings("nullness") // DatumReader.read accepts a null reuse but is unannotated
+  private T readWithoutReuse(BinaryDecoder decoderInstance) throws IOException {
     return datumReader.read(null, decoderInstance);
   }
 
@@ -808,10 +823,10 @@ private void checkMap(String context, TypeDescriptor<?> type, Schema schema) {
     }
 
     private void checkArray(String context, TypeDescriptor<?> type, Schema schema) {
-      TypeDescriptor<?> elementType = null;
+      TypeDescriptor<?> elementType;
       if (type.isArray()) {
         // The type is an array (with ordering)-> deterministic iff the element is deterministic.
-        elementType = type.getComponentType();
+        elementType = checkNotNull(type.getComponentType());
       } else if (isSubtypeOf(type, Collection.class)) {
         if (isSubtypeOf(type, List.class, SortedSet.class)) {
           // Ordered collection -> deterministic iff the element is deterministic
@@ -895,16 +910,12 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
       this.datumReader = cachedCoder.get().datumReader;
       this.datumWriter = cachedCoder.get().datumWriter;
     } else {
-      initializeAvroDatumReaderAndWriter();
+      Schema schema = this.schemaSupplier.get();
+      this.datumReader = this.datumFactory.apply(schema, schema);
+      this.datumWriter = this.datumFactory.apply(schema);
     }
   }
 
-  private void initializeAvroDatumReaderAndWriter() {
-    this.datumReader =
-        this.datumFactory.apply(this.schemaSupplier.get(), this.schemaSupplier.get());
-    this.datumWriter = this.datumFactory.apply(this.schemaSupplier.get());
-  }
-
   enum AvroCoderType {
     SPECIFIC,
     REFLECT;
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
index 7d2fd43d7ab2..1f010fa4a9bd 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroDatumFactory.java
@@ -34,9 +34,6 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /** Create {@link DatumReader} and {@link DatumWriter} for given schemas. */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public abstract class AvroDatumFactory<T>
     implements AvroSource.DatumReaderFactory<T>, AvroSink.DatumWriterFactory<T> {
 
@@ -172,14 +169,14 @@ public ReflectDatumFactory(Class<T> type) {
 
     @Override
     public DatumReader<T> apply(Schema writer, Schema reader) {
-      ReflectData data = new ReflectData(type.getClassLoader());
+      ReflectData data = newReflectData(type);
       AvroUtils.addLogicalTypeConversions(data);
       return new ReflectDatumReader<>(writer, reader, data);
     }
 
     @Override
     public DatumWriter<T> apply(Schema writer) {
-      ReflectData data = new ReflectData(type.getClassLoader());
+      ReflectData data = newReflectData(type);
       AvroUtils.addLogicalTypeConversions(data);
       return new ReflectDatumWriter<>(writer, data);
     }
@@ -187,5 +184,10 @@ public DatumWriter<T> apply(Schema writer) {
     public static <T> ReflectDatumFactory<T> of(Class<T> type) {
       return new ReflectDatumFactory<>(type);
     }
+
+    @SuppressWarnings("nullness") // ReflectData tolerates a null class loader but is unannotated
+    private static ReflectData newReflectData(Class<?> type) {
+      return new ReflectData(type.getClassLoader());
+    }
   }
 }
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
index 6b23695c21ae..33b78abad4fa 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroIO.java
@@ -19,8 +19,8 @@
 
 import static org.apache.beam.sdk.io.FileIO.ReadMatches.DirectoryTreatment;
 import static org.apache.beam.sdk.io.ReadAllViaFileBasedSource.ReadFileRangesFnExceptionHandler;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
-import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.auto.value.AutoValue;
 import java.io.IOException;
@@ -344,9 +344,6 @@
  * TypedWrite#withBadRecordErrorHandler(ErrorHandler)}. See documentation in {@link FileIO} for
  * details on usage
  */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class AvroIO {
   /**
    * Reads records of the given type from an Avro file (or multiple Avro files matching a pattern).
@@ -619,7 +616,7 @@ private static <UserT, OutputT> TypedWrite.Builder<UserT, Void, OutputT> default
   }
 
   private static <T> PCollection<T> setBeamSchema(
-      PCollection<T> pc, Class<T> clazz, @Nullable Schema schema) {
+      PCollection<T> pc, Class<T> clazz, Schema schema) {
     return pc.setCoder(AvroUtils.schemaCoder(clazz, schema));
   }
 
@@ -637,9 +634,9 @@ public abstract static class Read<T> extends PTransform<PBegin, PCollection<T>>
 
     abstract MatchConfiguration getMatchConfiguration();
 
-    abstract @Nullable Class<T> getRecordClass();
+    abstract Class<T> getRecordClass();
 
-    abstract @Nullable Schema getSchema();
+    abstract Schema getSchema();
 
     abstract boolean getInferBeamSchema();
 
@@ -761,8 +758,7 @@ public Read<T> withDatumReaderFactory(AvroSource.DatumReaderFactory<T> readerFac
     @Override
     @SuppressWarnings("unchecked")
     public PCollection<T> expand(PBegin input) {
-      checkNotNull(getFilepattern(), "filepattern");
-      checkNotNull(getSchema(), "schema");
+      ValueProvider<String> filepattern = checkStateNotNull(getFilepattern(), "filepattern");
 
       if (getMatchConfiguration().getWatchInterval() == null && !getHintMatchesManyFiles()) {
         PCollection<T> read =
@@ -770,7 +766,7 @@ public PCollection<T> expand(PBegin input) {
                 "Read",
                 org.apache.beam.sdk.io.Read.from(
                     createSource(
-                        getFilepattern(),
+                        filepattern,
                         getMatchConfiguration().getEmptyMatchTreatment(),
                         getRecordClass(),
                         getSchema(),
@@ -784,15 +780,21 @@ public PCollection<T> expand(PBegin input) {
           (getRecordClass() == GenericRecord.class)
               ? (ReadFiles<T>) readFilesGenericRecords(getSchema())
               : readFiles(getRecordClass());
+      AvroSource.DatumReaderFactory<T> readerFactory = getDatumReaderFactory();
+      if (readerFactory != null) {
+        readFiles = readFiles.withDatumReaderFactory(readerFactory);
+      }
+      Coder<T> coder = getCoder();
+      if (coder != null) {
+        readFiles = readFiles.withCoder(coder);
+      }
       return input
-          .apply("Create filepattern", Create.ofProvider(getFilepattern(), StringUtf8Coder.of()))
+          .apply("Create filepattern", Create.ofProvider(filepattern, StringUtf8Coder.of()))
           .apply("Match All", FileIO.matchAll().withConfiguration(getMatchConfiguration()))
           .apply(
               "Read Matches",
               FileIO.readMatches().withDirectoryTreatment(DirectoryTreatment.PROHIBIT))
-          .apply(
-              "Via ReadFiles",
-              readFiles.withDatumReaderFactory(getDatumReaderFactory()).withCoder(getCoder()));
+          .apply("Via ReadFiles", readFiles);
     }
 
     @Override
@@ -842,9 +844,9 @@ private static <T> AvroSource<T> createSource(
   public abstract static class ReadFiles<T>
       extends PTransform<PCollection<ReadableFile>, PCollection<T>> {
 
-    abstract @Nullable Class<T> getRecordClass();
+    abstract Class<T> getRecordClass();
 
-    abstract @Nullable Schema getSchema();
+    abstract Schema getSchema();
 
     abstract boolean getUsesReshuffle();
 
@@ -924,7 +926,6 @@ public ReadFiles<T> withDatumReaderFactory(AvroSource.DatumReaderFactory<T> fact
 
     @Override
     public PCollection<T> expand(PCollection<ReadableFile> input) {
-      checkNotNull(getSchema(), "schema");
       Coder<T> coder =
           Optional.ofNullable(getCoder()).orElse(AvroCoder.of(getRecordClass(), getSchema()));
       PCollection<T> read =
@@ -965,9 +966,9 @@ public void populateDisplayData(DisplayData.Builder builder) {
   public abstract static class ReadAll<T> extends PTransform<PCollection<String>, PCollection<T>> {
     abstract MatchConfiguration getMatchConfiguration();
 
-    abstract @Nullable Class<T> getRecordClass();
+    abstract Class<T> getRecordClass();
 
-    abstract @Nullable Schema getSchema();
+    abstract Schema getSchema();
 
     abstract long getDesiredBundleSizeBytes();
 
@@ -1025,7 +1026,6 @@ public ReadAll<T> withBeamSchemas(boolean withBeamSchemas) {
 
     @Override
     public PCollection<T> expand(PCollection<String> input) {
-      checkNotNull(getSchema(), "schema");
       PCollection<T> read =
           input
               .apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
@@ -1052,13 +1052,13 @@ private static class CreateSourceFn<T>
     private final Class<T> recordClass;
     private final Supplier<Schema> schemaSupplier;
     private final Coder<T> coder;
-    private final AvroSource.DatumReaderFactory<T> readerFactory;
+    private final AvroSource.@Nullable DatumReaderFactory<T> readerFactory;
 
     CreateSourceFn(
         Class<T> recordClass,
         String jsonSchema,
         Coder<T> coder,
-        AvroSource.DatumReaderFactory<T> readerFactory) {
+        AvroSource.@Nullable DatumReaderFactory<T> readerFactory) {
       this.recordClass = recordClass;
       this.schemaSupplier =
           Suppliers.memoize(
@@ -1158,18 +1158,18 @@ public Parse<T> withHintMatchesManyFiles() {
 
     @Override
     public PCollection<T> expand(PBegin input) {
-      checkNotNull(getFilepattern(), "filepattern");
+      ValueProvider<String> filepattern = checkStateNotNull(getFilepattern(), "filepattern");
       Coder<T> coder = inferCoder(getCoder(), getParseFn(), input.getPipeline().getCoderRegistry());
 
       if (getMatchConfiguration().getWatchInterval() == null && !getHintMatchesManyFiles()) {
         return input.apply(
             org.apache.beam.sdk.io.Read.from(
-                AvroSource.from(getFilepattern()).withParseFn(getParseFn(), coder)));
+                AvroSource.from(filepattern).withParseFn(getParseFn(), coder)));
       }
 
       // All other cases go through FileIO + ParseFilesGenericRecords.
       return input
-          .apply("Create filepattern", Create.ofProvider(getFilepattern(), StringUtf8Coder.of()))
+          .apply("Create filepattern", Create.ofProvider(filepattern, StringUtf8Coder.of()))
           .apply("Match All", FileIO.matchAll().withConfiguration(getMatchConfiguration()))
           .apply(
               "Read Matches",
@@ -1377,12 +1377,15 @@ public ParseAll<T> withDesiredBundleSizeBytes(long desiredBundleSizeBytes) {
 
     @Override
     public PCollection<T> expand(PCollection<String> input) {
+      ParseFiles<T> parseFiles = parseFilesGenericRecords(getParseFn());
+      Coder<T> coder = getCoder();
+      if (coder != null) {
+        parseFiles = parseFiles.withCoder(coder);
+      }
       return input
           .apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
           .apply(FileIO.readMatches().withDirectoryTreatment(DirectoryTreatment.PROHIBIT))
-          .apply(
-              "Parse all via FileBasedSource",
-              parseFilesGenericRecords(getParseFn()).withCoder(getCoder()));
+          .apply("Parse all via FileBasedSource", parseFiles);
     }
 
     @Override
@@ -1404,6 +1407,13 @@ public abstract static class TypedWrite<UserT, DestinationT, OutputT>
     static final SerializableAvroCodecFactory DEFAULT_SERIALIZABLE_CODEC =
         new SerializableAvroCodecFactory(DEFAULT_CODEC);
 
+    private static final String SCHEMA_REQUIRED =
+        "Unless using DynamicDestinations, .withSchema() is required.";
+    private static final String FORMAT_FUNCTION_REQUIRED =
+        "Unless using DynamicDestinations, .withFormatFunction() is required.";
+    private static final String FILENAME_PREFIX_REQUIRED =
+        "Need to set either the filename prefix or the tempDirectory of a AvroIO.Write transform.";
+
     abstract @Nullable SerializableFunction<UserT, OutputT> getFormatFunction();
 
     abstract @Nullable ValueProvider<ResourceId> getFilenamePrefix();
@@ -1753,7 +1763,7 @@ DynamicAvroDestinations<UserT, DestinationT, OutputT> resolveDynamicDestinations
         if (usedFilenamePolicy == null) {
           usedFilenamePolicy =
               DefaultFilenamePolicy.fromStandardParameters(
-                  getFilenamePrefix(),
+                  checkStateNotNull(getFilenamePrefix(), FILENAME_PREFIX_REQUIRED),
                   getShardTemplate(),
                   getFilenameSuffix(),
                   getWindowedWrites());
@@ -1762,10 +1772,10 @@ DynamicAvroDestinations<UserT, DestinationT, OutputT> resolveDynamicDestinations
             (DynamicAvroDestinations<UserT, DestinationT, OutputT>)
                 constantDestinations(
                     usedFilenamePolicy,
-                    getSchema(),
+                    checkStateNotNull(getSchema(), SCHEMA_REQUIRED),
                     getMetadata(),
                     getCodec().getCodec(),
-                    getFormatFunction(),
+                    checkStateNotNull(getFormatFunction(), FORMAT_FUNCTION_REQUIRED),
                     getDatumWriterFactory());
       }
       return dynamicDestinations;
@@ -1774,9 +1784,7 @@ DynamicAvroDestinations<UserT, DestinationT, OutputT> resolveDynamicDestinations
     @Override
     public WriteFilesResult<DestinationT> expand(PCollection<UserT> input) {
       checkArgument(
-          getFilenamePrefix() != null || getTempDirectory() != null,
-          "Need to set either the filename prefix or the tempDirectory of a AvroIO.Write "
-              + "transform.");
+          getFilenamePrefix() != null || getTempDirectory() != null, FILENAME_PREFIX_REQUIRED);
       if (getFilenamePolicy() != null) {
         checkArgument(
             getShardTemplate() == null && getFilenameSuffix() == null,
@@ -1789,13 +1797,12 @@ public WriteFilesResult<DestinationT> expand(PCollection<UserT> input) {
             "A format function should not be specified "
                 + "with DynamicDestinations. Use DynamicDestinations.formatRecord instead");
       } else {
-        checkArgument(
-            getSchema() != null, "Unless using DynamicDestinations, .withSchema() is required.");
+        checkArgument(getSchema() != null, SCHEMA_REQUIRED);
       }
 
       ValueProvider<ResourceId> tempDirectory = getTempDirectory();
       if (tempDirectory == null) {
-        tempDirectory = getFilenamePrefix();
+        tempDirectory = checkStateNotNull(getFilenamePrefix(), FILENAME_PREFIX_REQUIRED);
       }
       WriteFiles<UserT, DestinationT, OutputT> write =
           WriteFiles.to(
@@ -1813,11 +1820,13 @@ public WriteFilesResult<DestinationT> expand(PCollection<UserT> input) {
       if (getNoSpilling()) {
         write = write.withNoSpilling();
       }
-      if (getMaxNumWritersPerBundle() != null) {
-        write = write.withMaxNumWritersPerBundle(getMaxNumWritersPerBundle());
+      Integer maxNumWritersPerBundle = getMaxNumWritersPerBundle();
+      if (maxNumWritersPerBundle != null) {
+        write = write.withMaxNumWritersPerBundle(maxNumWritersPerBundle);
       }
-      if (getBadRecordErrorHandler() != null) {
-        write = write.withBadRecordErrorHandler(getBadRecordErrorHandler());
+      ErrorHandler<BadRecord, ?> badRecordErrorHandler = getBadRecordErrorHandler();
+      if (badRecordErrorHandler != null) {
+        write = write.withBadRecordErrorHandler(badRecordErrorHandler);
       }
       return input.apply("Write", write);
     }
@@ -2020,13 +2029,13 @@ public interface RecordFormatter<ElementT> extends Serializable {
 
   private static class FormattedDatumWriter<ElementT> implements DatumWriter<ElementT> {
     private Schema root;
-    private RecordFormatter<ElementT> formatter;
-    private GenericDatumWriter<GenericRecord> writer;
+    private final RecordFormatter<ElementT> formatter;
+    private final GenericDatumWriter<GenericRecord> writer;
 
     public FormattedDatumWriter(Schema schema, RecordFormatter<ElementT> formatter) {
       this.formatter = formatter;
       this.writer = new GenericDatumWriter<>(schema);
-      setSchema(schema);
+      this.root = schema;
     }
 
     @Override
@@ -2101,7 +2110,7 @@ public abstract static class Sink<ElementT> implements FileIO.Sink<ElementT> {
     @Deprecated
     abstract @Nullable RecordFormatter<ElementT> getRecordFormatter();
 
-    abstract @Nullable String getJsonSchema();
+    abstract String getJsonSchema();
 
     abstract Map<String, Object> getMetadata();
 
@@ -2151,47 +2160,53 @@ public Sink<ElementT> withDatumWriterFactory(
       return toBuilder().setDatumWriterFactory(datumWriterFactory).build();
     }
 
-    private transient @Nullable Schema schema;
     private transient @Nullable DataFileWriter<ElementT> writer;
 
     @Override
     public void open(WritableByteChannel channel) throws IOException {
-      this.schema = new Schema.Parser().parse(getJsonSchema());
+      Schema schema = new Schema.Parser().parse(getJsonSchema());
+      RecordFormatter<ElementT> recordFormatter = getRecordFormatter();
+      AvroSink.DatumWriterFactory<ElementT> datumWriterFactory = getDatumWriterFactory();
       DatumWriter<ElementT> datumWriter;
-      if (getRecordFormatter() != null) {
-        datumWriter = new FormattedDatumWriter<>(schema, getRecordFormatter());
-      } else if (getDatumWriterFactory() != null) {
-        datumWriter = getDatumWriterFactory().apply(schema);
+      if (recordFormatter != null) {
+        datumWriter = new FormattedDatumWriter<>(schema, recordFormatter);
+      } else if (datumWriterFactory != null) {
+        datumWriter = datumWriterFactory.apply(schema);
       } else {
         datumWriter = new ReflectDatumWriter<>(schema);
       }
-      writer = new DataFileWriter<>(datumWriter);
-      writer.setCodec(getCodec().getCodec());
+      DataFileWriter<ElementT> dataFileWriter = new DataFileWriter<>(datumWriter);
+      dataFileWriter.setCodec(getCodec().getCodec());
       for (Map.Entry<String, Object> entry : getMetadata().entrySet()) {
         Object v = entry.getValue();
         if (v instanceof String) {
-          writer.setMeta(entry.getKey(), (String) v);
+          dataFileWriter.setMeta(entry.getKey(), (String) v);
         } else if (v instanceof Long) {
-          writer.setMeta(entry.getKey(), (Long) v);
+          dataFileWriter.setMeta(entry.getKey(), (Long) v);
         } else if (v instanceof byte[]) {
-          writer.setMeta(entry.getKey(), (byte[]) v);
+          dataFileWriter.setMeta(entry.getKey(), (byte[]) v);
         } else {
           throw new IllegalStateException(
               "Metadata value type must be one of String, Long, or byte[]. Found "
                   + v.getClass().getSimpleName());
         }
       }
-      writer.create(schema, Channels.newOutputStream(channel));
+      dataFileWriter.create(schema, Channels.newOutputStream(channel));
+      writer = dataFileWriter;
     }
 
     @Override
     public void write(ElementT element) throws IOException {
-      writer.append(element);
+      writer().append(element);
     }
 
     @Override
     public void flush() throws IOException {
-      writer.flush();
+      writer().flush();
+    }
+
+    private DataFileWriter<ElementT> writer() {
+      return checkStateNotNull(writer, "open() has not been called");
     }
   }
 
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
index 08a9f3a2946b..85431c4d9fb7 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSchemaIOProvider.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.extensions.avro.io;
 
+import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
+
 import com.google.auto.service.AutoService;
 import java.io.Serializable;
 import org.apache.avro.generic.GenericRecord;
@@ -45,9 +47,6 @@
  */
 @Internal
 @AutoService(SchemaIOProvider.class)
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class AvroSchemaIOProvider implements SchemaIOProvider {
   /** Returns an id that uniquely represents this IO. */
   @Override
@@ -69,8 +68,10 @@ public Schema configurationSchema() {
    * resides there, and some IO-specific configuration object.
    */
   @Override
-  public AvroSchemaIO from(String location, Row configuration, Schema dataSchema) {
-    return new AvroSchemaIO(location, dataSchema, configuration);
+  public AvroSchemaIO from(String location, Row configuration, @Nullable Schema dataSchema) {
+    // requiresDataSchema() is true, so callers must supply a data schema
+    return new AvroSchemaIO(
+        location, checkArgumentNotNull(dataSchema, "dataSchema is required"), configuration);
   }
 
   @Override
@@ -94,11 +95,8 @@ private static class AvroSchemaIO implements SchemaIO, Serializable {
     private AvroSchemaIO(String location, Schema dataSchema, Row configuration) {
       this.dataSchema = dataSchema;
       this.location = location;
-      if (configuration.getInt64("writeWindowSizeSeconds") != null) {
-        windowSize = Duration.standardSeconds(configuration.getInt64("writeWindowSizeSeconds"));
-      } else {
-        windowSize = null;
-      }
+      Long windowSizeSeconds = configuration.getInt64("writeWindowSizeSeconds");
+      windowSize = windowSizeSeconds == null ? null : Duration.standardSeconds(windowSizeSeconds);
     }
 
     @Override
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
index a6e6353235c6..2ec4f429dde8 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSink.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.extensions.avro.io;
 
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+
 import java.io.Serializable;
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
@@ -34,9 +36,6 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /** A {@link FileBasedSink} for Avro files. */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class AvroSink<UserT, DestinationT, OutputT>
     extends FileBasedSink<UserT, DestinationT, OutputT> {
   private final Class<OutputT> type;
@@ -121,33 +120,38 @@ protected void prepareWrite(WritableByteChannel channel) throws Exception {
               .orElse(AvroDatumFactory.of(type))
               .apply(schema);
 
-      dataFileWriter = new DataFileWriter<>(datumWriter).setCodec(codec);
+      DataFileWriter<OutputT> writer = new DataFileWriter<>(datumWriter).setCodec(codec);
       for (Map.Entry<String, Object> entry : metadata.entrySet()) {
         Object v = entry.getValue();
         if (v instanceof String) {
-          dataFileWriter.setMeta(entry.getKey(), (String) v);
+          writer.setMeta(entry.getKey(), (String) v);
         } else if (v instanceof Long) {
-          dataFileWriter.setMeta(entry.getKey(), (Long) v);
+          writer.setMeta(entry.getKey(), (Long) v);
         } else if (v instanceof byte[]) {
-          dataFileWriter.setMeta(entry.getKey(), (byte[]) v);
+          writer.setMeta(entry.getKey(), (byte[]) v);
         } else {
           throw new IllegalStateException(
               "Metadata value type must be one of String, Long, or byte[]. Found "
                   + v.getClass().getSimpleName());
         }
       }
-      dataFileWriter.setSyncInterval(syncInterval);
-      dataFileWriter.create(schema, Channels.newOutputStream(channel));
+      writer.setSyncInterval(syncInterval);
+      writer.create(schema, Channels.newOutputStream(channel));
+      dataFileWriter = writer;
     }
 
     @Override
     public void write(OutputT value) throws Exception {
-      dataFileWriter.append(value);
+      dataFileWriter().append(value);
     }
 
     @Override
     protected void finishWrite() throws Exception {
-      dataFileWriter.flush();
+      dataFileWriter().flush();
+    }
+
+    private DataFileWriter<OutputT> dataFileWriter() {
+      return checkStateNotNull(dataFileWriter, "prepareWrite() has not been called");
     }
   }
 }
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
index 6d65c800c0a2..bd21a0d14124 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/AvroSource.java
@@ -18,9 +18,9 @@
 package org.apache.beam.sdk.extensions.avro.io;
 
 import static org.apache.beam.sdk.io.FileBasedSource.Mode.SINGLE_FILE_OR_SUBRANGE;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
-import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -125,9 +125,6 @@
  */
 // CHECKSTYLE.ON: JavadocStyle
 
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class AvroSource<T> extends BlockBasedSource<T> {
   // Default minimum bundle size (chosen as two default-size Avro blocks to attempt to
   // ensure that every source has at least one block of records).
@@ -174,7 +171,8 @@ private Mode(
         @Nullable Coder<T> outputCoder,
         @Nullable DatumReaderFactory<?> readerFactory) {
       this.type = type;
-      this.readerSchemaString = internSchemaString(readerSchemaString);
+      this.readerSchemaString =
+          readerSchemaString == null ? null : internSchemaString(readerSchemaString);
       this.parseFn = parseFn;
       this.outputCoder = outputCoder;
       this.readerFactory = readerFactory;
@@ -182,21 +180,24 @@ private Mode(
 
     private void readObject(ObjectInputStream is) throws IOException, ClassNotFoundException {
       is.defaultReadObject();
-      readerSchemaString = internSchemaString(readerSchemaString);
+      String schemaString = readerSchemaString;
+      readerSchemaString = schemaString == null ? null : internSchemaString(schemaString);
     }
 
     private Coder<T> getOutputCoder() {
       if (parseFn == null && outputCoder == null) {
+        // validate() rejects a null readerSchemaString when there is no parse fn
+        Schema readerSchema =
+            internOrParseSchemaString(checkStateNotNull(readerSchemaString, "readerSchemaString"));
         if (readerFactory != null && readerFactory instanceof AvroDatumFactory) {
           // create custom AvroCoder from the AvroDatumFactory
-          return AvroCoder.of(
-              (AvroDatumFactory<T>) readerFactory, internOrParseSchemaString(readerSchemaString));
+          return AvroCoder.of((AvroDatumFactory<T>) readerFactory, readerSchema);
         } else {
           // fallback with default avro coder for the type & schema
-          return AvroCoder.of((Class<T>) type, internOrParseSchemaString(readerSchemaString));
+          return AvroCoder.of((Class<T>) type, readerSchema);
         }
       } else {
-        return outputCoder;
+        return checkStateNotNull(outputCoder, "outputCoder is required when using a parse fn");
       }
     }
 
@@ -476,7 +477,7 @@ static AvroMetadata readMetadataFromFile(ResourceId fileResource) throws IOExcep
     String schemaString = null;
     byte[] syncMarker;
     try (InputStream stream = Channels.newInputStream(FileSystems.open(fileResource))) {
-      BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(stream, null);
+      BinaryDecoder decoder = newBinaryDecoder(stream);
 
       // The header of an object container file begins with a four-byte magic number, followed
       // by the file metadata (including the schema and codec), encoded as a map. Finally, the
@@ -519,8 +520,16 @@ static AvroMetadata readMetadataFromFile(ResourceId fileResource) throws IOExcep
       syncMarker = new byte[DataFileConstants.SYNC_SIZE];
       decoder.readFixed(syncMarker);
     }
-    checkState(schemaString != null, "No schema present in Avro file metadata %s", fileResource);
-    return new AvroMetadata(syncMarker, codec, schemaString);
+    return new AvroMetadata(
+        syncMarker,
+        codec,
+        checkStateNotNull(
+            schemaString, "No schema present in Avro file metadata %s", fileResource));
+  }
+
+  @SuppressWarnings("nullness") // DecoderFactory accepts a null reuse but is unannotated
+  private static BinaryDecoder newBinaryDecoder(InputStream stream) {
+    return DecoderFactory.get().binaryDecoder(stream, null);
   }
 
   // A logical reference cache used to store schemas and schema strings to allow us to
@@ -583,12 +592,14 @@ static class AvroBlock<T> extends Block<T> {
 
     private final Iterator<?> iterator;
 
-    private final SerializableFunction<GenericRecord, T> parseFn;
+    private final @Nullable SerializableFunction<GenericRecord, T> parseFn;
 
     private final long numRecordsInBlock;
 
     AvroBlock(
-        Iterator<?> iter, SerializableFunction<GenericRecord, T> parseFn, long numRecordsInBlock) {
+        Iterator<?> iter,
+        @Nullable SerializableFunction<GenericRecord, T> parseFn,
+        long numRecordsInBlock) {
       this.iterator = iter;
       this.parseFn = parseFn;
       this.numRecordsInBlock = numRecordsInBlock;
@@ -596,7 +607,7 @@ static class AvroBlock<T> extends Block<T> {
 
     @Override
     public T getCurrentRecord() {
-      return currentRecord;
+      return checkStateNotNull(currentRecord, "readNextRecord() has not been called");
     }
 
     @Override
@@ -605,7 +616,7 @@ public boolean readNextRecord() {
         return false;
       }
 
-      Object record = iterator.next();
+      Object record = checkStateNotNull(iterator.next(), "Avro block contained a null record");
       currentRecord = (parseFn == null) ? ((T) record) : parseFn.apply((GenericRecord) record);
       currentRecordIndex++;
       return true;
@@ -699,34 +710,38 @@ public synchronized AvroSource<T> getCurrentSource() {
     // Postcondition: same as above, but for the new current (formerly next) block.
     @Override
     public boolean readNextBlock() {
-      if (!dataFileReader.hasNext()) {
+      DataFileReader<?> reader = dataFileReader();
+      if (!reader.hasNext()) {
         return false;
       }
 
       long headerLength =
-          (long) VarInt.getLength(dataFileReader.getBlockCount())
-              + VarInt.getLength(dataFileReader.getBlockSize())
+          (long) VarInt.getLength(reader.getBlockCount())
+              + VarInt.getLength(reader.getBlockSize())
               + DataFileConstants.SYNC_SIZE;
 
       currentBlock =
-          new AvroBlock<>(
-              dataFileReader, getCurrentSource().mode.parseFn, dataFileReader.getBlockCount());
+          new AvroBlock<>(reader, getCurrentSource().mode.parseFn, reader.getBlockCount());
 
       // Atomically update both the position and offset of the new block.
       synchronized (progressLock) {
-        currentBlockOffset = dataFileReader.previousSync();
+        currentBlockOffset = reader.previousSync();
         // Total block size includes the header, block content, and trailing sync marker.
-        currentBlockSizeBytes = dataFileReader.getBlockSize() + headerLength;
+        currentBlockSizeBytes = reader.getBlockSize() + headerLength;
       }
 
       return true;
     }
 
     @Override
-    public AvroBlock<T> getCurrentBlock() {
+    public @Nullable AvroBlock<T> getCurrentBlock() {
       return currentBlock;
     }
 
+    private DataFileReader<?> dataFileReader() {
+      return checkStateNotNull(dataFileReader, "startReading() has not been called");
+    }
+
     @Override
     public long getCurrentBlockOffset() {
       synchronized (progressLock) {
@@ -773,23 +788,33 @@ protected void startReading(ReadableByteChannel channel) throws IOException {
       }
 
       DatumReader<?> reader =
-          Optional.<DatumReaderFactory<?>>ofNullable(this.getCurrentSource().mode.readerFactory)
-              .orElse(AvroDatumFactory.of(this.getCurrentSource().mode.type))
-              .apply(readerSchema, readerSchema);
+          newDatumReader(
+              Optional.<DatumReaderFactory<?>>ofNullable(this.getCurrentSource().mode.readerFactory)
+                  .orElse(AvroDatumFactory.of(this.getCurrentSource().mode.type)),
+              readerSchema);
 
-      dataFileReader = new DataFileReader<>(seekableChannelInput, reader);
+      DataFileReader<?> fileReader = new DataFileReader<>(seekableChannelInput, reader);
 
       long startOffset = getCurrentSource().getStartOffset();
       if (startOffset != 0) {
         // the start offset may be in the middle of a sync marker, by rewinding SYNC_SIZE bytes we
         // ensure that we won't miss the block if so.
-        dataFileReader.sync(Math.max(0, startOffset - DataFileConstants.SYNC_SIZE));
+        fileReader.sync(Math.max(0, startOffset - DataFileConstants.SYNC_SIZE));
       }
 
       synchronized (progressLock) {
-        currentBlockOffset = dataFileReader.previousSync();
+        currentBlockOffset = fileReader.previousSync();
         currentBlockSizeBytes = 0;
       }
+      dataFileReader = fileReader;
+    }
+
+    // A null schema tells the factory to derive both the writer and the reader schema from the
+    // file header. DatumReaderFactory predates nullness annotations and cannot express that.
+    @SuppressWarnings("nullness")
+    private static <X> DatumReader<X> newDatumReader(
+        DatumReaderFactory<X> factory, @Nullable Schema schema) {
+      return factory.apply(schema, schema);
     }
   }
 }
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
index 5b3683c349c9..d92cef53b34c 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/ConstantAvroDestination.java
@@ -32,9 +32,6 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /** Always returns a constant {@link FilenamePolicy}, {@link Schema}, metadata, and codec. */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 class ConstantAvroDestination<UserT, OutputT>
     extends DynamicAvroDestinations<UserT, Void, OutputT> {
   private static class SchemaFunction implements Serializable, Function<String, Schema> {
@@ -51,7 +48,7 @@ public Schema apply(String input) {
   private final Map<String, Object> metadata;
   private final SerializableAvroCodecFactory codec;
   private final SerializableFunction<UserT, OutputT> formatFunction;
-  private final AvroSink.DatumWriterFactory<OutputT> datumWriterFactory;
+  private final AvroSink.@Nullable DatumWriterFactory<OutputT> datumWriterFactory;
 
   private class Metadata implements HasDisplayData {
     @Override
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
index 215d3b4dd5fc..fec1b7ed0161 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/io/SerializableAvroCodecFactory.java
@@ -40,9 +40,6 @@
  * A wrapper that allows {@link CodecFactory}s to be serialized using Java's standard serialization
  * mechanisms.
  */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 class SerializableAvroCodecFactory implements Externalizable {
   private static final long serialVersionUID = 7445324844109564303L;
   private static final List<String> noOptAvroCodecs =
@@ -64,7 +61,7 @@ public SerializableAvroCodecFactory(CodecFactory codecFactory) {
     this.codecFactory = codecFactory;
   }
 
-  private boolean checkIsSupportedCodec(CodecFactory codecFactory) {
+  private static boolean checkIsSupportedCodec(CodecFactory codecFactory) {
     final String codecStr = codecFactory.toString();
     return noOptAvroCodecs.contains(codecStr)
         || deflatePattern.matcher(codecStr).matches()
@@ -74,7 +71,7 @@ private boolean checkIsSupportedCodec(CodecFactory codecFactory) {
 
   @Override
   public void writeExternal(ObjectOutput out) throws IOException {
-    out.writeUTF(codecFactory.toString());
+    out.writeUTF(getCodec().toString());
   }
 
   @Override
@@ -91,32 +88,37 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
 
     Matcher deflateMatcher = deflatePattern.matcher(codecStr);
     if (deflateMatcher.find()) {
-      codecFactory = CodecFactory.deflateCodec(Integer.parseInt(deflateMatcher.group("level")));
+      codecFactory = CodecFactory.deflateCodec(matchedLevel(deflateMatcher));
       return;
     }
 
     Matcher xzMatcher = xzPattern.matcher(codecStr);
     if (xzMatcher.find()) {
-      codecFactory = CodecFactory.xzCodec(Integer.parseInt(xzMatcher.group("level")));
+      codecFactory = CodecFactory.xzCodec(matchedLevel(xzMatcher));
       return;
     }
 
     Matcher zstdMatcher = zstdPattern.matcher(codecStr);
     if (zstdMatcher.find()) {
-      codecFactory = CodecFactory.zstandardCodec(Integer.parseInt(zstdMatcher.group("level")));
+      codecFactory = CodecFactory.zstandardCodec(matchedLevel(zstdMatcher));
       return;
     }
 
     throw new IllegalStateException(codecStr + " is not supported");
   }
 
+  /** Reads the {@code level} group of a matcher that has just matched successfully. */
+  private static int matchedLevel(Matcher matcher) {
+    return Integer.parseInt(checkNotNull(matcher.group("level")));
+  }
+
   public CodecFactory getCodec() {
-    return codecFactory;
+    return checkNotNull(
+        codecFactory, "Inner CodecFactory is null, please use non default constructor");
   }
 
   @Override
   public String toString() {
-    checkNotNull(codecFactory, "Inner CodecFactory is null, please use non default constructor");
-    return codecFactory.toString();
+    return getCodec().toString();
   }
 }
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
index 0a82663c1771..1812a5738c3a 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroByteBuddyUtils.java
@@ -46,10 +46,7 @@
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
 
-@SuppressWarnings({
-  "nullness", // TODO(https://github.com/apache/beam/issues/20497)
-  "rawtypes"
-})
+@SuppressWarnings({"rawtypes"})
 class AvroByteBuddyUtils {
   private static final ByteBuddy BYTE_BUDDY = new ByteBuddy();
 
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
index 853d43ce3e7c..302c524722ae 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
@@ -35,6 +35,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -151,13 +152,12 @@
  *
  * is used.
  */
-@SuppressWarnings({
-  "nullness", // TODO(https://github.com/apache/beam/issues/20497)
-  "rawtypes"
-})
+@SuppressWarnings({"rawtypes"})
 public class AvroUtils {
   public static final String VERSION_AVRO =
-      org.apache.avro.Schema.class.getPackage().getImplementationVersion();
+      Optional.ofNullable(org.apache.avro.Schema.class.getPackage())
+          .map(Package::getImplementationVersion)
+          .orElse("");
   private static final ForLoadedType BYTES = new ForLoadedType(byte[].class);
   private static final ForLoadedType JAVA_INSTANT = new ForLoadedType(java.time.Instant.class);
   private static final ForLoadedType JAVA_LOCALE_DATE =
@@ -196,6 +196,12 @@ private static void builderSet(
       builder.set(fieldName, castToNonNull(value));
     }
 
+    private static org.apache.avro.Schema.Field newFieldWithoutDefault(
+        String name, org.apache.avro.Schema schema, String doc) {
+      // a null defaultValue means "no default", but the parameter lacks the @Nullable annotation
+      return new org.apache.avro.Schema.Field(name, schema, doc, castToNonNull(null));
+    }
+
     private static Object createFixed(
         @Nullable Object old, byte[] bytes, org.apache.avro.Schema schema) {
       // old is tolerated when null, due to an instanceof check
@@ -528,8 +534,8 @@ public static Field toBeamField(org.apache.avro.Schema.Field field) {
   public static org.apache.avro.Schema.Field toAvroField(Field field, String namespace) {
     org.apache.avro.Schema fieldSchema =
         getFieldSchema(field.getType(), field.getName(), namespace);
-    return new org.apache.avro.Schema.Field(
-        field.getName(), fieldSchema, field.getDescription(), (Object) null);
+    return NullnessCheckerWorkarounds.newFieldWithoutDefault(
+        field.getName(), fieldSchema, field.getDescription());
   }
 
   private AvroUtils() {}
@@ -1437,8 +1443,8 @@ private static org.apache.avro.Schema getFieldSchema(
     }
   }
 
-  private static Object convertLogicalType(
-      @PolyNull Object value,
+  private static @Nullable Object convertLogicalType(
+      @Nonnull Object value,
       @Nonnull org.apache.avro.Schema avroSchema,
       @Nonnull FieldType fieldType,
       @Nonnull GenericData genericData) {
@@ -1447,11 +1453,11 @@ private static Object convertLogicalType(
     // TODO: Remove this workaround once Avro is upgraded to 1.12+ where timestamp-nanos
     if (TIMESTAMP_NANOS_LOGICAL_TYPE.equals(type.type.getProp("logicalType"))) {
       if (type.type.getType() == org.apache.avro.Schema.Type.LONG) {
-        Long nanos = (Long) value;
+        long nanos = (Long) value;
         // Check if Beam expects Timestamp logical type
         if (fieldType.getTypeName() == TypeName.LOGICAL_TYPE
             && org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER.equals(
-                fieldType.getLogicalType().getIdentifier())) {
+                checkNotNull(fieldType.getLogicalType()).getIdentifier())) {
           long seconds = Math.floorDiv(nanos, 1_000_000_000L);
           long nanoAdjustment = Math.floorMod(nanos, 1_000_000_000L);
           return java.time.Instant.ofEpochSecond(seconds, nanoAdjustment);
@@ -1473,7 +1479,10 @@ private static Object convertLogicalType(
     if (conversion != null) {
       convertedType = conversion.getConvertedType();
       if (convertedType.isInstance(value)) {
-        rawType = Conversions.convertToRawType(value, avroSchema, logicalType, conversion);
+        // type.type rather than avroSchema: Conversions.convertToRawType switches on the schema
+        // type and silently returns the value unconverted for a UNION, so a nullable field would
+        // never get converted.
+        rawType = Conversions.convertToRawType(value, type.type, logicalType, conversion);
       }
     }
 
@@ -1786,8 +1795,8 @@ private static <T> T checkRawType(
       Object value,
       LogicalType logicalType,
       Object rawType,
-      Conversion<?> conversion,
-      Class<?> convertedType) {
+      @Nullable Conversion<?> conversion,
+      @Nullable Class<?> convertedType) {
     String msg =
         String.format(
             "Value %s of class %s is not a supported type for logical type %s (%s). "
diff --git a/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java b/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
index d2452452f14f..928cdbfe10bf 100644
--- a/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
+++ b/sdks/java/extensions/avro/src/test/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtilsTest.java
@@ -73,6 +73,7 @@
 import org.joda.time.DateTimeZone;
 import org.joda.time.Days;
 import org.joda.time.Instant;
+import org.joda.time.LocalDate;
 import org.joda.time.LocalTime;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -1112,10 +1113,90 @@ public void testGenericRecordToBeamRow() {
 
     // Alternatively, a timestamp-millis logical type can have a joda datum.
     genericRecord.put("timestampMillis", new DateTime(genericRecord.get("timestampMillis")));
-    row = AvroUtils.toBeamRowStrict(getGenericRecord(), null);
+    row = AvroUtils.toBeamRowStrict(genericRecord, null);
     assertEquals(getBeamRow(), row);
   }
 
+  @Test
+  public void testNullableLogicalTypeGenericRecordToBeamRow() {
+    org.apache.avro.Schema decimalSchema =
+        LogicalTypes.decimal(Integer.MAX_VALUE)
+            .addToSchema(org.apache.avro.Schema.create(org.apache.avro.Schema.Type.BYTES));
+    org.apache.avro.Schema avroSchema =
+        org.apache.avro.Schema.createRecord(
+            "topLevelRecord",
+            null,
+            null,
+            false,
+            Lists.newArrayList(
+                new org.apache.avro.Schema.Field(
+                    "date",
+                    ReflectData.makeNullable(
+                        LogicalTypes.date()
+                            .addToSchema(
+                                org.apache.avro.Schema.create(org.apache.avro.Schema.Type.INT))),
+                    "",
+                    (Object) null),
+                new org.apache.avro.Schema.Field(
+                    "timestampMillis",
+                    ReflectData.makeNullable(
+                        LogicalTypes.timestampMillis()
+                            .addToSchema(
+                                org.apache.avro.Schema.create(org.apache.avro.Schema.Type.LONG))),
+                    "",
+                    (Object) null),
+                new org.apache.avro.Schema.Field(
+                    "decimal", ReflectData.makeNullable(decimalSchema), "", (Object) null)));
+
+    Schema beamSchema =
+        Schema.builder()
+            .addNullableField("date", FieldType.DATETIME)
+            .addNullableField("timestampMillis", FieldType.DATETIME)
+            .addNullableField("decimal", FieldType.DECIMAL)
+            .build();
+    assertEquals(beamSchema, AvroUtils.toBeamSchema(avroSchema));
+
+    // Data written through a GenericData with logical type conversions registered carries the
+    // converted values (joda LocalDate/DateTime, BigDecimal) rather than the raw int/long/bytes.
+    GenericRecord converted =
+        new GenericRecordBuilder(avroSchema)
+            .set("date", new LocalDate(1979, 3, 14))
+            .set("timestampMillis", DATE_TIME)
+            .set("decimal", BIG_DECIMAL)
+            .build();
+    assertEquals(
+        Row.withSchema(beamSchema)
+            .addValues(new DateTime(1979, 3, 14, 0, 0, DateTimeZone.UTC), DATE_TIME, BIG_DECIMAL)
+            .build(),
+        AvroUtils.toBeamRowStrict(converted, beamSchema));
+
+    // The same fields holding their unconverted avro representations.
+    GenericRecord raw =
+        new GenericRecordBuilder(avroSchema)
+            .set("date", (int) java.time.LocalDate.of(1979, 3, 14).toEpochDay())
+            .set("timestampMillis", DATE_TIME.getMillis())
+            .set(
+                "decimal",
+                new Conversions.DecimalConversion()
+                    .toBytes(BIG_DECIMAL, decimalSchema, decimalSchema.getLogicalType()))
+            .build();
+    assertEquals(
+        Row.withSchema(beamSchema)
+            .addValues(new DateTime(1979, 3, 14, 0, 0, DateTimeZone.UTC), DATE_TIME, BIG_DECIMAL)
+            .build(),
+        AvroUtils.toBeamRowStrict(raw, beamSchema));
+
+    GenericRecord nulls =
+        new GenericRecordBuilder(avroSchema)
+            .set("date", null)
+            .set("timestampMillis", null)
+            .set("decimal", null)
+            .build();
+    assertEquals(
+        Row.withSchema(beamSchema).addValues(null, null, null).build(),
+        AvroUtils.toBeamRowStrict(nulls, beamSchema));
+  }
+
   @Test
   public void testGenericRecordToRowFunction() {
     SerializableUtils.ensureSerializable(AvroUtils.getGenericRecordToRowFunction(Schema.of()));