root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

KAFKA-12824 Delete unused doBranch method (#16981) 

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
TengYao Chi 2024-08-25 14:24:54 +08:00 committed by GitHub
parent a95dfe215b
commit 3d436f52bf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 0 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
View File

@ -68,7 +68,6 @@ import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.VersionedBytesStoreSupplier; import org.apache.kafka.streams.state.VersionedBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.RocksDBTimeOrderedKeyValueBuffer; import org.apache.kafka.streams.state.internals.RocksDBTimeOrderedKeyValueBuffer;
import java.lang.reflect.Array;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collections; import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
@ -104,10 +103,6 @@ public class KStreamImpl<K, V> extends AbstractStream<K, V> implements KStream<K
static final String REPARTITION_TOPIC_SUFFIX = "-repartition"; static final String REPARTITION_TOPIC_SUFFIX = "-repartition";
private static final String BRANCH_NAME = "KSTREAM-BRANCH-";
private static final String BRANCHCHILD_NAME = "KSTREAM-BRANCHCHILD-";
private static final String FILTER_NAME = "KSTREAM-FILTER-"; private static final String FILTER_NAME = "KSTREAM-FILTER-";
private static final String PEEK_NAME = "KSTREAM-PEEK-"; private static final String PEEK_NAME = "KSTREAM-PEEK-";
@ -450,45 +445,6 @@ public class KStreamImpl<K, V> extends AbstractStream<K, V> implements KStream<K
builder); builder);
} }
@SuppressWarnings({"unchecked", "rawtypes"})
private KStream<K, V>[] doBranch(final NamedInternal named,
final Predicate<? super K, ? super V>... predicates) {
Objects.requireNonNull(predicates, "predicates can't be a null array");
if (predicates.length == 0) {
throw new IllegalArgumentException("branch() requires at least one predicate");
}
for (final Predicate<? super K, ? super V> predicate : predicates) {
Objects.requireNonNull(predicate, "predicates can't be null");
}
final String branchName = named.orElseGenerateWithPrefix(builder, BRANCH_NAME);
final String[] childNames = new String[predicates.length];
for (int i = 0; i < predicates.length; i++) {
childNames[i] = named.suffixWithOrElseGet("-predicate-" + i, builder, BRANCHCHILD_NAME);
}
final ProcessorParameters processorParameters =
new ProcessorParameters<>(new KStreamBranch(Arrays.asList(predicates.clone()),
Arrays.asList(childNames)), branchName);
final ProcessorGraphNode<K, V> branchNode =
new ProcessorGraphNode<>(branchName, processorParameters);
builder.addGraphNode(graphNode, branchNode);
final KStream<K, V>[] branchChildren = (KStream<K, V>[]) Array.newInstance(KStream.class, predicates.length);
for (int i = 0; i < predicates.length; i++) {
final ProcessorParameters innerProcessorParameters =
new ProcessorParameters<>(new PassThrough<K, V>(), childNames[i]);
final ProcessorGraphNode<K, V> branchChildNode =
new ProcessorGraphNode<>(childNames[i], innerProcessorParameters);
builder.addGraphNode(branchNode, branchChildNode);
branchChildren[i] = new KStreamImpl<>(childNames[i], keySerde, valueSerde, subTopologySourceNodes, repartitionRequired, branchChildNode, builder);
}
return branchChildren;
}
@Override @Override
public BranchedKStream<K, V> split() { public BranchedKStream<K, V> split() {
return new BranchedKStreamImpl<>(this, repartitionRequired, NamedInternal.empty()); return new BranchedKStreamImpl<>(this, repartitionRequired, NamedInternal.empty());