This is an automated email from the ASF dual-hosted git repository.
guozhang pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.0 by this push:
new 26f6ed0 MINOR: (doc) Disable spurious left/outer stream-stream join
fix (#11247)
26f6ed0 is described below
commit 26f6ed06589e0c2482f0cb0ce3fbcffa0f6f8d03
Author: Luke Chen <[email protected]>
AuthorDate: Tue Aug 24 02:59:55 2021 +0800
MINOR: (doc) Disable spurious left/outer stream-stream join fix (#11247)
This is the doc rollback for #11233
---
docs/streams/developer-guide/dsl-api.html | 82 +++++--------------------------
docs/streams/upgrade-guide.html | 7 ---
2 files changed, 11 insertions(+), 78 deletions(-)
diff --git a/docs/streams/developer-guide/dsl-api.html
b/docs/streams/developer-guide/dsl-api.html
index 45affd9..151e515 100644
--- a/docs/streams/developer-guide/dsl-api.html
+++ b/docs/streams/developer-guide/dsl-api.html
@@ -1799,7 +1799,7 @@ KStream<String, Double> right = ...;
// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.join(right,
(leftValue, rightValue) -> "left=" + leftValue + ",
right=" + rightValue, /* ValueJoiner */
- JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
+ JoinWindows.of(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
@@ -1814,7 +1814,7 @@ KStream<String, String> joined = left.join(right,
return "left=" + leftValue + ", right=" +
rightValue;
}
},
- JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
+ JoinWindows.of(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
@@ -1856,7 +1856,7 @@ KStream<String, Double> right = ...;
// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.leftJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ",
right=" + rightValue, /* ValueJoiner */
- JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
+ JoinWindows.of(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
@@ -1871,7 +1871,7 @@ KStream<String, String> joined =
left.leftJoin(right,
return "left=" + leftValue + ", right=" +
rightValue;
}
},
- JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
+ JoinWindows.of(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
@@ -1891,8 +1891,7 @@ KStream<String, String> joined =
left.leftJoin(right,
</div></blockquote>
</li>
<li><p class="first">For each input
record on the left side that does not have any match on the right side, the
<code class="docutils literal"><span class="pre">ValueJoiner</span></code> will
be called with <code class="docutils literal"><span
class="pre">ValueJoiner#apply(leftRecord.value,</span> <span
class="pre">null)</span></code>;
- this explains the row with
timestamp=60 and timestampe=80 in the table below, which lists <code
class="docutils literal"><span class="pre">[E,</span> <span
class="pre">null]</span></code> and <code class="docutils literal"><span
class="pre">[F,</span> <span class="pre">null]</span></code>in the LEFT JOIN
column.
- Note that these left results are
emitted after the specified grace period passed. <strong>Caution:</strong>
using the deprecated <code class="docutils literal"><span
class="pre">JoinWindows.of(...).grace(...)</span></code> API might result in
eagerly emitted spurious left results.</p>
+ this explains the row with
timestamp=3 in the table below, which lists <code class="docutils
literal"><span class="pre">[A,</span> <span class="pre">null]</span></code> in
the LEFT JOIN column.</p>
</li>
</ul>
<p class="last">See the semantics overview
at the bottom of this section for a detailed description.</p>
@@ -1917,7 +1916,7 @@ KStream<String, Double> right = ...;
// Java 8+ example, using lambda expressions
KStream<String, String> joined = left.outerJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ",
right=" + rightValue, /* ValueJoiner */
- JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
+ JoinWindows.of(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
@@ -1932,7 +1931,7 @@ KStream<String, String> joined =
left.outerJoin(right,
return "left=" + leftValue + ", right=" +
rightValue;
}
},
- JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
+ JoinWindows.of(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
@@ -1952,9 +1951,8 @@ KStream<String, String> joined =
left.outerJoin(right,
</div></blockquote>
</li>
<li><p class="first">For each input
record on one side that does not have any match on the other side, the <code
class="docutils literal"><span class="pre">ValueJoiner</span></code> will be
called with <code class="docutils literal"><span
class="pre">ValueJoiner#apply(leftRecord.value,</span> <span
class="pre">null)</span></code> or
- <code class="docutils
literal"><span class="pre">ValueJoiner#apply(null,</span> <span
class="pre">rightRecord.value)</span></code>, respectively; this explains the
row with timestamp=60, timestamp=80, and timestamp=100 in the table below,
which lists <code class="docutils literal"><span class="pre">[E,</span> <span
class="pre">null]</span></code>,
- <code class="docutils
literal"><span class="pre">[F,</span> <span class="pre">null]</span></code>,
and <code class="docutils literal"><span class="pre">[null,</span> <span
class="pre">f]</span></code> in the OUTER JOIN column.
- Note that these left and right
results are emitted after the specified grace period passed.
<strong>Caution:</strong> using the deprecated <code class="docutils
literal"><span class="pre">JoinWindows.of(...).grace(...)</span></code> API
might result in eagerly emitted spurious left/right results.</p>
+ <code class="docutils
literal"><span class="pre">ValueJoiner#apply(null,</span> <span
class="pre">rightRecord.value)</span></code>, respectively; this explains the
row with timestamp=3 in the table below, which lists <code class="docutils
literal"><span class="pre">[A,</span> <span class="pre">null]</span></code> in
the OUTER JOIN column
+ (unlike LEFT JOIN, <code
class="docutils literal"><span class="pre">[null,</span> <span
class="pre">x]</span></code> is possible, too, but no such example is shown in
the table).</p>
</li>
</ul>
<p class="last">See the semantics overview
at the bottom of this section for a detailed description.</p>
@@ -1966,8 +1964,6 @@ KStream<String, String> joined =
left.outerJoin(right,
The semantics of the various stream-stream join
variants are explained below.
To improve the readability of the table, assume
that (1) all records have the same key (and thus the key in the table is
omitted), and (2) all records are processed in timestamp order.
We assume a join window size of 10 seconds with a
grace period of 5 seconds.</p>
- <p><b>Note:</b> If you use the old and now deprecated
API to specify the grace period, i.e., <code class="docutils literal"><span
class="pre">JoinWindows.of(...).grace(...)</span></code>,
- left/outer join results are emitted eagerly, and
the observed result might differ from the result shown below.</p>
<p>The columns INNER JOIN, LEFT JOIN, and OUTER JOIN
denote what is passed as arguments to the user-supplied
<a class="reference external"
href="/{{version}}/javadoc/org/apache/kafka/streams/kstream/ValueJoiner.html">ValueJoiner</a>
for the <code class="docutils literal"><span class="pre">join</span></code>,
<code class="docutils literal"><span class="pre">leftJoin</span></code>, and
<code class="docutils literal"><span
class="pre">outerJoin</span></code> methods, respectively, whenever a new input
record is received on either side of the join. An empty table
@@ -2001,8 +1997,8 @@ KStream<String, String> joined =
left.outerJoin(right,
<td>A</td>
<td> </td>
<td> </td>
- <td> </td>
- <td> </td>
+ <td>[A, null]</td>
+ <td>[A, null]</td>
</tr>
<tr class="row-odd"><td>4</td>
<td> </td>
@@ -2088,62 +2084,6 @@ KStream<String, String> joined =
left.outerJoin(right,
<td>[D, a], [D, b], [D, c], [D, d]</td>
<td>[D, a], [D, b], [D, c], [D, d]</td>
</tr>
- <tr class="row-odd"><td> ... </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- </tr>
- <tr class="even-row"><td>40</td>
- <td>E</td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- </tr>
- <tr class="row-odd"><td> ... </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- </tr>
- <tr class="even-row"><td>60</td>
- <td>F</td>
- <td> </td>
- <td> </td>
- <td>[E, null]</td>
- <td>[E, null]</td>
- </tr>
- <tr class="row-odd"><td> ... </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- </tr>
- <tr class="even-row"><td>80</td>
- <td> </td>
- <td>f</td>
- <td> </td>
- <td>[F, null]</td>
- <td>[F, null]</td>
- </tr>
- <tr class="row-odd"><td> ... </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td> </td>
- </tr>
- <tr class="even-row"><td>100</td>
- <td>G</td>
- <td> </td>
- <td> </td>
- <td> </td>
- <td>[null, f]</td>
- </tr>
</tbody>
</table>
</div>
diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index ecc8cb8..59e46a2 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -166,13 +166,6 @@
<a
href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-633%3A+Drop+24+hour+default+of+grace+period+in+Streams";>KIP-633</a>
</p>
<p>
- Additionally, in older versions Kafka Streams emitted stream-stream
left/outer join results eagerly. This behavior may lead to spurious left/outer
join result records.
- In this release, we changed the behavior to avoid spurious results and
left/outer join result are only emitted after the join window is closed, i.e.,
after the grace period elapsed.
- To maintain backward compatibility, the old API
<code>JoinWindows#of(timeDifference)</code> preserves the old eager-emit
behavior and only the new
- APIs <code>JoinWindows#ofTimeDifferenceAndGrace()</code> and
<code>JoinsWindows#ofTimeDifferenceNoGrace</code> enable the new behavior.
Check out
- <a
href="https://issues.apache.org/jira/browse/KAFKA-10847";>KAFKA-10847</a> for
more information.
- </p>
- <p>
The public <code>topicGroupId</code> and <code>partition</code> fields
on TaskId have been deprecated and replaced with getters. Please migrate to
using the new <code>TaskId.subtopology()</code>
(which replaces <code>topicGroupId</code>) and
<code>TaskId.partition()</code> APIs instead. Also, the
<code>TaskId#readFrom</code> and <code>TaskId#writeTo</code> methods have been
deprecated
and will be removed, as they were never intended for public use. We
have also deprecated the
<code>org.apache.kafka.streams.processsor.TaskMetadata</code> class and
introduced a new interface
