Skip to content

Commit 8d3014c

Browse files
maniskummaniskum
committed
refactor
Added IT tests
1 parent 5a7af34 commit 8d3014c

File tree

6 files changed

+113
-23
lines changed

6 files changed

+113
-23
lines changed

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

+7-3
Original file line numberDiff line numberDiff line change
@@ -137,9 +137,13 @@ public void publishData(List<Pair<String, SinkRecord>> messages) throws AEPStrea
137137
}
138138
}
139139

140-
private Integer getFailedMessageIndex(final JsonNode messageResponse) {
141-
final String xactionId = messageResponse.get(XACTIONID_KEY).asText();
142-
return Integer.parseInt(xactionId.substring(xactionId.lastIndexOf(DASH) + 1));
140+
private Integer getFailedMessageIndex(final JsonNode messageResponse) throws HttpException {
141+
if (messageResponse.hasNonNull(XACTIONID_KEY)) {
142+
final String xactionId = messageResponse.get(XACTIONID_KEY).asText();
143+
return Integer.parseInt(xactionId.substring(xactionId.lastIndexOf(DASH) + 1));
144+
}
145+
throw new HttpException(String.format("xactionId is missing in the failed message error response : %s",
146+
messageResponse));
143147
}
144148

145149
public void stop() {

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

+41-8
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ public class AEPSinkConnectorErrorReporterTest extends AbstractConnectorTest {
4747
private static final String AEP_KAFKA_ERROR_CONNECTOR_CONFIG = "aep-connector-error-reporter.json";
4848
private static final String AEP_KAFKA_ERROR_CONNECTOR_HEADER_CONFIG = "aep-connector-error-reporter-header.json";
4949
private static final String XDM_PAYLOAD_FILE = "xdm-data.json";
50+
private static final String XDM_MULTI_MESSAGE_PAYLOAD_FILE = "xdm-data-multiple-messages.json";
5051
private static final String DEAD_LETTER_TOPIC = "errors.deadletterqueue.topic.name";
5152
private static final String ERROR_CLASS_NAME = "__connect.errors.exception.class.name";
5253
private static final String ERROR_HEADER_MESSAGE = "__connect.errors.exception.message";
@@ -70,7 +71,7 @@ public void kafkaErrorReporterTest() throws HttpException, IOException, Interrup
7071
LOG.info("Starting connector cluster with connector : {}", CONNECTOR_NAME);
7172
getConnect().configureConnector(CONNECTOR_NAME, connectorConfig);
7273

73-
String xdmData = xdmData();
74+
String xdmData = xdmData(XDM_PAYLOAD_FILE);
7475
getConnect().kafka().produce(TOPIC_NAME, xdmData);
7576
waitForConnectorStart(CONNECTOR_NAME, 1, 8000);
7677

@@ -87,7 +88,39 @@ public void kafkaErrorReporterTest() throws HttpException, IOException, Interrup
8788

8889
// Verify inlet endpoint received 1 XDM record
8990
getWiremockServer().verify(postRequestedFor(urlEqualTo(getRelativeUrl()))
90-
.withRequestBody(equalToJson(payloadReceivedXdmData())));
91+
.withRequestBody(equalToJson(payloadReceivedXdmData(XDM_PAYLOAD_FILE))));
92+
}
93+
94+
@Test
95+
public void kafkaErrorReporterMultiMessageTest() throws HttpException, IOException, InterruptedException {
96+
inletMultiStatusSuccessfulResponse();
97+
getConnect().kafka().createTopic(TOPIC_NAME, TOPIC_PARTITION);
98+
99+
// Create error topic to dump failed data
100+
Map<String, String> connectorConfig = connectorConfig(AEP_KAFKA_ERROR_CONNECTOR_CONFIG);
101+
getConnect().kafka().createTopic(connectorConfig.get(DEAD_LETTER_TOPIC), TOPIC_PARTITION);
102+
103+
LOG.info("Starting connector cluster with connector : {}", CONNECTOR_NAME);
104+
getConnect().configureConnector(CONNECTOR_NAME, connectorConfig);
105+
106+
String xdmData = xdmData(XDM_MULTI_MESSAGE_PAYLOAD_FILE);
107+
getConnect().kafka().produce(TOPIC_NAME, xdmData);
108+
waitForConnectorStart(CONNECTOR_NAME, 1, 8000);
109+
110+
// Check if error record sent to error topic
111+
ConsumerRecords<byte[], byte[]> consumerRecords = getConnect().kafka()
112+
.consume(1, 8000, connectorConfig.get(DEAD_LETTER_TOPIC));
113+
114+
Assertions.assertEquals(1, consumerRecords.count());
115+
116+
ConsumerRecord<byte[], byte[]> consumerRecord = consumerRecords.iterator().next();
117+
JsonNode record = JacksonFactory.OBJECT_MAPPER.readTree(consumerRecord.value());
118+
119+
Assertions.assertEquals(JacksonFactory.OBJECT_MAPPER.readTree(xdmData).toString(), record.toString());
120+
121+
// Verify inlet endpoint received 1 XDM record
122+
getWiremockServer().verify(postRequestedFor(urlEqualTo(getRelativeUrl()))
123+
.withRequestBody(equalToJson(payloadReceivedXdmData(XDM_MULTI_MESSAGE_PAYLOAD_FILE))));
91124
}
92125

93126
@Test
@@ -101,7 +134,7 @@ public void kafkaErrorReporterWithHeadersTest() throws HttpException, IOExceptio
101134
LOG.info("Starting connector cluster with connector : {}", CONNECTOR_NAME);
102135
getConnect().configureConnector(CONNECTOR_NAME, connectorConfig);
103136

104-
String xdmData = xdmData();
137+
String xdmData = xdmData(XDM_PAYLOAD_FILE);
105138
getConnect().kafka().produce(TOPIC_NAME, xdmData);
106139
waitForConnectorStart(CONNECTOR_NAME, 1, 8000);
107140

@@ -124,11 +157,11 @@ public void kafkaErrorReporterWithHeadersTest() throws HttpException, IOExceptio
124157

125158
// Verify inlet endpoint received 1 XDM record
126159
getWiremockServer().verify(postRequestedFor(urlEqualTo(getRelativeUrl()))
127-
.withRequestBody(equalToJson(payloadReceivedXdmData())));
160+
.withRequestBody(equalToJson(payloadReceivedXdmData(XDM_PAYLOAD_FILE))));
128161
}
129162

