for ES, when using aliases, the index name returned does not match th…

…e name of the alias, so offsets are wrong

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceTask.java

@@ -35,6 +35,7 @@
 import org.apache.kafka.connect.source.SourceTask;
 import org.apache.kafka.connect.source.SourceTaskContext;
 
+import com.github.castorm.kafka.connect.http.model.Offset;
 import com.github.castorm.kafka.connect.http.request.template.TemplateHttpRequestFactoryConfig;
 
 import lombok.extern.slf4j.Slf4j;
@@ -89,7 +90,7 @@ public List<SourceRecord> poll() throws InterruptedException {
     }
 
     private HttpSourceTaskSingleEndpoint getTaskForRecord(SourceRecord record) {
-        String endpoint = record.topic();
+        String endpoint = Offset.getEndpointFromPartition(record.sourcePartition());
         HttpSourceTaskSingleEndpoint task = tasks.get(endpoint);
         if (task == null) {
             throw new ConnectException("No HttpSourceTaskSingleEndpoint found for topic " + endpoint);

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/HttpSourceTaskSingleEndpoint.java

@@ -121,7 +121,7 @@ public List<SourceRecord> poll() throws InterruptedException {
                 .filter(recordFilterFactory.create(offset))
                 .collect(toList());
 
-        log.info("Request for offset {} yields {}/{} new records", offset.toMap(), unseenRecords.size(), records.size());
+        log.info("Request for offset {} yields {}/{} new records, endpoint {}", offset.toMap(), unseenRecords.size(), records.size(), endpoint);
 
         confirmationWindow = new ConfirmationWindow<>(extractOffsets(unseenRecords));
 

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/model/Offset.java

@@ -54,6 +54,10 @@ private Offset(String endpoint, Map<String, ?> properties) {
         return Map.of("endpoint", endpoint);
     }
 
+    public static String getEndpointFromPartition(Map<String, ?> partition) {
+        return partition.get("endpoint").toString();
+    }
+
     public static Offset of(Map<String, ?> properties, String endpoint) {
         return new Offset(endpoint, properties);
     }

kafka-connect-http/src/test/java/com/github/castorm/kafka/connect/http/StreamkapElasticConnectorTest.java

@@ -2,7 +2,7 @@
 
 import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
@@ -19,6 +19,7 @@
 import java.util.Map;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.kafka.connect.data.Struct;
@@ -31,6 +32,8 @@
 import org.opensearch.testcontainers.OpensearchContainer;
 import org.testcontainers.utility.DockerImageName;
 
+import com.github.castorm.kafka.connect.http.model.Offset;
+
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -82,26 +85,28 @@ private String loadTestData(int nbIndexes, int nbRecordsPerIndex) throws Excepti
         List<String> indexes = new ArrayList<>();
         for (int i = 0; i < nbIndexes; i++) {
             long now = new Date().getTime();
+            String indexName = "index-" + i;
             for (int j = 0; j < nbRecordsPerIndex; j++) {
                 String id = "" + i + "-" + j;
-                sendRequest("/index" + i + "/_doc/" + id, "PUT", 
+                sendRequest("/" + indexName + "/_doc/" + id, "PUT", 
                     "{ \"my_timestamp\": \"" + Instant.ofEpochMilli(now + j).toString() + "\", \"message\": \"Hello OpenSearch " + id + "\" }");
             }
-            indexes.add("index" + i);
+            indexes.add(indexName);
         }
         Thread.sleep(2000);//wait for ES to index the data and make it available for search
         return StringUtils.join(indexes, ",");
     }
 
-    private List<SourceRecord> runTasks(Map<String, String> config, int nbTasks) throws InterruptedException {
+    private List<SourceRecord> runTasks(Map<String, String> config, int nbTasks) throws Exception {
         return runTasks(config, nbTasks, 1);
     }
-    private List<SourceRecord> runTasks(Map<String, String> config, int nbTasks, int nbPolls) throws InterruptedException {
+    private List<SourceRecord> runTasks(Map<String, String> config, int nbTasks, int nbPolls) throws Exception {
         HttpSourceConnector connector = new HttpSourceConnector();
         connector.start(config);
         List<Map<String, String>> taskConfigs = connector.taskConfigs(2);
         ForkJoinPool pool = new ForkJoinPool(2);
         List<SourceRecord> records = Collections.synchronizedList(new ArrayList<>());
+        AtomicReference<Exception> taskEx = new AtomicReference<>();
         for (Map<String, String> taskConfig : taskConfigs) {
             HttpSourceTask task = new HttpSourceTask();
             task.initialize(getContext(emptyMap()));
@@ -113,17 +118,24 @@ private List<SourceRecord> runTasks(Map<String, String> config, int nbTasks, int
                         List<SourceRecord> polledRecords = task.poll();
                         records.addAll(polledRecords);
                         for (SourceRecord record : polledRecords) {
+                            assertEquals(record.topic(), 
+                                Offset.getEndpointFromPartition(
+                                    record.sourcePartition()).replaceAll("[^a-zA-Z0-9_]", "_"));
                             task.commitRecord(record, null);
                         }
                         task.commit();
-                    } catch (InterruptedException e) {
+                    } catch (Exception e) {
+                        taskEx.set(e);
                         throw new RuntimeException(e);
                     }
                 }
             });
         }
         pool.shutdown();
         pool.awaitTermination(1000, TimeUnit.SECONDS);
+        if (taskEx.get() != null) {
+            throw taskEx.get();
+        }
         log.info("Tasks done, got {} records", records.size());
 
         return records;