Addressing code review comments

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

Author: reta

flink-connector-opensearch-e2e-tests/src/main/java/org/apache/flink/streaming/tests/OpensearchAsyncSinkExample.java

@@ -29,9 +29,6 @@
 import org.apache.http.HttpHost;
 import org.opensearch.action.index.IndexRequest;
 
-import java.util.ArrayList;
-import java.util.List;
-
 /** End to end test for OpensearchAsyncSink. */
 public class OpensearchAsyncSinkExample {
 
@@ -62,12 +59,9 @@ public void flatMap(
                                     }
                                 });
 
-        List<HttpHost> httpHosts = new ArrayList<>();
-        httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));
-
         OpensearchAsyncSinkBuilder<Tuple2<String, String>> osSinkBuilder =
                 OpensearchAsyncSink.<Tuple2<String, String>>builder()
-                        .setHosts(new HttpHost("localhost:9200"))
+                        .setHosts(new HttpHost("localhost", 9200, "http"))
                         .setElementConverter(
                                 (element, context) ->
                                         new IndexRequest("my-index")

flink-connector-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchAsyncSink.java

@@ -54,13 +54,13 @@ public class OpensearchAsyncSink<InputT> extends AsyncSinkBase<InputT, DocSerdeR
      * Constructor creating an Opensearch async sink.
      *
      * @param maxBatchSize the maximum size of a batch of entries that may be sent
-     * @param maxInFlightRequests he maximum number of in flight requests that may exist, if any
+     * @param maxInFlightRequests the maximum number of in flight requests that may exist, if any
      *     more in flight requests need to be initiated once the maximum has been reached, then it
      *     will be blocked until some have completed
      * @param maxBufferedRequests the maximum number of elements held in the buffer, requests to add
      *     elements will be blocked while the number of elements in the buffer is at the maximum
-     * @param maxBatchSizeInBytes the maximum size of a batch of entries that may be sent to KDS
-     *     measured in bytes
+     * @param maxBatchSizeInBytes the maximum size of a batch of entries that may be sent to
+     *     OpenSearch measured in bytes
      * @param maxTimeInBufferMS the maximum amount of time an entry is allowed to live in the
      *     buffer, if any element reaches this age, the entire buffer will be flushed immediately
      * @param maxRecordSizeInBytes the maximum size of a record the sink will accept into the

flink-connector-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchAsyncSinkBuilder.java

@@ -27,6 +27,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -84,7 +85,7 @@ public OpensearchAsyncSinkBuilder<InputT> setElementConverter(
      */
     public OpensearchAsyncSinkBuilder<InputT> setHosts(HttpHost... hosts) {
         checkNotNull(hosts);
-        checkState(hosts.length > 0, "Hosts cannot be empty.");
+        checkArgument(hosts.length > 0, "Hosts cannot be empty.");
         this.hosts = Arrays.asList(hosts);
         return this;
     }
@@ -97,6 +98,7 @@ public OpensearchAsyncSinkBuilder<InputT> setHosts(HttpHost... hosts) {
      */
     public OpensearchAsyncSinkBuilder<InputT> setConnectionUsername(String username) {
         checkNotNull(username);
+        checkArgument(!username.trim().isEmpty(), "Username cannot be empty");
         this.username = username;
         return this;
     }
@@ -109,6 +111,7 @@ public OpensearchAsyncSinkBuilder<InputT> setConnectionUsername(String username)
      */
     public OpensearchAsyncSinkBuilder<InputT> setConnectionPassword(String password) {
         checkNotNull(password);
+        checkArgument(!password.trim().isEmpty(), "Password cannot be empty");
         this.password = password;
         return this;
     }
@@ -129,40 +132,66 @@ public OpensearchAsyncSinkBuilder<InputT> setConnectionPathPrefix(String prefix)
      * Sets the timeout for requesting the connection of the Opensearch cluster from the connection
      * manager.
      *
-     * @param timeout for the connection request
+     * @param timeout timeout for the connection request (in milliseconds)
      * @return this builder
      */
     public OpensearchAsyncSinkBuilder<InputT> setConnectionRequestTimeout(int timeout) {
-        checkState(timeout >= 0, "Connection request timeout must be larger than or equal to 0.");
+        checkArgument(
+                timeout >= 0, "Connection request timeout must be larger than or equal to 0.");
         this.connectionRequestTimeout = timeout;
         return this;
     }
 
     /**
      * Sets the timeout for establishing a connection of the Opensearch cluster.
      *
-     * @param timeout for the connection
+     * @param timeout timeout for the connection (in milliseconds)
      * @return this builder
      */
     public OpensearchAsyncSinkBuilder<InputT> setConnectionTimeout(int timeout) {
-        checkState(timeout >= 0, "Connection timeout must be larger than or equal to 0.");
+        checkArgument(timeout >= 0, "Connection timeout must be larger than or equal to 0.");
         this.connectionTimeout = timeout;
         return this;
     }
 
+    /**
+     * Sets the timeout for establishing a connection of the Opensearch cluster.
+     *
+     * @param timeout timeout for the connection (in milliseconds)
+     * @param timeUnit timeout time unit
+     * @return this builder
+     */
+    public OpensearchAsyncSinkBuilder<InputT> setConnectionTimeout(int timeout, TimeUnit timeUnit) {
+        checkNotNull(timeUnit, "TimeUnit cannot be null.");
+        return setConnectionTimeout((int) timeUnit.toMillis(timeout));
+    }
+
     /**
      * Sets the timeout for waiting for data or, put differently, a maximum period inactivity
      * between two consecutive data packets.
      *
-     * @param timeout for the socket
+     * @param timeout timeout for the socket (in milliseconds)
      * @return this builder
      */
     public OpensearchAsyncSinkBuilder<InputT> setSocketTimeout(int timeout) {
-        checkState(timeout >= 0, "Socket timeout must be larger than or equal to 0.");
+        checkArgument(timeout >= 0, "Socket timeout must be larger than or equal to 0.");
         this.socketTimeout = timeout;
         return this;
     }
 
+    /**
+     * Sets the timeout for waiting for data or, put differently, a maximum period inactivity
+     * between two consecutive data packets.
+     *
+     * @param timeout timeout for the socket
+     * @param timeUnit timeout time unit
+     * @return this builder
+     */
+    public OpensearchAsyncSinkBuilder<InputT> setSocketTimeout(int timeout, TimeUnit timeUnit) {
+        checkNotNull(timeUnit, "TimeUnit cannot be null.");
+        return setSocketTimeout((int) timeUnit.toMillis(timeout));
+    }
+
     /**
      * Allows to bypass the certificates chain validation and connect to insecure network endpoints
      * (for example, servers which use self-signed certificates).
@@ -177,8 +206,6 @@ public OpensearchAsyncSinkBuilder<InputT> setAllowInsecure(boolean allowInsecure
 
     @Override
     public OpensearchAsyncSink<InputT> build() {
-        checkArgument(!hosts.isEmpty(), "Hosts cannot be empty.");
-
         return new OpensearchAsyncSink<InputT>(
                 nonNullOrDefault(
                         getMaxBatchSize(),

flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchAsyncSinkBuilderTest.java

@@ -71,11 +71,11 @@ void testThrowIfElementConverterNotSet() {
     @Test
     void testThrowIfSetInvalidTimeouts() {
         assertThatThrownBy(() -> createEmptyBuilder().setConnectionRequestTimeout(-1).build())
-                .isInstanceOf(IllegalStateException.class);
+                .isInstanceOf(IllegalArgumentException.class);
         assertThatThrownBy(() -> createEmptyBuilder().setConnectionTimeout(-1).build())
-                .isInstanceOf(IllegalStateException.class);
+                .isInstanceOf(IllegalArgumentException.class);
         assertThatThrownBy(() -> createEmptyBuilder().setSocketTimeout(-1).build())
-                .isInstanceOf(IllegalStateException.class);
+                .isInstanceOf(IllegalArgumentException.class);
     }
 
     private OpensearchAsyncSinkBuilder<Object> createEmptyBuilder() {