Enhance robustness of InternalOpenSearchSink tests through scenario-driven coverage, refactoring, and regression prevention:

- Add InternalOpenSearchSinkIntegrationTestConcreteIndex: covers the index
  creation path with a plain concrete date-based index (no alias)
- Extract shared tests into AbstractInternalOpenSearchSinkIntegrationTest
  using Template Method pattern: testPersistsAuditEventsToTarget and
  testAuditDocumentContainsMandatoryFields now run against both concrete-index
  and alias variants without duplication
- Refactor InternalOpenSearchSinkIntegrationTestAuditAlias: now extends the
  abstract base, removing 3 duplicate methods (countAuditDocs, generateAuditEvent,
  testAuditDocumentsViaAliasContainMandatoryFields); retains only the specialized
  test testRecognizesAuditTargetAsWriteAlias
- Add InternalOpenSearchSinkTest unit tests: covers race condition
  (ResourceAlreadyExistsException), generic cluster failures, and
  acknowledged=false scenarios that cannot be tested reliably on
  single-node embedded cluster
- Fix resource leak in testCreatesAuditIndexAutomatically

Closes #6145

Signed-off-by: Pietro Paolo Castagna <PietroPaolo.Castagna@gmail.com>

Author: pCastq

src/integrationTest/java/org/opensearch/security/auditlog/sink/AbstractInternalOpenSearchSinkIntegrationTest.java

@@ -34,17 +34,17 @@
  * Base class for {@link InternalOpenSearchSink} integration tests.
  *
  * <p>Holds test methods that are valid regardless of whether the audit target
