diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java
index 8601e1c6a..c5e892a09 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Joined.java
@@ -27,13 +27,36 @@ public class Joined<K, V, VO> {
 private Serde<K> keySerde;
 private Serde<V> valueSerde;
 private Serde<VO> otherValueSerde;
+ private String joinName;
 
 private Joined(final Serde<K> keySerde,
 final Serde<V> valueSerde,
- final Serde<VO> otherValueSerde) {
+ final Serde<VO> otherValueSerde,
+ final String joinName) {
 this.keySerde = keySerde;
 this.valueSerde = valueSerde;
 this.otherValueSerde = otherValueSerde;
+ this.joinName = joinName;
+ }
+
+ /**
+ * Create an instance of {@code Joined} with key, value, and otherValue {@link Serde} instances.
+ * {@code null} values are accepted and will be replaced by the default serdes as defined in config.
+ *
+ * @param keySerde the key serde to use. If {@code null} the default key serde from config will be used
+ * @param valueSerde the value serde to use. If {@code null} the default value serde from config will be used
+ * @param otherValueSerde the otherValue serde to use. If {@code null} the default value serde from config will be used
+ * @param <K> key type
+ * @param <V> value type
+ * @param <VO> other value type
+ * @param joinName name of the join used to generate backing topic names
+ * @return new {@code Joined} instance with the provided serdes
+ */
+ public static <K, V, VO> Joined<K, V, VO> with(final Serde<K> keySerde,
+ final Serde<V> valueSerde,
+ final Serde<VO> otherValueSerde,
+ final String joinName) {
+ return new Joined<>(keySerde, valueSerde, otherValueSerde, joinName);
 }
 
 /**
@@ -51,7 +74,7 @@ public class Joined<K, V, VO> {
 public static <K, V, VO> Joined<K, V, VO> with(final Serde<K> keySerde,
 final Serde<V> valueSerde,
 final Serde<VO> otherValueSerde) {
- return new Joined<>(keySerde, valueSerde, otherValueSerde);
+ return with(keySerde, valueSerde, otherValueSerde, null);
 }
 
 /**
@@ -96,6 +119,20 @@ public class Joined<K, V, VO> {
 return with(null, null, otherValueSerde);
 }
 
+ /**
+ * Create an instance of {@code Joined} with an other value {@link Serde}.
+ * {@code null} values are accepted and will be replaced by the default value serde as defined in config.
+ *
+ * @param joinName the joinName to use. If {@code null} the default value serde from config will be used
+ * @param <K> key type
+ * @param <V> value type
+ * @param <VO> other value type
+ * @return new {@code Joined} instance configured with the joinName
+ */
+ public static <K, V, VO> Joined<K, V, VO> otherValueSerde(final String joinName) {
+ return with(null, null, null, joinName);
+ }
+
 /**
 * Set the key {@link Serde} to be used. Null values are accepted and will be replaced by the default
 * key serde as defined in config
@@ -132,6 +169,18 @@ public class Joined<K, V, VO> {
 return this;
 }
 
+ /**
+ * Set the joinName to be used. Null values are accepted and will be replaced by the default
+ * value serde as defined in config
+ *
+ * @param joinName the joinName to use. If null the default joinName from config will be used
+ * @return this
+ */
+ public Joined<K, V, VO> withJoinName(final String joinName) {
+ this.joinName = joinName;
+ return this;
+ }
+
 public Serde<K> keySerde() {
 return keySerde;
 }
@@ -143,4 +192,8 @@ public class Joined<K, V, VO> {
 public Serde<VO> otherValueSerde() {
 return otherValueSerde;
 }
+
+ public String joinName() {
+ return joinName;
+ }
 }

diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index 8e80315fa..8f9df8c19 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
@@ -362,12 +362,12 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements KStream<K, V
 return branchChildren;
 }
 
- @Override 
+ @Override
 public KStream<K, V> merge(final KStream<K, V> stream) {
 Objects.requireNonNull(stream);
 return merge(builder, stream);
 }
- 
+
 private KStream<K, V> merge(final InternalStreamsBuilder builder,
 final KStream<K, V> stream) {
 KStreamImpl<K, V> streamImpl = (KStreamImpl<K, V>) stream;
@@ -880,6 +880,10 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements KStream<K, V
 this.rightOuter = rightOuter;
 }
 
+ private <K1, V1, V2> String newProcessorName(final String prefix, final Joined<K1, V1, V2> joined) {
+ return joined.joinName() != null ? (prefix + joined.joinName()) : builder.newProcessorName(prefix);
+ }
+
 @SuppressWarnings("unchecked")
 public <K1, R, V1, V2> KStream<K1, R> join(final KStream<K1, V1> lhs,
 final KStream<K1, V2> other,
@@ -888,8 +892,8 @@ public class KStreamImpl<K, V> extends AbstractStream<K> implements KStream<K, V
 final Joined<K1, V1, V2> joined) {
 String thisWindowStreamName = builder.newProcessorName(WINDOWED_NAME);
 String otherWindowStreamName = builder.newProcessorName(WINDOWED_NAME);
- String joinThisName = rightOuter ? builder.newProcessorName(OUTERTHIS_NAME) : builder.newProcessorName(JOINTHIS_NAME);
- String joinOtherName = leftOuter ? builder.newProcessorName(OUTEROTHER_NAME) : builder.newProcessorName(JOINOTHER_NAME);
+ String joinThisName = rightOuter ? newProcessorName(OUTERTHIS_NAME, joined) : newProcessorName(JOINTHIS_NAME, joined);
+ String joinOtherName = leftOuter ? newProcessorName(OUTEROTHER_NAME, joined) : newProcessorName(JOINOTHER_NAME, joined);
 String joinMergeName = builder.newProcessorName(MERGE_NAME);
 
 final StoreBuilder<WindowStore<K1, V1>> thisWindow =