fix(cloudwatch-logs): add @NotNull guard, rate-limit entity rejection log, remove duplicate test

- Add @NotNull on attributes field to prevent NPE from explicit null in YAML
- Use NOISY marker on entity rejection WARN to rate-limit under load
- Remove redundant empty-key-attributes test (default is already emptyMap)
- Add test for @NotNull constraint violation on null attributes

Signed-off-by: Nikhil Bagmar <nikhilbagmar73@gmail.com>

Author: bagmarnikhil

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/client/CloudWatchLogsDispatcher.java

@@ -177,7 +177,7 @@ public void upload() {
                 }
                 if (putLogEventsResponse != null && putLogEventsResponse.rejectedEntityInfo() != null) {
                     cloudWatchLogsMetrics.increaseEntityRejectedCounter(1);
-                    LOG.warn("Entity was rejected by CloudWatch: {}",
+                    LOG.warn(NOISY, "Entity was rejected by CloudWatch: {}",
                             putLogEventsResponse.rejectedEntityInfo().errorTypeAsString());
                 }
                 cloudWatchLogsMetrics.increaseLogEventSuccessCounter(totalEventCount - dlqObjects.size());

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/config/EntityConfig.java

@@ -12,6 +12,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
 
 import java.util.Collections;
 import java.util.Map;
@@ -23,6 +24,7 @@ public class EntityConfig {
     private Map<String, String> keyAttributes = Collections.emptyMap();
 
     @JsonProperty("attributes")
+    @NotNull
     private Map<String, String> attributes = Collections.emptyMap();
 
     public Map<String, String> getKeyAttributes() {

data-prepper-plugins/cloudwatch-logs/src/test/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/config/EntityConfigTest.java

@@ -19,7 +19,6 @@
 import org.junit.jupiter.api.Test;
 import org.opensearch.dataprepper.test.helper.ReflectivelySetField;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -122,10 +121,8 @@ void GIVEN_entity_config_with_only_key_attributes_in_yaml_WHEN_deserialized_SHOU
     }
 
     @Test
-    void GIVEN_entity_config_with_empty_key_attributes_WHEN_validated_SHOULD_fail_NotEmpty_constraint()
-            throws NoSuchFieldException, IllegalAccessException {
+    void GIVEN_entity_config_with_default_key_attributes_WHEN_validated_SHOULD_fail_NotEmpty_constraint() {
         final EntityConfig entityConfig = new EntityConfig();
-        ReflectivelySetField.setField(EntityConfig.class, entityConfig, KEY_ATTRIBUTES_FIELD, Collections.emptyMap());
 
         final Set<ConstraintViolation<EntityConfig>> violations = validator.validate(entityConfig);
 
@@ -135,14 +132,19 @@ void GIVEN_entity_config_with_empty_key_attributes_WHEN_validated_SHOULD_fail_No
     }
 
     @Test
-    void GIVEN_entity_config_with_default_key_attributes_WHEN_validated_SHOULD_fail_NotEmpty_constraint() {
+    void GIVEN_null_attributes_WHEN_validated_SHOULD_fail_NotNull_constraint()
+            throws NoSuchFieldException, IllegalAccessException {
         final EntityConfig entityConfig = new EntityConfig();
+        final Map<String, String> keyAttributes = new HashMap<>();
+        keyAttributes.put("Type", "Service");
+        keyAttributes.put("Name", "my-app");
+        ReflectivelySetField.setField(EntityConfig.class, entityConfig, KEY_ATTRIBUTES_FIELD, keyAttributes);
+        ReflectivelySetField.setField(EntityConfig.class, entityConfig, ATTRIBUTES_FIELD, null);
 
         final Set<ConstraintViolation<EntityConfig>> violations = validator.validate(entityConfig);
 
         assertThat(violations, hasSize(1));
-        final ConstraintViolation<EntityConfig> violation = violations.iterator().next();
-        assertThat(violation.getPropertyPath().toString(), equalTo(KEY_ATTRIBUTES_FIELD));
+        assertThat(violations.iterator().next().getPropertyPath().toString(), equalTo(ATTRIBUTES_FIELD));
     }
 
     @Test