class WebKafkaConsumerFactory {

    public WebKafkaConsumer createWebClient(
        final View view,
        final Collection<FilterDefinition> filterDefinitions,
        final SessionIdentifier sessionIdentifier) {

        // Create client config builder
        final ClientConfig clientConfig = createClientConfig(view, filterDefinitions, sessionIdentifier)
            // Always resume from existing state.
            .withStartingPosition(StartingPosition.newResumeFromExistingState())
            .build();

        // If we've been passed an executor service
        if (multiThreadedConsumerThreadPool != null) {
            // Assume we want to use multi-threaded consumer.
            return new ParallelWebKafkaConsumer(kafkaConsumerFactory, clientConfig, multiThreadedConsumerThreadPool);
        } else {
            // Create single threaded kafka consumer
            final KafkaConsumer kafkaConsumer = createKafkaConsumer(clientConfig);
            return new DefaultWebKafkaConsumer(kafkaConsumer, clientConfig);
        }
    }

    public  WebKafkaConsumerFactory(
        final PluginFactory<Deserializer> deserializerPluginFactory,
        final PluginFactory<RecordFilter> recordFilterPluginFactory,
        final SecretManager secretManager,
        final KafkaConsumerFactory kafkaConsumerFactory,
        final ExecutorService multiThreadedConsumerThreadPool);

}