KafkaProducerRecordBuilder { static List<KafkaProducerRecord<String, String>> buildRecords(String topic, Buffer payload) throws ValidationException { List<KafkaProducerRecord<String, String>> kafkaProducerRecords = new ArrayList<>(); JsonObject payloadObj; try { payloadObj = new JsonObject(payload); } catch (DecodeException de){ throw new ValidationException("Error while parsing payload"); } JsonArray recordsArray; try { recordsArray = payloadObj.getJsonArray(RECORDS); if(recordsArray == null){ throw new ValidationException("Missing 'records' array"); } for (int i = 0; i < recordsArray.size(); i++) { kafkaProducerRecords.add(fromRecordJsonObject(topic, recordsArray.getJsonObject(i))); } } catch (ClassCastException cce) { throw new ValidationException("Property '"+RECORDS+"' must be of type JsonArray holding JsonObject objects"); } return kafkaProducerRecords; }    }
@Test public void buildRecordsInvalidJson() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Error while parsing payload"); buildRecords("myTopic", Buffer.buffer("notValidJson")); }
@Test public void buildRecordsMissingRecordsArray() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Missing 'records' array"); buildRecords("myTopic", Buffer.buffer("{}")); }
@Test public void buildRecordsNotArray() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'records' must be of type JsonArray holding JsonObject objects"); buildRecords("myTopic", Buffer.buffer("{\"records\": \"shouldBeAnArray\"}")); }
@Test public void buildRecordsInvalidRecordsType() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'records' must be of type JsonArray holding JsonObject objects"); buildRecords("myTopic", Buffer.buffer("{\"records\": [123]}")); }
@Test public void buildRecordsEmptyRecordsArray(TestContext context) throws ValidationException { final List<KafkaProducerRecord<String, String>> records = buildRecords("myTopic", Buffer.buffer("{\"records\": []}")); context.assertTrue(records.isEmpty()); }
@Test public void buildRecordsInvalidKeyType() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'key' must be of type String"); buildRecords("myTopic", Buffer.buffer("{\"records\": [{\"key\": 123,\"value\": {}}]}")); }
@Test public void buildRecordsInvalidValueType() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'value' must be of type JsonObject"); buildRecords("myTopic", Buffer.buffer("{\"records\":[{\"value\":123}]}")); }
@Test public void buildRecordsMissingValue() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'value' is required"); buildRecords("myTopic", Buffer.buffer("{\"records\":[{}]}")); }
@Test public void buildRecordsInvalidHeadersType() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'headers' must be of type JsonObject"); buildRecords("myTopic", Buffer.buffer("{\"records\": [{\"value\":{},\"headers\": 123}]}")); }
@Test public void buildRecordsInvalidHeadersValueType() throws ValidationException { thrown.expect( ValidationException.class ); thrown.expectMessage("Property 'headers' must be of type JsonObject holding String values only"); buildRecords("myTopic", Buffer.buffer("{\"records\": [{\"value\": {},\"headers\": {\"key\": 555}}]}")); }
@Test public void buildRecordsValidNoKeyNoHeaders(TestContext context) throws ValidationException { JsonObject payload = buildPayload(null, null); final List<KafkaProducerRecord<String, String>> records = buildRecords("myTopic", Buffer.buffer(payload.encode())); context.assertFalse(records.isEmpty()); context.assertEquals(1, records.size()); context.assertEquals("myTopic", records.get(0).topic()); context.assertNull(records.get(0).key()); context.assertEquals(payload.getJsonArray("records").getJsonObject(0).getJsonObject("value").encode(), records.get(0).value()); context.assertTrue(records.get(0).headers().isEmpty()); }
@Test public void buildRecordsValidWithKeyNoHeaders(TestContext context) throws ValidationException { JsonObject payload = buildPayload("someKey", null); final List<KafkaProducerRecord<String, String>> records = buildRecords("myTopic", Buffer.buffer(payload.encode())); context.assertFalse(records.isEmpty()); context.assertEquals(1, records.size()); context.assertEquals("myTopic", records.get(0).topic()); context.assertEquals("someKey", records.get(0).key()); context.assertEquals(payload.getJsonArray("records").getJsonObject(0).getJsonObject("value").encode(), records.get(0).value()); context.assertTrue(records.get(0).headers().isEmpty()); }
@Test public void buildRecordsValidWithKeyWithHeaders(TestContext context) throws ValidationException { JsonObject payload = buildPayload("someKey", new CaseInsensitiveHeaders() .add("header_1", "value_1") .add("header_2", "value_2")); final List<KafkaProducerRecord<String, String>> records = buildRecords("myTopic", Buffer.buffer(payload.encode())); context.assertFalse(records.isEmpty()); context.assertEquals(1, records.size()); context.assertEquals("myTopic", records.get(0).topic()); context.assertEquals("someKey", records.get(0).key()); context.assertEquals(payload.getJsonArray("records").getJsonObject(0).getJsonObject("value").encode(), records.get(0).value()); context.assertFalse(records.get(0).headers().isEmpty()); context.assertEquals(2, records.get(0).headers().size()); context.assertEquals("header_1", records.get(0).headers().get(0).key()); context.assertEquals("value_1", records.get(0).headers().get(0).value().toString()); context.assertEquals("header_2", records.get(0).headers().get(1).key()); context.assertEquals("value_2", records.get(0).headers().get(1).value().toString()); }