@Test
public void shouldReceiveAMessageFromTheServer() throws Exception {

    final String expectedSessionId = "MYSESSIONID";
    final String topicName = "TestTopic" + System.currentTimeMillis();
    try {
        // Sanity test, no active consumers
        Assert.assertEquals("Should have no active consumers", 0, webSocketConsumersManager.countActiveConsumers());
        // Create a topic
        sharedKafkaTestResource
            .getKafkaTestUtils()
            .createTopic(topicName, 2, (short) 1);
        // Create cluster
        final Cluster cluster = clusterTestTools
            .createCluster("TestCluster", sharedKafkaTestResource.getKafkaConnectString());
        // Create view
        final View view = viewTestTools
            .createViewWithCluster("TestView-" + System.currentTimeMillis(), cluster);
        // Sanity test, no enforced partitions
        assertEquals("Partitions Property should be empty string", "", view.getPartitions());
        final ConsumeRequest consumeRequest = new ConsumeRequest();
        consumeRequest.setAction("head");
        consumeRequest.setPartitions("0,1");
        consumeRequest.setFilters(new ArrayList<>());
        consumeRequest.setResultsPerPartition(10);
        final SimpMessageHeaderAccessor mockHeaderAccessor = mock(SimpMessageHeaderAccessor.class);
        final Authentication mockPrincipal = mock(Authentication.class);
        when(mockHeaderAccessor.getUser())
            .thenReturn(mockPrincipal);
        when(mockPrincipal.getPrincipal())
            .thenReturn(nonAdminUserDetails);
        when(mockHeaderAccessor.getSessionId())
            .thenReturn(expectedSessionId);
        final String result = streamController.newConsumer(
            view.getId(),
            consumeRequest,
            mockHeaderAccessor
        );
        assertEquals("Should return success response", "{success: true}", result);
        // Verify consumer stood up
        assertEquals("Should have one active consumer", 1, webSocketConsumersManager.countActiveConsumers());
        // Lets refresh the view entity and verify the partitions field is still empty.
        // Validates ISSUE-212
        final View updatedView = viewRepository.findById(view.getId()).get();
        assertEquals("Partitions Property should be empty string", "", updatedView.getPartitions());
    } finally {
        // Cleanup, disconnect websocket consumers
        webSocketConsumersManager.removeConsumersForSessionId(expectedSessionId);
        // Remove created data.
        viewTestTools.deleteAllViews();
        clusterTestTools.deleteAllClusters();
    }
}