130-
public String payloadReceivedXdmData() throws HttpException, JsonProcessingException {
131-
String xdmData = xdmData();
163+
public String payloadReceivedXdmData(String payloadfile) throws HttpException, JsonProcessingException {
164+
String xdmData = xdmData(payloadfile);
132165
ObjectNode messageNode = JacksonFactory.OBJECT_MAPPER.createObjectNode();
133166
ArrayNode xdmDataValues = JacksonFactory.OBJECT_MAPPER.createArrayNode();
134167
xdmDataValues.add(JacksonFactory.OBJECT_MAPPER.readTree(xdmData));
@@ -137,8 +170,8 @@ public String payloadReceivedXdmData() throws HttpException, JsonProcessingExcep
137170
return JacksonFactory.OBJECT_MAPPER.writeValueAsString(messageNode);
138171
}
139172

140-
public String xdmData() throws HttpException {
141-
return HttpUtil.streamToString(this.getClass().getClassLoader().getResourceAsStream(XDM_PAYLOAD_FILE));
173+
public String xdmData(String payloadfile) throws HttpException {
174+
return HttpUtil.streamToString(this.getClass().getClassLoader().getResourceAsStream(payloadfile));
142175
}
143176

144177
public Map<String, String> connectorConfig(String configFile) throws HttpException, JsonProcessingException {

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

+11
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,8 @@ public abstract class AbstractConnectorTest {
5656
private static final String AEP_CONNECTOR_INLET_SUCCESSFUL_RESPONSE =
5757
"aep-connector-inlet-successful-response.json";
5858

59+
private static final String AEP_CONNECTOR_INLET_MULTI_STATUS_SUCCESSFUL_RESPONSE =
60+
"aep-connector-inlet-multi-message-response.json";
5961
protected static final int TOPIC_PARTITION = 1;
6062
protected static final int NUMBER_OF_TASKS = 1;
6163
protected static final String CONNECTOR_NAME = "aep-sink-connector";
@@ -132,6 +134,15 @@ public void inletSuccessfulResponse() throws IOException {
132134
.getResourceAsStream(AEP_CONNECTOR_INLET_SUCCESSFUL_RESPONSE)))));
133135
}
134136

