public final <KIn, VIn, KOut, VOut> void addProcessor(final String name,
                                                          final ProcessorSupplier<KIn, VIn, KOut, VOut> supplier,
                                                          final String... predecessorNames) {
        Objects.requireNonNull(name, "name must not be null");
        Objects.requireNonNull(supplier, "supplier must not be null");
        Objects.requireNonNull(predecessorNames, "predecessor names must not be null");
        ApiUtils.checkSupplier(supplier);
        if (nodeFactories.containsKey(name)) {
            throw new TopologyException("Processor " + name + " is already added.");
        }
        if (predecessorNames.length == 0) {
            throw new TopologyException("Processor " + name + " must have at least one parent");
        }

        for (final String predecessor : predecessorNames) {
            Objects.requireNonNull(predecessor, "predecessor name must not be null");
            if (predecessor.equals(name)) {
                throw new TopologyException("Processor " + name + " cannot be a predecessor of itself.");
            }
            if (!nodeFactories.containsKey(predecessor)) {
                throw new TopologyException("Predecessor processor " + predecessor + " is not added yet for " + name);
            }
        }

        nodeFactories.put(name, new ProcessorNodeFactory<>(name, predecessorNames, supplier));
        nodeGrouper.add(name);
        nodeGrouper.unite(name, predecessorNames);
        nodeGroups = null;
    }
@Test
    public void shouldNotAllowNullNameWhenAddingProcessor() {
        assertThrows(
            NullPointerException.class,
            () -> builder.addProcessor(
                null,
                (ProcessorSupplier<Object, Object, Object, Object>) () -> null
            )
        );
    }