This page is for marking off proposed interfaces as they're implemented.

Public Interfaces

First, we propose to add one new interface NamedOperation that can be used to customize stateless operations as well as stateful ones. The objective to create a new class is to keep consistent with the overall API design.

package org.apache.kafka.streams.kstream;

/**
 * Default interface which can be used to personalized the named of operations, internal topics or store.
 */
public interface NamedOperation<T extends NamedOperation<T>> {

    /**
     * Sets the name to be used for operation.
     *
     * @param name  the name to use.
     * @return an instance of {@link NamedOperation}
     */
    T withName(final String name);

}

The NamedOperation interface will be implemented/extended by following classes that already exist to configure operations : 

• Produced 
• Consumed 
• Printed 
• Joined 
• Grouped 
• Suppressed 

In addition, we propose to add a new static method with the following signature to each of those class as(final String processorName).

(Suppressed is missing it, but we discussed this and decided not to include it)

Printed is missing it, not sure if this is intentional

Deprecration

In order to fix some inconsistencies in API, we also propose to deprecate the method named(String) from the class Joined in favor of new method as().

In addition, as no configuration classes expose the processor name, we will also deprecate the method name() from the class Joined . This method should be removed in a future release.

Overloaded methods

Then, we propose to overload all stateless methods that do not have one of the existing control classes listed above to accept a NamedOperation implementation.

For that we will added a new default class Named implementing NamedOperation :

public class Named implements NamedOperation<Named> {

    private static final int MAX_NAME_LENGTH = 249;

    protected String name;

    protected Named(final String name) {
        this.name = name;
        if (name != null)
            validate(name);
    }

    /**
     * Create a Named instance with provided name.
     *
     * @param name  the processor name to be used. If {@code null} a default processor name will be generated.
     * @return      A new {@link Named} instance configured with name
     */
    public static Named as(final String name) {
        Objects.requireNonNull(name, "name can't be null");
        return new Named(name);
    }

    @Override
    public Named withName(final String name) {
        Objects.requireNonNull(name, "name can't be null");
        return new Named(name);
    }
...
}

The below tables will resume all new and existing methods :

KStream (16 new methods)

KTable (16 new methods)

KGroupedStream (6 new methods)

KGroupedTable (6 new methods)

TimeWindowedKStream (6 new methods)

SessionWindowedKStream (6 new methods)

Name Validation

User provided node name should follow the same restrictions that ones currently apply to state stores during the create of Materialized instance.

Currently, the Materialized class relies on the static method Topic#validate. This method ensure that a provided name only contains legal characters [a-zA-Z0-9._-] and have a maximum length of 249.

We propose to copy methods from Topic#validate into Named. This new method will be used validate both store names and node names. The benefit is to remove a dependency with the core module.

In addition, the Materialized class will throw a TopologyException while building the topology in case of a unvalid name instead of InvalidTopicException .

Proposed Changes

• Implement the new interface NamedOperation and default class Named 
• Update all parameter class to implement NamedOperation : Produced Consumed Printed Joined Grouped Suppressed
• Overload methods stateless for classes KStreams, KTables, KGroupedStream, KGroupedTable, TimeWindowedKStream, TimeWindowedKTable
• The processor names specified by developer will be used in place of the static processor prefix. Statics prefixes will still be used if no custom processor name are specified.
• Processor names should follow the same restrictions as the topic names. So legal characters are [a-zA-Z0-9._-] and the maximum length of 249.