@Test
public void testRemoveConsumersForSessionId() {

    final long userId = 1212L;
    final String sessionId1 = "abc123";
    final String sessionId2 = "123abc";
    final long viewId = 5454L;
    // Sanity test, manager should have no consumers.
    Assert.assertTrue("Manager should start empty", manager.getConsumers().isEmpty());
    // Define consumer inputs.
    final View view = new View();
    view.setId(viewId);
    final Collection<FilterDefinition> filters = new ArrayList<>();
    final StartingPosition startingPosition = StartingPosition.newHeadPosition();
    final SessionIdentifier sessionIdentifier1 = SessionIdentifier.newStreamIdentifier(userId, sessionId1);
    final SessionIdentifier sessionIdentifier2 = SessionIdentifier.newStreamIdentifier(userId, sessionId2);
    final SocketKafkaConsumer mockSocketKafkaConsumer1 = mock(SocketKafkaConsumer.class);
    final SocketKafkaConsumer mockSocketKafkaConsumer2 = mock(SocketKafkaConsumer.class);
    // Configure mocks
    when(mockConsumerFactory.createWebSocketClient(view, filters, startingPosition, sessionIdentifier1))
        .thenReturn(mockSocketKafkaConsumer1);
    when(mockConsumerFactory.createWebSocketClient(view, filters, startingPosition, sessionIdentifier2))
        .thenReturn(mockSocketKafkaConsumer2);
    // Add consumer once
    manager.addNewConsumer(view, filters, startingPosition, sessionIdentifier1);
    // Add another consumer, with different sessionIdentifier.
    manager.addNewConsumer(view, filters, startingPosition, sessionIdentifier2);
    // Ask for consumers
    final Collection<StreamConsumerDetails> consumers = manager.getConsumers();
    Assert.assertFalse("Should not be empty", consumers.isEmpty());
    Assert.assertEquals(2, consumers.size());
    // Ask to remove by sessionId
    manager.removeConsumersForSessionId(sessionId1);
    // Verify it requested the consumer to stop.
    verify(mockSocketKafkaConsumer1, times(1)).requestStop();
    verify(mockSocketKafkaConsumer2, times(0)).requestStop();
}