Skip to content

rename ProcessorFactory to ProcessorDef #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ymatsuda wants to merge 1 commit into from
Closed

rename ProcessorFactory to ProcessorDef #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/KStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

/**
* KStream is an abstraction of a stream of key-value pairs.
Expand Down Expand Up @@ -133,7 +133,7 @@ public interface KStream<K, V> {
/**
* Processes all elements in this stream by applying a processor.
*
* @param processorFactory the class of ProcessorFactory
* @param processorDef the class of ProcessorFactory
*/
<K1, V1> KStream<K1, V1> process(ProcessorFactory processorFactory);
<K1, V1> KStream<K1, V1> process(ProcessorDef processorDef);
}
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamBranch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@
package org.apache.kafka.streaming.kstream.internals;

import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;
import org.apache.kafka.streaming.kstream.Predicate;

class KStreamBranch<K, V> implements ProcessorFactory {
class KStreamBranch<K, V> implements ProcessorDef {

private final Predicate<K, V>[] predicates;

Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@

import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.kstream.Predicate;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

class KStreamFilter<K, V> implements ProcessorFactory {
class KStreamFilter<K, V> implements ProcessorDef {

private final Predicate<K, V> predicate;
private final boolean filterOut;
Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@
import org.apache.kafka.streaming.kstream.KeyValue;
import org.apache.kafka.streaming.kstream.KeyValueFlatMap;
import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

class KStreamFlatMap<K1, V1, K2, V2> implements ProcessorFactory {
class KStreamFlatMap<K1, V1, K2, V2> implements ProcessorDef {

private final KeyValueFlatMap<K1, V1, K2, V2> mapper;

Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValues.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@

import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.kstream.ValueMapper;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

class KStreamFlatMapValues<K1, V1, V2> implements ProcessorFactory {
class KStreamFlatMapValues<K1, V1, V2> implements ProcessorDef {

private final ValueMapper<V1, ? extends Iterable<V2>> mapper;

Expand Down
6 changes: 3 additions & 3 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streaming.kstream.KeyValueFlatMap;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;
import org.apache.kafka.streaming.processor.TopologyBuilder;
import org.apache.kafka.streaming.kstream.KStreamWindowed;
import org.apache.kafka.streaming.kstream.KeyValueMapper;
Expand Down Expand Up @@ -185,10 +185,10 @@ public void sendTo(String topic, Serializer<K> keySerializer, Serializer<V> valS

@SuppressWarnings("unchecked")
@Override
public <K1, V1> KStream<K1, V1> process(final ProcessorFactory processorFactory) {
public <K1, V1> KStream<K1, V1> process(final ProcessorDef processorDef) {
String name = PROCESSOR_NAME + INDEX.getAndIncrement();

topology.addProcessor(name, processorFactory, this.name);
topology.addProcessor(name, processorDef, this.name);

return new KStreamImpl<>(topology, name);
}
Expand Down
6 changes: 3 additions & 3 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,11 @@
import org.apache.kafka.streaming.processor.ProcessorContext;
import org.apache.kafka.streaming.kstream.ValueJoiner;
import org.apache.kafka.streaming.kstream.Window;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

import java.util.Iterator;

class KStreamJoin<K, V, V1, V2> implements ProcessorFactory {
class KStreamJoin<K, V, V1, V2> implements ProcessorDef {

private static abstract class Finder<K, T> {
abstract Iterator<T> find(K key, long timestamp);
Expand All @@ -37,7 +37,7 @@ private static abstract class Finder<K, T> {
private final boolean prior;

private Processor processorForOtherStream = null;
public final ProcessorFactory processorFactoryForOtherStream = new ProcessorFactory() {
public final ProcessorDef processorDefForOtherStream = new ProcessorDef() {
@Override
public Processor build() {
return processorForOtherStream;
Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@
import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.kstream.KeyValue;
import org.apache.kafka.streaming.kstream.KeyValueMapper;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

class KStreamMap<K1, V1, K2, V2> implements ProcessorFactory {
class KStreamMap<K1, V1, K2, V2> implements ProcessorDef {

private final KeyValueMapper<K1, V1, K2, V2> mapper;

Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValues.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@

import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.kstream.ValueMapper;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

class KStreamMapValues<K1, V1, V2> implements ProcessorFactory {
class KStreamMapValues<K1, V1, V2> implements ProcessorDef {

private final ValueMapper<V1, V2> mapper;

Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamPassThrough.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,9 @@
package org.apache.kafka.streaming.kstream.internals;

import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;

class KStreamPassThrough<K, V> implements ProcessorFactory {
class KStreamPassThrough<K, V> implements ProcessorDef {

@Override
public Processor build() {
Expand Down
4 changes: 2 additions & 2 deletions stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,11 @@

import org.apache.kafka.streaming.kstream.Window;
import org.apache.kafka.streaming.processor.Processor;
import org.apache.kafka.streaming.processor.ProcessorFactory;
import org.apache.kafka.streaming.processor.ProcessorDef;
import org.apache.kafka.streaming.processor.ProcessorContext;
import org.apache.kafka.streaming.kstream.WindowDef;

public class KStreamWindow<K, V> implements ProcessorFactory {
public class KStreamWindow<K, V> implements ProcessorDef {

private final WindowDef<K, V> windowDef;

Expand Down
2 changes: 1 addition & 1 deletion stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ private <V1, V2> KStream<K, V2> join(KStreamWindowed<K, V1> other, boolean prior
String joinName = JOIN_NAME + INDEX.getAndIncrement();
String joinOtherName = JOINOTHER_NAME + INDEX.getAndIncrement();
topology.addProcessor(joinName, join, this.name);
topology.addProcessor(joinOtherName, join.processorFactoryForOtherStream, ((KStreamImpl) other).name);
topology.addProcessor(joinOtherName, join.processorDefForOtherStream, ((KStreamImpl) other).name);

return new KStreamImpl<>(topology, joinName);
}
Expand Down
3 changes: 2 additions & 1 deletion ...streaming/processor/ProcessorFactory.java → ...fka/streaming/processor/ProcessorDef.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,8 @@

package org.apache.kafka.streaming.processor;

public interface ProcessorFactory {
public interface ProcessorDef {

Processor build();

}
8 changes: 4 additions & 4 deletions stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,9 +47,9 @@ private interface NodeFactory {
private class ProcessorNodeFactory implements NodeFactory {
public final String[] parents;
private final String name;
private final ProcessorFactory factory;
private final ProcessorDef factory;

public ProcessorNodeFactory(String name, String[] parents, ProcessorFactory factory) {
public ProcessorNodeFactory(String name, String[] parents, ProcessorDef factory) {
this.name = name;
this.parents = parents.clone();
this.factory = factory;
Expand Down Expand Up @@ -134,7 +134,7 @@ public final void addSink(String name, String topic, Serializer keySerializer, S
nodeFactories.add(new SinkNodeFactory(name, parentNames, topic, keySerializer, valSerializer));
}

public final void addProcessor(String name, ProcessorFactory factory, String... parentNames) {
public final void addProcessor(String name, ProcessorDef factory, String... parentNames) {
if (nodeNames.contains(name))
throw new IllegalArgumentException("Processor " + name + " is already added.");

Expand Down Expand Up @@ -186,7 +186,7 @@ public ProcessorTopology build() {
processorMap.get(parent).addChild(node);
}
} else {
throw new IllegalStateException("unknown factory class: " + factory.getClass().getName());
throw new IllegalStateException("unknown node factory class: " + factory.getClass().getName());
}
}
} catch (Exception e) {
Expand Down