TaskConfig extends MapConfig { public Set<SystemStreamPartition> getBroadcastSystemStreamPartitions() { HashSet<SystemStreamPartition> systemStreamPartitionSet = new HashSet<>(); List<String> systemStreamPartitions = getList(BROADCAST_INPUT_STREAMS, Collections.emptyList()); for (String systemStreamPartition : systemStreamPartitions) { int hashPosition = systemStreamPartition.indexOf("#"); if (hashPosition == -1) { throw new IllegalArgumentException("incorrect format in " + systemStreamPartition + ". Broadcast stream names should be in the form 'system.stream#partitionId' or 'system.stream#[partitionN-partitionM]'"); } else { String systemStreamName = systemStreamPartition.substring(0, hashPosition); String partitionSegment = systemStreamPartition.substring(hashPosition + 1); SystemStream systemStream = StreamUtil.getSystemStreamFromNames(systemStreamName); if (Pattern.matches(BROADCAST_STREAM_PATTERN, partitionSegment)) { systemStreamPartitionSet.add(new SystemStreamPartition(systemStream, new Partition(Integer.valueOf(partitionSegment)))); } else { if (Pattern.matches(BROADCAST_STREAM_RANGE_PATTERN, partitionSegment)) { int partitionStart = Integer.valueOf(partitionSegment.substring(1, partitionSegment.lastIndexOf("-"))); int partitionEnd = Integer.valueOf(partitionSegment.substring(partitionSegment.lastIndexOf("-") + 1, partitionSegment.indexOf("]"))); if (partitionStart > partitionEnd) { LOGGER.warn("The starting partition in stream " + systemStream.toString() + " is bigger than the ending Partition. No partition is added"); } for (int i = partitionStart; i <= partitionEnd; i++) { systemStreamPartitionSet.add(new SystemStreamPartition(systemStream, new Partition(i))); } } else { throw new IllegalArgumentException("incorrect format in " + systemStreamPartition + ". Broadcast stream names should be in the form 'system.stream#partitionId' or 'system.stream#[partitionN-partitionM]'"); } } } } return systemStreamPartitionSet; } 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 testGetBroadcastSystemStreamPartitions() { assertEquals(Collections.emptySet(), new TaskConfig(new MapConfig()).getBroadcastSystemStreamPartitions()); HashMap<String, String> map = new HashMap<>(); map.put("task.broadcast.inputs", "kafka.foo#4, kafka.boo#5, kafka.z-o-o#[12-14], kafka.foo.bar#[3-4]"); Config config = new MapConfig(map); TaskConfig taskConfig = new TaskConfig(config); Set<SystemStreamPartition> systemStreamPartitionSet = taskConfig.getBroadcastSystemStreamPartitions(); HashSet<SystemStreamPartition> expected = new HashSet<>(); expected.add(new SystemStreamPartition("kafka", "foo", new Partition(4))); expected.add(new SystemStreamPartition("kafka", "boo", new Partition(5))); expected.add(new SystemStreamPartition("kafka", "z-o-o", new Partition(12))); expected.add(new SystemStreamPartition("kafka", "z-o-o", new Partition(13))); expected.add(new SystemStreamPartition("kafka", "z-o-o", new Partition(14))); expected.add(new SystemStreamPartition("kafka", "foo.bar", new Partition(3))); expected.add(new SystemStreamPartition("kafka", "foo.bar", new Partition(4))); assertEquals(expected, systemStreamPartitionSet); map.put("task.broadcast.inputs", "kafka.foo"); taskConfig = new TaskConfig(new MapConfig(map)); boolean catchCorrectException = false; try { taskConfig.getBroadcastSystemStreamPartitions(); } catch (IllegalArgumentException e) { catchCorrectException = true; } assertTrue(catchCorrectException); map.put("task.broadcast.inputs", "kafka.org.apache.events.WhitelistedIps#1-2"); taskConfig = new TaskConfig(new MapConfig(map)); boolean invalidFormatException = false; try { taskConfig.getBroadcastSystemStreamPartitions(); } catch (IllegalArgumentException e) { invalidFormatException = true; } assertTrue(invalidFormatException); }