@Test public void testFactory() { Configuration configuration = new Configuration(); Reductions reductions = Reductions.create( configuration, new StatefulFunctionsUniverse(MessageFactoryType.WITH_KRYO_PAYLOADS), new FakeRuntimeContext(), new FakeKeyedStateBackend(), new FakeTimerServiceFactory(), new FakeInternalListState(), new HashMap<>(), new FakeOutput(), TestUtils.ENVELOPE_FACTORY, MoreExecutors.directExecutor(), new FakeMetricGroup(), new FakeMapState(), MoreExecutors.directExecutor()); assertThat(reductions, notNullValue()); }
static Reductions create( Configuration configuration, StatefulFunctionsUniverse statefulFunctionsUniverse, RuntimeContext context, KeyedStateBackend<Object> keyedStateBackend, TimerServiceFactory timerServiceFactory, InternalListState<String, Long, Message> delayedMessagesBufferState, Map<EgressIdentifier<?>, OutputTag<Object>> sideOutputs, Output<StreamRecord<Message>> output, MessageFactory messageFactory, Executor mailboxExecutor, MetricGroup metricGroup, MapState<Long, Message> asyncOperations, Executor checkpointLockExecutor) { ObjectContainer container = new ObjectContainer(); container.add("function-providers", Map.class, statefulFunctionsUniverse.functions()); container.add( "function-repository", FunctionRepository.class, StatefulFunctionRepository.class); container.add("runtime-context", RuntimeContext.class, context); container.add("keyed-state-backend", KeyedStateBackend.class, keyedStateBackend); container.add(new DynamicallyRegisteredTypes(statefulFunctionsUniverse.types())); if (configuration.getBoolean(StatefulFunctionsJobConstants.MULTIPLEX_FLINK_STATE)) { container.add("state", State.class, MultiplexedState.class); } else { container.add("state", State.class, FlinkState.class); } container.add(messageFactory); container.add( new Partition( context.getMaxNumberOfParallelSubtasks(), context.getNumberOfParallelSubtasks(), context.getIndexOfThisSubtask())); container.add(new RemoteSink(output)); container.add(new SideOutputSink(sideOutputs, output)); container.add("applying-context", ApplyingContext.class, ReusableContext.class); container.add(LocalSink.class); container.add("function-loader", FunctionLoader.class, PredefinedFunctionLoader.class); container.add(StateBinder.class); container.add(Reductions.class); container.add(LocalFunctionGroup.class); container.add("metrics-factory", MetricsFactory.class, new FlinkMetricsFactory(metricGroup)); container.add( "delayed-messages-buffer-state", InternalListState.class, delayedMessagesBufferState); container.add( "delayed-messages-buffer", DelayedMessagesBuffer.class, FlinkStateDelayedMessagesBuffer.class); container.add( "delayed-messages-timer-service-factory", TimerServiceFactory.class, timerServiceFactory); container.add(DelaySink.class); container.add("function-group", new Lazy<>(LocalFunctionGroup.class)); container.add("reductions", new Lazy<>(Reductions.class)); container.add("mailbox-executor", Executor.class, mailboxExecutor); container.add("async-operations", MapState.class, asyncOperations); container.add("checkpoint-lock-executor", Executor.class, checkpointLockExecutor); container.add(AsyncSink.class); return container.get(Reductions.class); }
Reductions { static Reductions create( Configuration configuration, StatefulFunctionsUniverse statefulFunctionsUniverse, RuntimeContext context, KeyedStateBackend<Object> keyedStateBackend, TimerServiceFactory timerServiceFactory, InternalListState<String, Long, Message> delayedMessagesBufferState, Map<EgressIdentifier<?>, OutputTag<Object>> sideOutputs, Output<StreamRecord<Message>> output, MessageFactory messageFactory, Executor mailboxExecutor, MetricGroup metricGroup, MapState<Long, Message> asyncOperations, Executor checkpointLockExecutor) { ObjectContainer container = new ObjectContainer(); container.add("function-providers", Map.class, statefulFunctionsUniverse.functions()); container.add( "function-repository", FunctionRepository.class, StatefulFunctionRepository.class); container.add("runtime-context", RuntimeContext.class, context); container.add("keyed-state-backend", KeyedStateBackend.class, keyedStateBackend); container.add(new DynamicallyRegisteredTypes(statefulFunctionsUniverse.types())); if (configuration.getBoolean(StatefulFunctionsJobConstants.MULTIPLEX_FLINK_STATE)) { container.add("state", State.class, MultiplexedState.class); } else { container.add("state", State.class, FlinkState.class); } container.add(messageFactory); container.add( new Partition( context.getMaxNumberOfParallelSubtasks(), context.getNumberOfParallelSubtasks(), context.getIndexOfThisSubtask())); container.add(new RemoteSink(output)); container.add(new SideOutputSink(sideOutputs, output)); container.add("applying-context", ApplyingContext.class, ReusableContext.class); container.add(LocalSink.class); container.add("function-loader", FunctionLoader.class, PredefinedFunctionLoader.class); container.add(StateBinder.class); container.add(Reductions.class); container.add(LocalFunctionGroup.class); container.add("metrics-factory", MetricsFactory.class, new FlinkMetricsFactory(metricGroup)); container.add( "delayed-messages-buffer-state", InternalListState.class, delayedMessagesBufferState); container.add( "delayed-messages-buffer", DelayedMessagesBuffer.class, FlinkStateDelayedMessagesBuffer.class); container.add( "delayed-messages-timer-service-factory", TimerServiceFactory.class, timerServiceFactory); container.add(DelaySink.class); container.add("function-group", new Lazy<>(LocalFunctionGroup.class)); container.add("reductions", new Lazy<>(Reductions.class)); container.add("mailbox-executor", Executor.class, mailboxExecutor); container.add("async-operations", MapState.class, asyncOperations); container.add("checkpoint-lock-executor", Executor.class, checkpointLockExecutor); container.add(AsyncSink.class); return container.get(Reductions.class); } }
Reductions { static Reductions create( Configuration configuration, StatefulFunctionsUniverse statefulFunctionsUniverse, RuntimeContext context, KeyedStateBackend<Object> keyedStateBackend, TimerServiceFactory timerServiceFactory, InternalListState<String, Long, Message> delayedMessagesBufferState, Map<EgressIdentifier<?>, OutputTag<Object>> sideOutputs, Output<StreamRecord<Message>> output, MessageFactory messageFactory, Executor mailboxExecutor, MetricGroup metricGroup, MapState<Long, Message> asyncOperations, Executor checkpointLockExecutor) { ObjectContainer container = new ObjectContainer(); container.add("function-providers", Map.class, statefulFunctionsUniverse.functions()); container.add( "function-repository", FunctionRepository.class, StatefulFunctionRepository.class); container.add("runtime-context", RuntimeContext.class, context); container.add("keyed-state-backend", KeyedStateBackend.class, keyedStateBackend); container.add(new DynamicallyRegisteredTypes(statefulFunctionsUniverse.types())); if (configuration.getBoolean(StatefulFunctionsJobConstants.MULTIPLEX_FLINK_STATE)) { container.add("state", State.class, MultiplexedState.class); } else { container.add("state", State.class, FlinkState.class); } container.add(messageFactory); container.add( new Partition( context.getMaxNumberOfParallelSubtasks(), context.getNumberOfParallelSubtasks(), context.getIndexOfThisSubtask())); container.add(new RemoteSink(output)); container.add(new SideOutputSink(sideOutputs, output)); container.add("applying-context", ApplyingContext.class, ReusableContext.class); container.add(LocalSink.class); container.add("function-loader", FunctionLoader.class, PredefinedFunctionLoader.class); container.add(StateBinder.class); container.add(Reductions.class); container.add(LocalFunctionGroup.class); container.add("metrics-factory", MetricsFactory.class, new FlinkMetricsFactory(metricGroup)); container.add( "delayed-messages-buffer-state", InternalListState.class, delayedMessagesBufferState); container.add( "delayed-messages-buffer", DelayedMessagesBuffer.class, FlinkStateDelayedMessagesBuffer.class); container.add( "delayed-messages-timer-service-factory", TimerServiceFactory.class, timerServiceFactory); container.add(DelaySink.class); container.add("function-group", new Lazy<>(LocalFunctionGroup.class)); container.add("reductions", new Lazy<>(Reductions.class)); container.add("mailbox-executor", Executor.class, mailboxExecutor); container.add("async-operations", MapState.class, asyncOperations); container.add("checkpoint-lock-executor", Executor.class, checkpointLockExecutor); container.add(AsyncSink.class); return container.get(Reductions.class); } @Inject  Reductions(LocalFunctionGroup functionGroup); }
Reductions { static Reductions create( Configuration configuration, StatefulFunctionsUniverse statefulFunctionsUniverse, RuntimeContext context, KeyedStateBackend<Object> keyedStateBackend, TimerServiceFactory timerServiceFactory, InternalListState<String, Long, Message> delayedMessagesBufferState, Map<EgressIdentifier<?>, OutputTag<Object>> sideOutputs, Output<StreamRecord<Message>> output, MessageFactory messageFactory, Executor mailboxExecutor, MetricGroup metricGroup, MapState<Long, Message> asyncOperations, Executor checkpointLockExecutor) { ObjectContainer container = new ObjectContainer(); container.add("function-providers", Map.class, statefulFunctionsUniverse.functions()); container.add( "function-repository", FunctionRepository.class, StatefulFunctionRepository.class); container.add("runtime-context", RuntimeContext.class, context); container.add("keyed-state-backend", KeyedStateBackend.class, keyedStateBackend); container.add(new DynamicallyRegisteredTypes(statefulFunctionsUniverse.types())); if (configuration.getBoolean(StatefulFunctionsJobConstants.MULTIPLEX_FLINK_STATE)) { container.add("state", State.class, MultiplexedState.class); } else { container.add("state", State.class, FlinkState.class); } container.add(messageFactory); container.add( new Partition( context.getMaxNumberOfParallelSubtasks(), context.getNumberOfParallelSubtasks(), context.getIndexOfThisSubtask())); container.add(new RemoteSink(output)); container.add(new SideOutputSink(sideOutputs, output)); container.add("applying-context", ApplyingContext.class, ReusableContext.class); container.add(LocalSink.class); container.add("function-loader", FunctionLoader.class, PredefinedFunctionLoader.class); container.add(StateBinder.class); container.add(Reductions.class); container.add(LocalFunctionGroup.class); container.add("metrics-factory", MetricsFactory.class, new FlinkMetricsFactory(metricGroup)); container.add( "delayed-messages-buffer-state", InternalListState.class, delayedMessagesBufferState); container.add( "delayed-messages-buffer", DelayedMessagesBuffer.class, FlinkStateDelayedMessagesBuffer.class); container.add( "delayed-messages-timer-service-factory", TimerServiceFactory.class, timerServiceFactory); container.add(DelaySink.class); container.add("function-group", new Lazy<>(LocalFunctionGroup.class)); container.add("reductions", new Lazy<>(Reductions.class)); container.add("mailbox-executor", Executor.class, mailboxExecutor); container.add("async-operations", MapState.class, asyncOperations); container.add("checkpoint-lock-executor", Executor.class, checkpointLockExecutor); container.add(AsyncSink.class); return container.get(Reductions.class); } @Inject  Reductions(LocalFunctionGroup functionGroup);  }
Reductions { static Reductions create( Configuration configuration, StatefulFunctionsUniverse statefulFunctionsUniverse, RuntimeContext context, KeyedStateBackend<Object> keyedStateBackend, TimerServiceFactory timerServiceFactory, InternalListState<String, Long, Message> delayedMessagesBufferState, Map<EgressIdentifier<?>, OutputTag<Object>> sideOutputs, Output<StreamRecord<Message>> output, MessageFactory messageFactory, Executor mailboxExecutor, MetricGroup metricGroup, MapState<Long, Message> asyncOperations, Executor checkpointLockExecutor) { ObjectContainer container = new ObjectContainer(); container.add("function-providers", Map.class, statefulFunctionsUniverse.functions()); container.add( "function-repository", FunctionRepository.class, StatefulFunctionRepository.class); container.add("runtime-context", RuntimeContext.class, context); container.add("keyed-state-backend", KeyedStateBackend.class, keyedStateBackend); container.add(new DynamicallyRegisteredTypes(statefulFunctionsUniverse.types())); if (configuration.getBoolean(StatefulFunctionsJobConstants.MULTIPLEX_FLINK_STATE)) { container.add("state", State.class, MultiplexedState.class); } else { container.add("state", State.class, FlinkState.class); } container.add(messageFactory); container.add( new Partition( context.getMaxNumberOfParallelSubtasks(), context.getNumberOfParallelSubtasks(), context.getIndexOfThisSubtask())); container.add(new RemoteSink(output)); container.add(new SideOutputSink(sideOutputs, output)); container.add("applying-context", ApplyingContext.class, ReusableContext.class); container.add(LocalSink.class); container.add("function-loader", FunctionLoader.class, PredefinedFunctionLoader.class); container.add(StateBinder.class); container.add(Reductions.class); container.add(LocalFunctionGroup.class); container.add("metrics-factory", MetricsFactory.class, new FlinkMetricsFactory(metricGroup)); container.add( "delayed-messages-buffer-state", InternalListState.class, delayedMessagesBufferState); container.add( "delayed-messages-buffer", DelayedMessagesBuffer.class, FlinkStateDelayedMessagesBuffer.class); container.add( "delayed-messages-timer-service-factory", TimerServiceFactory.class, timerServiceFactory); container.add(DelaySink.class); container.add("function-group", new Lazy<>(LocalFunctionGroup.class)); container.add("reductions", new Lazy<>(Reductions.class)); container.add("mailbox-executor", Executor.class, mailboxExecutor); container.add("async-operations", MapState.class, asyncOperations); container.add("checkpoint-lock-executor", Executor.class, checkpointLockExecutor); container.add(AsyncSink.class); return container.get(Reductions.class); } @Inject  Reductions(LocalFunctionGroup functionGroup);   }