fix rrio optionals

Author: stankiewicz

sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Call.java

@@ -362,7 +362,14 @@ private void backoffIfNeeded(BackOff backOff, Sleeper sleeper) {
   abstract static class Configuration<RequestT, ResponseT> implements Serializable {
 
     static <RequestT, ResponseT> Builder<RequestT, ResponseT> builder() {
-      return new AutoValue_Call_Configuration.Builder<RequestT, ResponseT>();
+      return new AutoValue_Call_Configuration.Builder<RequestT, ResponseT>()
+          .setShouldRepeat(false)
+          .setTimeout(RequestResponseIO.DEFAULT_TIMEOUT)
+          .setCallShouldBackoff(new NoopCallShouldBackoff<>())
+          .setSleeperSupplier((SerializableSupplier<Sleeper>) () -> Sleeper.DEFAULT)
+          .setBackOffSupplier(new DefaultSerializableBackoffSupplier())
+          .setMonitoringConfiguration(Monitoring.builder().build())
+          .setSetupTeardown(new NoopSetupTeardown());
     }
 
     /** The user custom code that converts a {@link RequestT} into a {@link ResponseT}. */
@@ -438,7 +445,7 @@ abstract static class Builder<RequestT, ResponseT> {
       /** See {@link Configuration#getTimeout}. */
       abstract Builder<RequestT, ResponseT> setTimeout(Duration value);
 
-      abstract Optional<Duration> getTimeout();
+      abstract Duration getTimeout();
 
       /** See {@link Configuration#getShouldRepeat}. */
       abstract Builder<RequestT, ResponseT> setShouldRepeat(boolean value);
@@ -449,49 +456,25 @@ abstract static class Builder<RequestT, ResponseT> {
       abstract Builder<RequestT, ResponseT> setCallShouldBackoff(
           CallShouldBackoff<ResponseT> value);
 
-      abstract Optional<CallShouldBackoff<ResponseT>> getCallShouldBackoff();
+      abstract CallShouldBackoff<ResponseT> getCallShouldBackoff();
 
       /** See {@link Configuration#getSleeperSupplier}. */
       abstract Builder<RequestT, ResponseT> setSleeperSupplier(SerializableSupplier<Sleeper> value);
 
-      abstract Optional<SerializableSupplier<Sleeper>> getSleeperSupplier();
+      abstract SerializableSupplier<Sleeper> getSleeperSupplier();
 
       /** See {@link Configuration#getBackOffSupplier}. */
       abstract Builder<RequestT, ResponseT> setBackOffSupplier(SerializableSupplier<BackOff> value);
 
-      abstract Optional<SerializableSupplier<BackOff>> getBackOffSupplier();
+      abstract SerializableSupplier<BackOff> getBackOffSupplier();
 
       abstract Builder<RequestT, ResponseT> setMonitoringConfiguration(Monitoring value);
 
-      abstract Optional<Monitoring> getMonitoringConfiguration();
+      abstract Monitoring getMonitoringConfiguration();
 
       abstract Configuration<RequestT, ResponseT> autoBuild();
 
       final Configuration<RequestT, ResponseT> build() {
-        if (!getSetupTeardown().isPresent()) {
-          setSetupTeardown(new NoopSetupTeardown());
-        }
-
-        if (!getTimeout().isPresent()) {
-          setTimeout(RequestResponseIO.DEFAULT_TIMEOUT);
-        }
-
-        if (!getCallShouldBackoff().isPresent()) {
-          setCallShouldBackoff(new NoopCallShouldBackoff<>());
-        }
-
-        if (!getSleeperSupplier().isPresent()) {
-          setSleeperSupplier((SerializableSupplier<Sleeper>) () -> Sleeper.DEFAULT);
-        }
-
-        if (!getBackOffSupplier().isPresent()) {
-          setBackOffSupplier(new DefaultSerializableBackoffSupplier());
-        }
-
-        if (!getMonitoringConfiguration().isPresent()) {
-          setMonitoringConfiguration(Monitoring.builder().build());
-        }
-
         return autoBuild();
       }
     }

sdks/java/io/rrio/src/main/java/org/apache/beam/io/requestresponse/Monitoring.java

@@ -20,7 +20,6 @@
 import com.google.auto.value.AutoValue;
 import java.io.Serializable;
 import java.util.Arrays;
-import java.util.Optional;
 import org.apache.beam.sdk.metrics.Counter;
 import org.apache.beam.sdk.metrics.Metric;
 import org.apache.beam.sdk.transforms.DoFn;
@@ -288,8 +287,6 @@ public abstract static class Builder {
 
     public abstract Builder setCountCacheWriteFailures(boolean value);
 
-    abstract Optional<Boolean> getCountRequests();
-
     public abstract Monitoring build();
   }
 }