137+
public void inletMultiStatusSuccessfulResponse() throws IOException {
138+
wiremockExtension.getWireMockServer()
139+
.stubFor(WireMock
140+
.post(WireMock.urlEqualTo(getRelativeUrl()))
141+
.willReturn(ResponseDefinitionBuilder.responseDefinition()
142+
.withJsonBody(JacksonFactory.OBJECT_MAPPER.readTree(this.getClass().getClassLoader()
143+
.getResourceAsStream(AEP_CONNECTOR_INLET_MULTI_STATUS_SUCCESSFUL_RESPONSE)))));
144+
}
145+
135146
public void inletIMSAuthenticationSuccessfulResponse() throws JsonProcessingException {
136147
wiremockExtension.getWireMockServer()
137148
.stubFor(WireMock

‎streaming-connect-sink/src/test/resources/aep-connector-inlet-multi-message-response.json

+15
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
{
2+
"inletId": "9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5",
3+
"batchId": "1565638336649:1750:244",
4+
"receivedTimeMs": 1565638336705,
5+
"responses": [
6+
{
7+
"xactionId":"9341f8eb-494a-4c89-9879-4d06a58d2dc7-0",
8+
"status":400,
9+
"message":"The 'header' field is mandatory. Provide a valid 'header' value and try again."
10+
},
11+
{
12+
"xactionId": "9341f8eb-494a-4c89-9879-4d06a58d2dc7-1"
13+
}
14+
]
15+
}

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

-12
Original file line numberDiff line numberDiff line change
@@ -5,18 +5,6 @@
55
"responses": [
66
{
77
"xactionId": "1565650704337:2124:92:3"
8-
},
9-
{
10-
"status": 400,
11-
"message": "inletId: [9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5] imsOrgId: [{ORG_ID}] Message has unknown xdm format"
12-
},
13-
{
14-
"status": 400,
15-
"message": "inletId: [9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5] imsOrgId: [{ORG_ID}] Message has an absent or wrong ims org in the header"
16-
},
17-
{
18-
"status": 400,
19-
"message": "inletId: [9b0cb233972f3b0092992284c7353f5eead496218e8441a79b25e9421ea127f5] imsOrgId: [{ORG_ID}] Message has unknown xdm format"
208
}
219
]
2210
}

‎streaming-connect-sink/src/test/resources/xdm-data-multiple-messages.json

+39
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
[
2+
{
3+
"body": {
4+
"xdmMeta": {
5+
"schemaRef": {
6+
"id": "https://ns.adobe.com/{TENANT_ID}/schemas/{SCHEMA_ID}",
7+
"contentType": "application/vnd.adobe.xed-full+json;version={SCHEMA_VERSION}"
8+
}
9+
},
10+
"xdmEntity": {
11+
"firstname": "abc",
12+
"lastname": "def"
13+
}
14+
}
15+
},
16+
{
17+
"header": {
18+
"schemaRef": {
19+
"id": "https://ns.adobe.com/{TENANT_ID}/schemas/{SCHEMA_ID}",
20+
"contentType": "application/vnd.adobe.xed-full+json;version={SCHEMA_VERSION}"
21+
},
22+
"source": {
23+
"name": "aep-sink-connector"
24+
}
25+
},
26+
"body": {
27+
"xdmMeta": {
28+
"schemaRef": {
29+
"id": "https://ns.adobe.com/{TENANT_ID}/schemas/{SCHEMA_ID}",
30+
"contentType": "application/vnd.adobe.xed-full+json;version={SCHEMA_VERSION}"
31+
}
32+
},
33+
"xdmEntity": {
34+
"firstname": "abc",
35+
"lastname": "def"
36+
}
37+
}
38+
}
39+
]

0 commit comments

Comments
 (0)