- * is a date-based index or a write alias. Each subclass provides its own
- * {@link org.junit.ClassRule}-annotated {@link LocalCluster} and overrides
+ * is a plain concrete index or a write alias. Each subclass provides its own
+ * {@code @ClassRule} {@link LocalCluster} and overrides
  * {@link #cluster()} and {@link #auditTarget()} to supply the correct
  * cluster and query pattern for that configuration variant.</p>
  *
  * <p>Using a shared base class avoids duplicating {@code testPersistsAuditEventsToTarget}
  * and {@code testAuditDocumentContainsMandatoryFields} across the two concrete
  * integration-test classes that differ only in cluster setup.</p>
  *
- * @see InternalOpenSearchSinkIntegrationTest      default date-based index variant
- * @see InternalOpenSearchSinkIntegrationTestAuditAlias  write-alias variant
+ * @see InternalOpenSearchSinkIntegrationTestConcreteIndex concrete date-based index variant
+ * @see InternalOpenSearchSinkIntegrationTestAuditAlias    write-alias variant
  */
 abstract class AbstractInternalOpenSearchSinkIntegrationTest {
 
@@ -113,7 +113,7 @@ public void testPersistsAuditEventsToTarget() {
             generateAuditEvent("_cluster/health");
             generateAuditEvent("_cluster/stats");
 
-            await().atMost(10, SECONDS).pollInterval(100, MILLISECONDS).untilAsserted(() -> {
+            await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> {
                 refreshAuditTarget(client);
                 assertThat("At least 2 new audit events must be persisted",
                     countAuditDocs(client) - before, greaterThanOrEqualTo(2L));
@@ -130,7 +130,7 @@ public void testPersistsAuditEventsToTarget() {
      *   <li>{@code audit_category} — event classification (e.g., {@code GRANTED_PRIVILEGES})</li>
      *   <li>{@code audit_request_layer} — processing layer; expected value: {@code REST}</li>
      *   <li>{@code audit_request_origin} — origin layer; expected value: {@code REST}</li>
-     *   <li>{@code @timestamp} — ISO-8601 event timestamp</li>
+     *   <li>{@code @timestamp} — event timestamp</li>
      * </ul>
      *
      * <p><b>REST-Specific Fields:</b></p>
@@ -151,7 +151,7 @@ public void testAuditDocumentContainsMandatoryFields() {
 
             generateAuditEvent("_cluster/health");
 
-            await().atMost(10, SECONDS).pollInterval(100, MILLISECONDS).untilAsserted(() -> {
+            await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> {
                 refreshAuditTarget(client);
                 assertThat("Test must generate at least one new event",
                     countAuditDocs(client) - before, greaterThan(0L));
@@ -191,4 +191,4 @@ public void testAuditDocumentContainsMandatoryFields() {
                 doc.containsKey("audit_transport_request_type"), is(false));
         }
     }
-}
+}

src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTestAuditAlias.java

@@ -29,7 +29,6 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.awaitility.Awaitility.await;
 
@@ -42,8 +41,14 @@
  * Tests share a single cluster and use a before/after delta pattern so that
  * execution order does not matter.</p>
  *
+ * <p>Shared tests ({@code testPersistsAuditEventsToTarget} and
+ * {@code testAuditDocumentContainsMandatoryFields}) are inherited from
+ * {@link AbstractInternalOpenSearchSinkIntegrationTest}.</p>
+ *
+ * @see InternalOpenSearchSinkIntegrationTestConcreteIndex for the concrete date-based index variant
+ * @see InternalOpenSearchSinkTest for unit tests covering exception and race-condition branches
  */
-public class InternalOpenSearchSinkIntegrationTestAuditAlias {
+public class InternalOpenSearchSinkIntegrationTestAuditAlias extends AbstractInternalOpenSearchSinkIntegrationTest {
 
     private static final String AUDIT_ALIAS = "security-audit-write-alias";
     private static final String BACKING_INDEX = "security-audit-backend-000001";
@@ -52,27 +57,21 @@ public class InternalOpenSearchSinkIntegrationTestAuditAlias {
     static final TestAlias auditAlias = new TestAlias(AUDIT_ALIAS).on(backingIndex).writeIndex(backingIndex);
 
     @ClassRule
-    public static final LocalCluster cluster = new LocalCluster.Builder().clusterManager(ClusterManager.SINGLENODE)
+    public static final LocalCluster CLUSTER = new LocalCluster.Builder().clusterManager(ClusterManager.SINGLENODE)
         .nodeSettings(Map.of("plugins.security.audit.config.index", AUDIT_ALIAS))
         .internalAudit(new AuditConfiguration(true).filters(new AuditFilters().enabledRest(true).enabledTransport(false)))
         .indices(backingIndex)
         .aliases(auditAlias)
         .build();
 
-    /** Counts all audit documents reachable through the write alias. */
-    private long countAuditDocs(TestRestClient client) {
-        HttpResponse response = client.postJson(AUDIT_ALIAS + "/_search", """
-            {"query": {"match_all": {}}, "size": 0}
-            """);
-        response.assertStatusCode(200);
-        return response.getLongFromJsonBody("/hits/total/value");
+    @Override
+    LocalCluster cluster() {
+        return CLUSTER;
     }
 
-    /** Issues an authenticated REST GET that triggers an {@code AUTHENTICATED} audit event. */
-    private void generateAuditEvent(String path) {
-        try (TestRestClient restClient = cluster.getRestClient(cluster.getAdminCertificate())) {
-            restClient.get(path);
-        }
+    @Override
+    String auditTarget() {
+        return AUDIT_ALIAS;
     }
 
     /**
@@ -81,13 +80,23 @@ private void generateAuditEvent(String path) {
      *
      * <p>Generates one event, then checks that the alias still resolves to the
      * backing index and no spurious concrete index was created.</p>
+     *
+     * <p><b>Tested Code Path:</b> {@code metadata.hasAlias(indexName)} returns
+     * {@code true}, so the method logs a debug message and returns {@code true}
+     * immediately without attempting index creation.</p>
      */
     @Test
     public void testRecognizesAuditTargetAsWriteAlias() {
-        try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) {
+        try (TestRestClient client = CLUSTER.getRestClient(CLUSTER.getAdminCertificate())) {
             generateAuditEvent("_cluster/health");
 
-            await().until(() -> countAuditDocs(client) > 0);
+            await().until(() -> {
+                HttpResponse countResponse = client.postJson(AUDIT_ALIAS + "/_search", """
+                    {"query": {"match_all": {}}, "size": 0}
+                    """);
+                countResponse.assertStatusCode(200);
+                return countResponse.getLongFromJsonBody("/hits/total/value") > 0;
+            });
 
             HttpResponse aliasResponse = client.get("_alias/" + AUDIT_ALIAS);
             aliasResponse.assertStatusCode(200);
@@ -106,55 +115,4 @@ public void testRecognizesAuditTargetAsWriteAlias() {
             assertThat("Backing index must exist physically", indexExistsResponse.getStatusCode(), is(200));
         }
     }
-
-    /**
-     * The alias branch is invoked on every {@code doStore} call.
-     * Generates three distinct events and asserts all are persisted, confirming
-     * that repeated writes through the alias succeed.
-     */
-    @Test
-    public void testWritesEventsToAliasSuccessfully() {
-        try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) {
-            long before = countAuditDocs(client);
-
-            generateAuditEvent("_cluster/health");
-            generateAuditEvent("_cluster/stats");
-            generateAuditEvent("_nodes/info");
-
-            await().untilAsserted(
-                () -> assertThat("At least 3 events must be written through alias", countAuditDocs(client) - before, greaterThan(2L))
-            );
-        }
-    }
-
-    /**
-     * Documents written via the alias must contain the same mandatory audit fields
-     * as those written to a concrete index (category, timestamp, REST method/path,
-     * layer and origin). Transport-specific fields must be absent since transport
-     * audit is disabled.
-     */
-    @Test
-    public void testAuditDocumentsViaAliasContainMandatoryFields() {
-        try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) {
-            long before = countAuditDocs(client);
-            generateAuditEvent("_cluster/health");
-
-            await().until(() -> countAuditDocs(client) > before);
-
-            HttpResponse response = client.postJson(AUDIT_ALIAS + "/_search", """
-                {"query": {"match_all": {}}, "size": 1, "sort": [{"@timestamp": "desc"}]}
-                """);
-            response.assertStatusCode(200);
-
-            JsonNode source = response.bodyAsJsonNode().get("hits").get("hits").get(0).get("_source");
-
-            assertThat(source.has("audit_category"), is(true));
-            assertThat(source.has("@timestamp"), is(true));
-            assertThat(source.has("audit_rest_request_method"), is(true));
-            assertThat(source.has("audit_rest_request_path"), is(true));
-            assertThat(source.get("audit_request_layer").asText(), is("REST"));
-            assertThat(source.get("audit_request_origin").asText(), is("REST"));
-            assertThat(source.has("audit_transport_request_type"), is(false));
-        }
-    }
-}
+}

…ternalOpenSearchSinkIntegrationTest.java …rchSinkIntegrationTestConcreteIndex.javasrc/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTest.java renamed to src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTestConcreteIndex.java src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTest.java renamed to src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTestConcreteIndex.java

@@ -30,16 +30,19 @@
 import static org.hamcrest.Matchers.*;
 
 /**
- * Integration tests for {@link InternalOpenSearchSink} with default date-based index naming.
+ * Integration tests for {@link InternalOpenSearchSink} with a plain concrete index
+ * (no alias configured).
  *
- * <p>Tests the regular index creation path (no alias configured). The default pattern
- * {@code 'security-auditlog-'YYYY.MM.dd} creates daily indices
- * (e.g., {@code security-auditlog-2025.01.11}).</p>
+ * <p>Exercises the regular index creation path: the default pattern
+ * {@code 'security-auditlog-'YYYY.MM.dd} produces daily indices
+ * (e.g., {@code security-auditlog-2025.01.11}). No pre-existing index or alias is
+ * present when the cluster starts, so the sink must create the index on first write.</p>
  *
  * <h5>Tested Code Paths in {@code createIndexIfAbsent()}:</h5>
  * <ul>
  *   <li>First event: both {@code metadata.hasAlias()} and {@code metadata.hasIndex()} return
- *       {@code false}, triggering a {@code CreateIndexRequest} with the date-based index name.</li>
+ *       {@code false}, triggering a {@code CreateIndexRequest} with the date-based index name
+ *       ({@link #testCreatesAuditIndexAutomatically()}).</li>
  *   <li>Subsequent events: {@code metadata.hasIndex()} returns {@code true}; covered by
  *       the inherited {@link AbstractInternalOpenSearchSinkIntegrationTest#testPersistsAuditEventsToTarget()}.</li>
  * </ul>
@@ -49,9 +52,9 @@
  * {@link AbstractInternalOpenSearchSinkIntegrationTest}.</p>
  *
  * @see InternalOpenSearchSinkIntegrationTestAuditAlias for the write-alias variant
- * @see InternalOpenSearchSinkTest for unit tests covering exception branches
+ * @see InternalOpenSearchSinkTest for unit tests covering exception and race-condition branches
  */
-public class InternalOpenSearchSinkIntegrationTest extends AbstractInternalOpenSearchSinkIntegrationTest {
+public class InternalOpenSearchSinkIntegrationTestConcreteIndex extends AbstractInternalOpenSearchSinkIntegrationTest {
 
     private static final String AUDIT_INDEX_PREFIX = "security-auditlog-";
 
@@ -90,14 +93,14 @@ public void testCreatesAuditIndexAutomatically() {
 
             generateAuditEvent("_cluster/health");
 
-            await().atMost(10, SECONDS).pollInterval(100, MILLISECONDS).untilAsserted(() -> {
+            await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> {
                 refreshAuditTarget(client);
                 assertThat("At least one new audit event must be generated",
                     countAuditDocs(client), greaterThan(before));
             });
 
-            await().atMost(10, SECONDS).pollInterval(100, MILLISECONDS).untilAsserted(() -> {
-                GetIndexResponse response = CLUSTER.getInternalNodeClient()
+            await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> {
+                GetIndexResponse response = client
                     .admin()
                     .indices()
                     .getIndex(new GetIndexRequest().indices(AUDIT_INDEX_PREFIX + "*"))
@@ -112,4 +115,4 @@ public void testCreatesAuditIndexAutomatically() {
             });
         }
     }
-}
+}

src/test/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkTest.java

@@ -16,9 +16,11 @@
 
 import org.opensearch.ResourceAlreadyExistsException;
 import org.opensearch.action.admin.indices.create.CreateIndexRequest;
+import org.opensearch.action.admin.indices.create.CreateIndexResponse;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.action.ActionFuture;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.client.AdminClient;
@@ -31,27 +33,32 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 /**
  * Unit tests for {@link InternalOpenSearchSink#createIndexIfAbsent(String)}.
  *
- * <p>Covers the two exception branches that cannot be reliably triggered in an
- * integration test against a single-node embedded cluster:</p>
+ * <p>Covers three branches that cannot be reliably triggered in an integration
+ * test against a single-node embedded cluster:</p>
  * <ul>
  *   <li>{@code ResourceAlreadyExistsException} — thrown when two nodes race to
  *       create the same index simultaneously. The sink must treat this as success
  *       and return {@code true}.</li>
  *   <li>Generic {@code Exception} — thrown on unexpected cluster failures (e.g.,
  *       permission denied, cluster not available). The sink must absorb the error
  *       and return {@code false} without propagating the exception.</li>
+ *   <li>{@code acknowledged=false} — the cluster manager accepted the request but
+ *       timed out waiting for all shard copies to confirm. The index likely exists
+ *       on the primary node but may not yet be visible cluster-wide. The sink must
+ *       return {@code false} so the current audit event is skipped gracefully.</li>
  * </ul>
  *
  * <p>The remaining branches ({@code hasAlias()}, {@code hasIndex()}, and
  * successful creation) are covered by the integration tests:</p>
  * <ul>
- *   <li>{@link InternalOpenSearchSinkIntegrationTest}</li>
+ *   <li>{@link InternalOpenSearchSinkIntegrationTestConcreteIndex}</li>
  *   <li>{@link InternalOpenSearchSinkIntegrationTestAuditAlias}</li>
  * </ul>
  */
@@ -67,6 +74,7 @@ public class InternalOpenSearchSinkTest {
     @Mock private AdminClient adminClient;
     @Mock private IndicesAdminClient indicesAdminClient;
     @Mock private ThreadPool threadPool;
+    @Mock private ActionFuture<CreateIndexResponse> createIndexFuture;
 
     private InternalOpenSearchSink sink;
 
@@ -148,4 +156,49 @@ public void createIndexIfAbsent_returnsFalse_onUnexpectedException() {
             result, is(false));
         verify(indicesAdminClient).create(any(CreateIndexRequest.class));
     }
-}
+
+    /**
+     * Verifies that {@code createIndexIfAbsent()} returns {@code false} when the
+     * cluster manager acknowledges the request but the response is not acknowledged.
+     *
+     * <p><b>Scenario:</b> The {@code CreateIndexRequest} completes without throwing,
+     * but {@link CreateIndexResponse#isAcknowledged()} returns {@code false}. This
+     * can happen when the cluster manager times out waiting for all shard copies to
+     * confirm the mapping update. The index likely exists on the cluster manager node
+     * but may not yet be visible to other nodes. The sink logs an error and returns
+     * {@code false}, causing the current audit event to be dropped silently. The
+     * next event will find the index present and succeed normally.</p>
+     *
+     * <p><b>Tested Code Path:</b></p>
+     * <pre>{@code
+     * final boolean acknowledged = clientProvider.admin().indices()
+     *         .create(createIndexRequest).actionGet().isAcknowledged();
+     * if (acknowledged) { ... } else {
+     *     log.error("Failed to create audit log index '{}'. Index creation was not acknowledged.", indexName);
+     * }
+     * return acknowledged;  // returns false
+     * }</pre>
+     *
+     * <p><b>Implementation note:</b> {@code CreateIndexResponse} is instantiated directly
+     * (not mocked) because {@link org.opensearch.action.support.clustermanager.AcknowledgedResponse#isAcknowledged()}
+     * is declared {@code final} and cannot be stubbed by Mockito. {@code doReturn().when()}
+     * is used for {@code actionGet()} to avoid calling the generic method on the mock during
+     * stub setup (type erasure would confuse the standard {@code when().thenReturn()} form).</p>
+     */
+    @Test
+    public void createIndexIfAbsent_returnsFalse_whenCreationNotAcknowledged() {
+        // CreateIndexResponse is instantiated directly because isAcknowledged() is final
+        // in AcknowledgedResponse and cannot be stubbed by Mockito.
+        // doReturn().when() is used instead of when().thenReturn() to avoid calling the
+        // generic actionGet() on the mock during stub setup (prevents type-erasure issues).
+        CreateIndexResponse notAcknowledged = new CreateIndexResponse(false, false, TEST_INDEX);
+        doReturn(notAcknowledged).when(createIndexFuture).actionGet();
+        when(indicesAdminClient.create(any(CreateIndexRequest.class))).thenReturn(createIndexFuture);
+
+        boolean result = sink.createIndexIfAbsent(TEST_INDEX);
+
+        assertThat("Must return false when index creation is not acknowledged by the cluster",
+            result, is(false));
+        verify(indicesAdminClient).create(any(CreateIndexRequest.class));
+    }
+}