TaskConfig extends MapConfig { public boolean getTransactionalStateRestoreEnabled() { JobConfig jobConfig = new JobConfig(this); boolean standByEnabled = jobConfig.getStandbyTasksEnabled(); boolean asyncCommitEnabled = getAsyncCommit(); return !standByEnabled && !asyncCommitEnabled && getBoolean(TRANSACTIONAL_STATE_RESTORE_ENABLED, DEFAULT_TRANSACTIONAL_STATE_RESTORE_ENABLED); } TaskConfig(Config config); Set<SystemStream> getInputStreams(); long getWindowMs(); long getCommitMs(); Optional<String> getTaskClass(); String getCommandClass(String defaultCommandClass); String getMessageChooserClass(); boolean getDropDeserializationErrors(); boolean getDropSerializationErrors(); boolean getDropProducerErrors(); int getPollIntervalMs(); Optional<String> getIgnoredExceptions(); String getTaskNameGrouperFactory(); int getMaxConcurrency(); long getCallbackTimeoutMs(); boolean getAsyncCommit(); long getMaxIdleMs(); Optional<CheckpointManager> getCheckpointManager(MetricsRegistry metricsRegistry); boolean getCheckpointManagerConsumerStopAfterFirstRead(); Set<SystemStreamPartition> getBroadcastSystemStreamPartitions(); Set<SystemStream> getBroadcastSystemStreams(); Set<SystemStream> getAllInputStreams(); long getShutdownMs(); boolean getTransactionalStateCheckpointEnabled(); boolean getTransactionalStateRestoreEnabled(); boolean getTransactionalStateRetainExistingState(); static final Logger LOGGER; static final String INPUT_STREAMS; static final String WINDOW_MS; static final String COMMIT_MS; static final String TASK_SHUTDOWN_MS; static final String TASK_CLASS; static final String COMMAND_BUILDER; static final String MESSAGE_CHOOSER_CLASS_NAME; static final String DROP_DESERIALIZATION_ERRORS; static final String DROP_SERIALIZATION_ERRORS; static final String DROP_PRODUCER_ERRORS; static final String IGNORED_EXCEPTIONS; static final String GROUPER_FACTORY; static final String MAX_CONCURRENCY; static final String CALLBACK_TIMEOUT_MS; static final String ASYNC_COMMIT; static final String MAX_IDLE_MS; static final String POLL_INTERVAL_MS; static final int DEFAULT_POLL_INTERVAL_MS; static final String BROADCAST_INPUT_STREAMS; static final String CHECKPOINT_MANAGER_FACTORY; static final String INTERNAL_CHECKPOINT_MANAGER_CONSUMER_STOP_AFTER_FIRST_READ; static final String TRANSACTIONAL_STATE_CHECKPOINT_ENABLED; static final String TRANSACTIONAL_STATE_RESTORE_ENABLED; static final String TRANSACTIONAL_STATE_RETAIN_EXISTING_STATE; }
@Test public void testGetTransactionalStateRestoreEnabled() { Map<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TRANSACTIONAL_STATE_RESTORE_ENABLED, "true"); assertTrue(new TaskConfig(new MapConfig(configMap)).getTransactionalStateRestoreEnabled()); configMap.put(TaskConfig.ASYNC_COMMIT, "true"); configMap.put(JobConfig.STANDBY_TASKS_REPLICATION_FACTOR, "1"); assertFalse(new TaskConfig(new MapConfig(configMap)).getTransactionalStateRestoreEnabled()); configMap.put(TaskConfig.ASYNC_COMMIT, "false"); configMap.put(JobConfig.STANDBY_TASKS_REPLICATION_FACTOR, "2"); assertFalse(new TaskConfig(new MapConfig(configMap)).getTransactionalStateRestoreEnabled()); configMap.put(TaskConfig.ASYNC_COMMIT, "true"); configMap.put(JobConfig.STANDBY_TASKS_REPLICATION_FACTOR, "2"); assertFalse(new TaskConfig(new MapConfig(configMap)).getTransactionalStateRestoreEnabled()); }