public static void publishPriceUpdate(String topicArn, String payload, String groupId) {

        try {
            // Create and publish a message that updates the wholesale price.
            String subject = "Price Update";
            String dedupId = UUID.randomUUID().toString();
            String attributeName = "business";
            String attributeValue = "wholesale";

            MessageAttributeValue msgAttValue = MessageAttributeValue.builder()
                    .dataType("String")
                    .stringValue(attributeValue)
                    .build();

            Map<String, MessageAttributeValue> attributes = new HashMap<>();
            attributes.put(attributeName, msgAttValue);
            PublishRequest pubRequest = PublishRequest.builder()
                    .topicArn(topicArn)
                    .subject(subject)
                    .message(payload)
                    .messageGroupId(groupId)
                    .messageDeduplicationId(dedupId)
                    .messageAttributes(attributes)
                    .build();

            final PublishResponse response = snsClient.publish(pubRequest);
            System.out.println(response.messageId());
            System.out.println(response.sequenceNumber());
            System.out.println("Message was published to " + topicArn);

        } catch (SnsException e) {
            System.err.println(e.awsErrorDetails().errorMessage());
            System.exit(1);
        }
    }
@Test
    @Tag("IntegrationTest")
    void publishPriceUpdateTest() {
        String topicName = "MyTestTopic.fifo";
        String wholesaleQueueName = "wholesaleQueue.fifo";
        String retailQueueName = "retailQueue.fifo";
        String analyticsQueueName = "analyticsQueue";

        List<PriceUpdateExample.QueueData> queues = List.of(
                new PriceUpdateExample.QueueData(wholesaleQueueName, PriceUpdateExample.QueueType.FIFO),
                new PriceUpdateExample.QueueData(retailQueueName, PriceUpdateExample.QueueType.FIFO),
                new PriceUpdateExample.QueueData(analyticsQueueName, PriceUpdateExample.QueueType.Standard));

        createQueues(queues);
        String topicARN = createFIFOTopic(topicName);
        subscribeQueues(queues, topicARN);
        addAccessPolicyToQueuesFINAL(queues, topicARN);
        publishPriceUpdate(topicARN, "{\"product\": 214, \"price\": 79.99}", "Consumables");

        // Assert that each queue received the message published
        queues.forEach(queue -> {
            queue.testMessage = sqsClient.receiveMessage(builder -> builder.queueUrl(queue.queueURL).maxNumberOfMessages(1)).messages().get(0);
            Assertions.assertNotNull(queue.testMessage);
        });

        deleteSubscriptions(queues);
        deleteQueues(queues);
        deleteTopic(topicARN);
    }