Carry the notification subject on SNS batch entries (#1704)

Author: anjeongkyun

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/converter/DefaultSnsMessageConverter.java

@@ -18,6 +18,7 @@
 import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_DEDUPLICATION_ID_HEADER;
 import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_GROUP_ID_HEADER;
 import static io.awspring.cloud.sns.core.SnsHeaders.MESSAGE_ID_HEADER;
+import static io.awspring.cloud.sns.core.SnsHeaders.NOTIFICATION_SUBJECT_HEADER;
 
 import io.awspring.cloud.sns.core.SnsHeaderConverterUtil;
 import java.util.ArrayList;
@@ -96,6 +97,9 @@ private <T> void populateHeader(PublishBatchRequestEntry.Builder publishBatchReq
 				.filter(StringUtils::hasText).orElseGet(() -> UUID.randomUUID().toString());
 		publishBatchRequestEntry.id(id);
 
+		Optional.ofNullable(message.getHeaders().get(NOTIFICATION_SUBJECT_HEADER, String.class))
+				.ifPresent(publishBatchRequestEntry::subject);
+
 		Optional.ofNullable(message.getHeaders().get(MESSAGE_GROUP_ID_HEADER, String.class))
 				.ifPresent(publishBatchRequestEntry::messageGroupId);
 		Optional.ofNullable(message.getHeaders().get(MESSAGE_DEDUPLICATION_ID_HEADER, String.class))

spring-cloud-aws-sns/src/main/java/io/awspring/cloud/sns/core/batch/executor/SequentialBatchExecutionStrategy.java

@@ -55,7 +55,7 @@ public SequentialBatchExecutionStrategy(SnsClient snsClient) {
 	@Override
 	public BatchResult send(Arn topicArn, Collection<PublishBatchRequestEntry> entries) {
 		Assert.notNull(topicArn, "topicArn is required");
-		Assert.notNull(topicArn, "entries are required");
+		Assert.notNull(entries, "entries are required");
 
 		List<BatchResult.SnsResult> allResults = new ArrayList<>();
 		List<BatchResult.SnsError> allErrors = new ArrayList<>();

spring-cloud-aws-sns/src/test/java/io/awspring/cloud/sns/core/batch/converter/DefaultSnsMessageConverterTest.java

@@ -95,12 +95,25 @@ void setsAllFifoHeaders() {
 				.isEqualTo(MessageAttributeValue.builder().stringValue("val").dataType("String").build());
 	}
 
+	@Test
+	void setsNotificationSubject() {
+		Message<String> message = MessageBuilder.withPayload("test")
+				.setHeader(SnsHeaders.NOTIFICATION_SUBJECT_HEADER, "a subject").build();
+
+		PublishBatchRequestEntry entry = converter.convertMessage(message);
+
+		assertThat(entry.subject()).isEqualTo("a subject");
+		// the subject travels on the entry, not as a message attribute
+		assertThat(entry.messageAttributes()).doesNotContainKey(SnsHeaders.NOTIFICATION_SUBJECT_HEADER);
+	}
+
 	@Test
 	void leavesOptionalHeadersNullWhenAbsent() {
 		Message<String> message = MessageBuilder.withPayload("plain").build();
 
 		PublishBatchRequestEntry entry = converter.convertMessage(message);
 
+		assertThat(entry.subject()).isNull();
 		assertThat(entry.messageGroupId()).isNull();
 		assertThat(entry.messageDeduplicationId()).isNull();
 	}