Improved logging for the message published in AEP (#68)

* Improved logging for the message published in AEP

* Remove HttpException

---------

Co-authored-by: vivetiwa <[email protected]>

Author: maniskum

streaming-connect-sink/src/main/java/com/adobe/platform/streaming/sink/impl/AEPPublisher.java

@@ -44,6 +44,8 @@ public class AEPPublisher extends AbstractAEPPublisher {
   private static final Logger LOG = LoggerFactory.getLogger(AEPPublisher.class);
 
   private static final String MESSAGES_KEY = "messages";
+  private static final String RESPONSES_KEY = "responses";
+  private static final String STATUS_KEY = "status";
 
   private int count;
   private final HttpProducer producer;
@@ -61,6 +63,9 @@ public void publishData(List<Pair<String, SinkRecord>> messages) throws AEPStrea
       LOG.debug("No messages to publish");
       return;
     }
+    int failedMessageCount = 0;
+    int successMessageCount = 0;
+    int totalMessageCount;
 
     final ArrayNode jsonMessages = JacksonFactory.OBJECT_MAPPER.createArrayNode();
     try {
@@ -80,14 +85,32 @@ public void publishData(List<Pair<String, SinkRecord>> messages) throws AEPStrea
       final JsonNode payload = JacksonFactory.OBJECT_MAPPER.createObjectNode()
         .set(MESSAGES_KEY, jsonMessages);
 
+      totalMessageCount = jsonMessages.size();
+
       final JsonNode response = producer.post(
         StringUtils.EMPTY,
         JacksonFactory.OBJECT_MAPPER.writeValueAsBytes(payload),
         ContentHandler.jsonHandler()
       );
 
       count++;
-      LOG.debug("Successfully published data to Adobe Experience Platform: {}", response);
+
+      if (!response.isNull() && !response.isEmpty()) {
+        JsonNode publishMessagesResponses = response.get(RESPONSES_KEY);
+        for (JsonNode messageResponse : publishMessagesResponses) {
+          if (messageResponse.hasNonNull(STATUS_KEY)) {
+            failedMessageCount++;
+            LOG.debug("Failed to publish message to Adobe Experience Platform: {}", messageResponse);
+          } else {
+            successMessageCount++;
+          }
+        }
+        LOG.debug("Total publish message count = {}. Success message count = {}. Failed " +
+          "message count = {}.", totalMessageCount, successMessageCount, failedMessageCount);
+      } else {
+        LOG.error("Invalid Response received while publishing data to  Adobe Experience Platform: {}", response);
+      }
+
     } catch (JsonProcessingException jsonException) {
       LOG.error("Failed to publish data to Adobe Experience Platform", jsonException);
       if (Objects.nonNull(errorReporter)) {

streaming-connect-sink/src/test/java/com/adobe/platform/streaming/integration/AEPSinkConnectorErrorReporterTest.java

@@ -54,7 +54,7 @@ public class AEPSinkConnectorErrorReporterTest extends AbstractConnectorTest {
 
   @BeforeEach
   @Override
-  public void setup() throws JsonProcessingException {
+  public void setup() throws IOException {
     super.setup();
     inletFailedResponse();
   }

streaming-connect-sink/src/test/java/com/adobe/platform/streaming/integration/AEPSinkConnectorTest.java

@@ -27,6 +27,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static com.github.tomakehurst.wiremock.client.WireMock.equalToJson;
@@ -54,7 +55,7 @@ public class AEPSinkConnectorTest extends AbstractConnectorTest {
 
   @BeforeEach
   @Override
-  public void setup() throws JsonProcessingException {
+  public void setup() throws IOException {
     super.setup();
     inletSuccessfulResponse();
     inletSuccessfulResponseViaProxy();

streaming-connect-sink/src/test/java/com/adobe/platform/streaming/integration/AbstractConnectorTest.java

@@ -28,6 +28,7 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
@@ -52,6 +53,9 @@ public abstract class AbstractConnectorTest {
   private static final String AUTH_TOKEN_RESPONSE_OAUTH2 = "{\"access_token\":\"accessToken\"," +
     "\"token_type\":\"bearer\",\"expires_in\":86400}";
 
+  private static final String AEP_CONNECTOR_INLET_SUCCESSFUL_RESPONSE =
+    "aep-connector-inlet-successful-response.json";
+
   protected static final int TOPIC_PARTITION = 1;
   protected static final int NUMBER_OF_TASKS = 1;
   protected static final String CONNECTOR_NAME = "aep-sink-connector";
@@ -75,7 +79,7 @@ public abstract class AbstractConnectorTest {
   public static final WiremockExtension wiremockExtensionViaProxy = new WiremockExtension(PORT_VIA_PROXY);
 
   @BeforeEach
-  public void setup() throws JsonProcessingException {
+  public void setup() throws IOException {
     connect = new EmbeddedConnectCluster.Builder()
         .name("aep-connect-cluster")
         .numWorkers(numberOfWorkers)
@@ -119,12 +123,13 @@ public void stop() {
     connect.stop();
   }
 
-  public void inletSuccessfulResponse() throws JsonProcessingException {
+  public void inletSuccessfulResponse() throws IOException {
     wiremockExtension.getWireMockServer()
       .stubFor(WireMock
       .post(WireMock.urlEqualTo(getRelativeUrl()))
       .willReturn(ResponseDefinitionBuilder.responseDefinition()
-      .withJsonBody(JacksonFactory.OBJECT_MAPPER.readTree("{\"payloadReceived\": true}"))));
+      .withJsonBody(JacksonFactory.OBJECT_MAPPER.readTree(this.getClass().getClassLoader()
+      .getResourceAsStream(AEP_CONNECTOR_INLET_SUCCESSFUL_RESPONSE)))));
   }
 
   public void inletIMSAuthenticationSuccessfulResponse() throws JsonProcessingException {

streaming-connect-sink/src/test/resources/aep-connector-inlet-successful-response.json

@@ -0,0 +1,22 @@
+{
+  "inletId": "9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5",
+  "batchId": "1565638336649:1750:244",
+  "receivedTimeMs": 1565638336705,
+  "responses": [
+    {
+      "xactionId": "1565650704337:2124:92:3"
+    },
+    {
+      "status": 400,
+      "message": "inletId: [9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5] imsOrgId: [{ORG_ID}] Message has unknown xdm format"
+    },
+    {
+      "status": 400,
+      "message": "inletId: [9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5] imsOrgId: [{ORG_ID}] Message has an absent or wrong ims org in the header"
+    },
+    {
+      "status": 400,
+      "message": "inletId: [9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5] imsOrgId: [{ORG_ID}] Message has unknown xdm format"
+    }
+  ]
+}