addressing review comments

Author: ankitk-me

runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaOptionsConfigAdapter.java

@@ -22,6 +22,7 @@
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaCorrelationConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaIdempotencyConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaOptionsConfig;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaOptionsConfigBuilder;
 import io.aklivity.zilla.runtime.binding.http.kafka.internal.HttpKafkaBinding;
 import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.String16FW;
 import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.String8FW;
@@ -111,16 +112,16 @@ public JsonObject adaptToJson(
     public OptionsConfig adaptFromJson(
         JsonObject object)
     {
-        String header = null;
-        String correlationId = null;
-        String replyTo = null;
+        HttpKafkaOptionsConfigBuilder<HttpKafkaOptionsConfig> builder = HttpKafkaOptionsConfig.builder();
 
         if (object.containsKey(IDEMPOTENCY_NAME))
         {
             JsonObject idempotency = object.getJsonObject(IDEMPOTENCY_NAME);
             if (idempotency.containsKey(IDEMPOTENCY_HEADER_NAME))
             {
-                header = idempotency.getString(IDEMPOTENCY_HEADER_NAME);
+                builder.idempotency()
+                    .header(idempotency.getString(IDEMPOTENCY_HEADER_NAME))
+                    .build();
             }
         }
 
@@ -131,26 +132,17 @@ public OptionsConfig adaptFromJson(
             {
                 JsonObject headers = correlation.getJsonObject(CORRELATION_HEADERS_NAME);
 
-                if (headers.containsKey(CORRELATION_HEADERS_REPLY_TO_NAME))
-                {
-                    replyTo = headers.getString(CORRELATION_HEADERS_REPLY_TO_NAME);
-                }
-
-                if (headers.containsKey(CORRELATION_HEADERS_CORRELATION_ID_NAME))
-                {
-                    correlationId = headers.getString(CORRELATION_HEADERS_CORRELATION_ID_NAME);
-                }
+                builder.correlation()
+                    .replyTo(headers.containsKey(CORRELATION_HEADERS_REPLY_TO_NAME)
+                        ? headers.getString(CORRELATION_HEADERS_REPLY_TO_NAME)
+                        : null)
+                    .correlationId(headers.containsKey(CORRELATION_HEADERS_CORRELATION_ID_NAME)
+                        ? headers.getString(CORRELATION_HEADERS_CORRELATION_ID_NAME)
+                        : null)
+                    .build();
             }
         }
 
-        return HttpKafkaOptionsConfig.builder()
-            .idempotency()
-                .header(header)
-                .build()
-            .correlation()
-                .replyTo(replyTo)
-                .correlationId(correlationId)
-                .build()
-            .build();
+        return builder.build();
     }
 }