+ @Test
+ public void unsupportedOperationInMessage_thenNotFoundResponseWithNotImplementedOperation() throws Exception {
+ String message = dmaapInputMessage(Operation.PUT).toString();
+ String badOperation = "BAD";
+ message = message.replace(Operation.PUT.toString(), badOperation);
+
+ testedObject.handleDmaapMsg(message);
+
+ ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
+ verify(dmaapClient).send(captor.capture());
+ String actualMessage = captor.getValue();
+ assertThat(actualMessage
+ .contains(HttpStatus.BAD_REQUEST + "\",\"message\":\"Not implemented operation: " + badOperation)).isTrue();
+
+ verify(dmaapClient).sendBatchWithResponse();
+ verifyNoMoreInteractions(dmaapClient);
+ }
+
+ @Test
+ public void putWithoutPayload_thenNotFoundResponseWithWarning() throws Exception {
+ String message = dmaapInputMessage(Operation.PUT).toString();
+ message = message.replace(",\"payload\":{\"name\":\"name\",\"schema\":\"schema\"}", "");
+
+ final ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DmaapMessageHandler.class);
+
+ testedObject.handleDmaapMsg(message);
+
+ assertThat(logAppender.list.get(0).getLevel()).isEqualTo(Level.WARN);
+ assertThat(logAppender.list.toString().contains("Expected payload in message from DMAAP: ")).isTrue();
+ }