public Collection<Task> createTasks(final Consumer<byte[], byte[]> consumer,
                                        final Map<TaskId, Set<TopicPartition>> tasksToBeCreated) {
        final List<Task> createdTasks = new ArrayList<>();

        for (final Map.Entry<TaskId, Set<TopicPartition>> newTaskAndPartitions : tasksToBeCreated.entrySet()) {
            final TaskId taskId = newTaskAndPartitions.getKey();
            final LogContext logContext = getLogContext(taskId);
            final Set<TopicPartition> partitions = newTaskAndPartitions.getValue();
            final ProcessorTopology topology = topologyMetadata.buildSubtopology(taskId);

            final ProcessorStateManager stateManager = new ProcessorStateManager(
                taskId,
                Task.TaskType.ACTIVE,
                eosEnabled(applicationConfig),
                logContext,
                stateDirectory,
                storeChangelogReader,
                topology.storeToChangelogTopic(),
                partitions,
                stateUpdaterEnabled);

            final InternalProcessorContext<Object, Object> context = new ProcessorContextImpl(
                taskId,
                applicationConfig,
                stateManager,
                streamsMetrics,
                cache
            );

            createdTasks.add(
                createActiveTask(
                    taskId,
                    partitions,
                    consumer,
                    logContext,
                    topology,
                    stateManager,
                    context
                )
            );
        }
        return createdTasks;
    }
@SuppressWarnings("deprecation")
    @Test
    public void shouldFailOnGetThreadProducerIfEosAlphaEnabled() {
        properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
        mockClientSupplier.setApplicationIdForProducer("appId");

        createTasks();

        final IllegalStateException thrown = assertThrows(
            IllegalStateException.class,
            activeTaskCreator::threadProducer
        );

        assertThat(thrown.getMessage(), is("Expected AT_LEAST_ONCE or EXACTLY_ONCE_V2 to be enabled, but the processing mode was EXACTLY_ONCE_ALPHA"));
    }