Skip to content

More comments addressed #71

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

Merged
Guozhang Wang (guozhangwang) merged 1 commit into from
Sep 22, 2015
Merged

More comments addressed #71

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
2 changes: 1 addition & 1 deletion build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -538,7 +538,7 @@ project(':streams') {
compile "$slf4jlog4j"
compile 'org.rocksdb:rocksdbjni:3.10.1'

testCompile 'junit:junit:4.6'
testCompile '$junit'
testCompile project(path: ':clients', configuration: 'archives')
}

Expand Down
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ public interface KStream<K, V> {
<K1, V1> KStream<K1, V1> map(KeyValueMapper<K, V, KeyValue<K1, V1>> mapper);

/**
* Creates a new stream by transforming valuesa by a mapper to all values of this stream
* Creates a new stream by transforming values by a mapper to all values of this stream
*
* @param mapper the instance of ValueMapper
* @param <V1> the value type of the new stream
Expand Down
1 change: 0 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamFlatMapValues.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ class KStreamFlatMapValues<K1, V1, V2> implements ProcessorDef {

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

@SuppressWarnings("unchecked")
KStreamFlatMapValues(ValueMapper<V1, ? extends Iterable<V2>> mapper) {
this.mapper = mapper;
}
Expand Down
14 changes: 7 additions & 7 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ public class KStreamImpl<K, V> implements KStream<K, V> {

private static final String WINDOWED_NAME = "KAFKA-WINDOWED-";

private static final String SINK_NAME = "KAFKA-SINK-";

public static final String JOINTHIS_NAME = "KAFKA-JOINTHIS-";

public static final String JOINOTHER_NAME = "KAFKA-JOINOTHER-";
Expand All @@ -60,12 +62,10 @@ public class KStreamImpl<K, V> implements KStream<K, V> {

public static final String SOURCE_NAME = "KAFKA-SOURCE-";

public static final String SEND_NAME = "KAFKA-SEND-";

public static final AtomicInteger INDEX = new AtomicInteger(1);

protected TopologyBuilder topology;
protected String name;
protected final TopologyBuilder topology;
protected final String name;

public KStreamImpl(TopologyBuilder topology, String name) {
this.topology = topology;
Expand Down Expand Up @@ -160,7 +160,7 @@ public <K1, V1> KStream<K1, V1> through(String topic,
Serializer<V> valSerializer,
Deserializer<K1> keyDeserializer,
Deserializer<V1> valDeserializer) {
String sendName = SEND_NAME + INDEX.getAndIncrement();
String sendName = SINK_NAME + INDEX.getAndIncrement();

topology.addSink(sendName, topic, keySerializer, valSerializer, this.name);

Expand All @@ -178,14 +178,14 @@ public <K1, V1> KStream<K1, V1> through(String topic) {

@Override
public void to(String topic) {
String name = SEND_NAME + INDEX.getAndIncrement();
String name = SINK_NAME + INDEX.getAndIncrement();

topology.addSink(name, topic, this.name);
}

@Override
public void to(String topic, Serializer<K> keySerializer, Serializer<V> valSerializer) {
String name = SEND_NAME + INDEX.getAndIncrement();
String name = SINK_NAME + INDEX.getAndIncrement();

topology.addSink(name, topic, keySerializer, valSerializer, this.name);
}
Expand Down