style

yupeng9 · yupeng9 · commit b60c01b8af60 · 2025-03-26T20:38:14.000-07:00
Signed-off-by: Yupeng Fu &lt;yupeng@uber.com&gt;
diff --git a/plugins/ingestion-kinesis/src/internalClusterTest/java/org/opensearch/plugin/kinesis/IngestFromKinesisIT.java b/plugins/ingestion-kinesis/src/internalClusterTest/java/org/opensearch/plugin/kinesis/IngestFromKinesisIT.java
@@ -8,6 +8,15 @@
 
 package org.opensearch.plugin.kinesis;
 
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+
 import org.opensearch.action.admin.cluster.node.info.NodeInfo;
 import org.opensearch.action.admin.cluster.node.info.NodesInfoRequest;
 import org.opensearch.action.admin.cluster.node.info.NodesInfoResponse;
@@ -28,14 +37,6 @@
 import java.util.stream.Stream;
 
 import org.testcontainers.containers.localstack.LocalStackContainer;
-import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
-import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
-import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
-import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
-import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
-import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
-import software.amazon.awssdk.services.kinesis.model.Record;
-import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
 
 import static org.hamcrest.Matchers.is;
 import static org.awaitility.Awaitility.await;
@@ -107,10 +108,7 @@ public void testKinesisIngestion_RewindByOffset() throws InterruptedException {
         logger.info("Produced message with sequence number: {}", sequenceNumber);
         produceData("4", "name4", "21");
 
-        await()
-            .atMost(5, TimeUnit.SECONDS)
-            .until(() -> isRewinded(sequenceNumber));
-
+        await().atMost(5, TimeUnit.SECONDS).until(() -> isRewinded(sequenceNumber));
 
         // create an index with ingestion source from kinesis
         createIndex(
@@ -142,8 +140,9 @@ public void testKinesisIngestion_RewindByOffset() throws InterruptedException {
     }
 
     private boolean isRewinded(String sequenceNumber) {
-        DescribeStreamResponse describeStreamResponse =
-            kinesisClient.describeStream(DescribeStreamRequest.builder().streamName(streamName).build());
+        DescribeStreamResponse describeStreamResponse = kinesisClient.describeStream(
+            DescribeStreamRequest.builder().streamName(streamName).build()
+        );
 
         String shardId = describeStreamResponse.streamDescription().shards().get(0).shardId();
 
@@ -158,7 +157,9 @@ private boolean isRewinded(String sequenceNumber) {
         String shardIterator = iteratorResponse.shardIterator();
 
         // Use the iterator to read the record
-        GetRecordsRequest recordsRequest = GetRecordsRequest.builder().shardIterator(shardIterator).limit(1)  // Adjust as needed
+        GetRecordsRequest recordsRequest = GetRecordsRequest.builder()
+            .shardIterator(shardIterator)
+            .limit(1)  // Adjust as needed
             .build();
 
         GetRecordsResponse recordsResponse = kinesisClient.getRecords(recordsRequest);
