WebKafkaConsumerFactory { public WebKafkaConsumer createWebClient( final View view, final Collection<FilterDefinition> filterDefinitions, final SessionIdentifier sessionIdentifier) { final ClientConfig clientConfig = createClientConfig(view, filterDefinitions, sessionIdentifier) .withStartingPosition(StartingPosition.newResumeFromExistingState()) .build(); if (multiThreadedConsumerThreadPool != null) { return new ParallelWebKafkaConsumer(kafkaConsumerFactory, clientConfig, multiThreadedConsumerThreadPool); } else { final KafkaConsumer kafkaConsumer = createKafkaConsumer(clientConfig); return new DefaultWebKafkaConsumer(kafkaConsumer, clientConfig); } } WebKafkaConsumerFactory(
        final PluginFactory<Deserializer> deserializerPluginFactory,
        final PluginFactory<RecordFilter> recordFilterPluginFactory,
        final SecretManager secretManager,
        final KafkaConsumerFactory kafkaConsumerFactory,
        final ExecutorService multiThreadedConsumerThreadPool); WebKafkaConsumer createWebClient(
        final View view,
        final Collection<FilterDefinition> filterDefinitions,
        final SessionIdentifier sessionIdentifier); SocketKafkaConsumer createWebSocketClient(
        final View view,
        final Collection<FilterDefinition> filterDefinitions,
        final StartingPosition startingPosition,
        final SessionIdentifier sessionIdentifier);  }
@Test @Parameters(method = "provideFactoryInstances") public void smokeTestWebClient_noFilters_allPartitions(final WebKafkaConsumerFactory factory) { final int resultsPerPartition = 5; final View view = createDefaultView(topic1); view.setResultsPerPartition(resultsPerPartition); final SessionIdentifier sessionId = SessionIdentifier.newWebIdentifier(12L, "MySession"); try (final WebKafkaConsumer webKafkaConsumer = factory.createWebClient(view, new ArrayList<>(), sessionId)) { assertNotNull(webKafkaConsumer); final List<KafkaResult> results = consumeAllResults(webKafkaConsumer, resultsPerPartition); assertEquals( "Should have a total of 20 records", 20, results.size()); for (final KafkaResult kafkaResult: results) { assertTrue("Should be from partition 0 or 1", kafkaResult.getPartition() == 0 || kafkaResult.getPartition() == 1); } } }
@Test @Parameters(method = "provideFactoryInstances") public void smokeTestWebClient_noFilters_partitionFilter(final WebKafkaConsumerFactory factory) { final int resultsPerPartition = 5; final View view = createDefaultView(topic1); view.setResultsPerPartition(resultsPerPartition); view.setPartitions("1"); final SessionIdentifier sessionId = SessionIdentifier.newWebIdentifier(12L, "MySession"); try (final WebKafkaConsumer webKafkaConsumer = factory.createWebClient(view, new ArrayList<>(), sessionId)) { assertNotNull(webKafkaConsumer); final List<KafkaResult> results = consumeAllResults(webKafkaConsumer, resultsPerPartition); assertEquals("Should have 10 records", 10, results.size()); for (final KafkaResult kafkaResult: results) { assertEquals("Should be from partition 1", 1, kafkaResult.getPartition()); } } }