Skip to content

KAFKA-6054: Fix upgrade path from Kafka Streams v0.10.0 #4761

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
mjsax merged 4 commits into from
Mar 27, 2018
Merged

KAFKA-6054: Fix upgrade path from Kafka Streams v0.10.0 #4761

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7a3ccdc
KAFKA-6054: Fix upgrade path from Kafka Streams v0.10.0
mjsax Mar 18, 2018
0a2936c
Extends system tests for 0.11
mjsax Mar 22, 2018
cfc3e6a
Update upgrade docs
mjsax Mar 23, 2018
31dca1a
fix doc typo and review comments
mjsax Mar 27, 2018
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
40 changes: 31 additions & 9 deletions bin/kafka-run-class.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,28 +73,50 @@ do
fi
done

for file in "$base_dir"/clients/build/libs/kafka-clients*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
clients_lib_dir=$(dirname $0)/../clients/build/libs
streams_lib_dir=$(dirname $0)/../streams/build/libs
rocksdb_lib_dir=$(dirname $0)/../streams/build/dependant-libs-${SCALA_VERSION}
else
clients_lib_dir=/opt/kafka-$UPGRADE_KAFKA_STREAMS_TEST_VERSION/libs
streams_lib_dir=$clients_lib_dir
rocksdb_lib_dir=$streams_lib_dir
fi


for file in "$base_dir"/streams/build/libs/kafka-streams*.jar;
for file in "$clients_lib_dir"/kafka-clients*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done

for file in "$base_dir"/streams/examples/build/libs/kafka-streams-examples*.jar;
for file in "$streams_lib_dir"/kafka-streams*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done

for file in "$base_dir"/streams/build/dependant-libs-${SCALA_VERSION}/rocksdb*.jar;
if [ -z "$UPGRADE_KAFKA_STREAMS_TEST_VERSION" ]; then
for file in "$base_dir"/streams/examples/build/libs/kafka-streams-examples*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
else
VERSION_NO_DOTS=`echo $UPGRADE_KAFKA_STREAMS_TEST_VERSION | sed 's/\.//g'`
SHORT_VERSION_NO_DOTS=${VERSION_NO_DOTS:0:((${#VERSION_NO_DOTS} - 1))} # remove last char, ie, bug-fix number
for file in "$base_dir"/streams/upgrade-system-tests-$SHORT_VERSION_NO_DOTS/build/libs/kafka-streams-upgrade-system-tests*.jar;
do
if should_include_file "$file"; then
CLASSPATH="$CLASSPATH":"$file"
fi
done
fi

for file in "$rocksdb_lib_dir"/rocksdb*.jar;
do
CLASSPATH="$CLASSPATH":"$file"
done
Expand Down
36 changes: 36 additions & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -909,6 +909,42 @@ project(':streams:examples') {
}
}

project(':streams:upgrade-system-tests-0100') {
archivesBaseName = "kafka-streams-upgrade-system-tests-0100"

dependencies {
testCompile libs.kafkaStreams_0100
}

systemTestLibs {
dependsOn testJar
}
}

project(':streams:upgrade-system-tests-0101') {
archivesBaseName = "kafka-streams-upgrade-system-tests-0101"

dependencies {
testCompile libs.kafkaStreams_0101
}

systemTestLibs {
dependsOn testJar
}
}

project(':streams:upgrade-system-tests-0102') {
archivesBaseName = "kafka-streams-upgrade-system-tests-0102"

dependencies {
testCompile libs.kafkaStreams_0102
}

systemTestLibs {
dependsOn testJar
}
}

project(':jmh-benchmarks') {

apply plugin: 'com.github.johnrengelman.shadow'
Expand Down
9 changes: 4 additions & 5 deletions ...c/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,9 @@
*/
package org.apache.kafka.common.security.authenticator;

import java.util.Map;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.security.auth.AuthCallbackHandler;

import javax.security.auth.Subject;
import javax.security.auth.callback.Callback;
Expand All @@ -25,10 +27,7 @@
import javax.security.auth.callback.UnsupportedCallbackException;
import javax.security.sasl.AuthorizeCallback;
import javax.security.sasl.RealmCallback;

import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.security.auth.AuthCallbackHandler;
import java.util.Map;

/**
* Callback handler for Sasl clients. The callbacks required for the SASL mechanism
Expand Down
27 changes: 22 additions & 5 deletions docs/streams/upgrade-guide.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,16 +27,33 @@ <h1>Upgrade Guide &amp; API Changes</h1>
</p>

<p>
If you want to upgrade from 0.10.1.x to 0.10.2, see the <a href="/{{version}}/documentation/#upgrade_1020_streams"><b>Upgrade Section for 0.10.2</b></a>.
If you want to upgrade from 0.10.1.x to 0.11.0, see the <a href="/{{version}}/documentation/#upgrade_1020_streams"><b>Upgrade Section for 0.10.2</b></a>.
It highlights incompatible changes you need to consider to upgrade your code and application.
See <a href="#streams_api_changes_0102">below</a> a complete list of 0.10.2 API and semantical changes that allow you to advance your application and/or simplify your code base, including the usage of new features.
See below a complete list of <a href="#streams_api_changes_0102">0.10.2</a> and <a href="#streams_api_changes_0110">0.11.0</a> API and semantical changes
that allow you to advance your application and/or simplify your code base, including the usage of new features.
</p>

<p>
If you want to upgrade from 0.10.0.x to 0.10.1, see the <a href="/{{version}}/documentation/#upgrade_1010_streams"><b>Upgrade Section for 0.10.1</b></a>.
It highlights incompatible changes you need to consider to upgrade your code and application.
See <a href="#streams_api_changes_0101">below</a> a complete list of 0.10.1 API changes that allow you to advance your application and/or simplify your code base, including the usage of new features.
Upgrading from 0.10.0.x to 0.11.0.x directly is also possible.
Note, that a brokers must be on version 0.10.1 or higher to run a Kafka Streams application version 0.10.1 or higher.
See <a href="#streams_api_changes_0101">Streams API changes in 0.10.1</a>, <a href="#streams_api_changes_0102">Streams API changes in 0.10.2</a>,
and <a href="#streams_api_changes_0110">Streams API changes in 0.11.0</a> for a complete list of API changes.
Upgrading to 0.11.0.3 requires two rolling bounces with config <code>upgrade.from="0.10.0"</code> set for first upgrade phase
(cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade">KIP-268</a>).
As an alternative, an offline upgrade is also possible.
</p>
<ul>
<li> prepare your application instances for a rolling bounce and make sure that config <code>upgrade.from</code> is set to <code>"0.10.0"</code> for new version 0.11.0.3 </li>
<li> bounce each instance of your application once </li>
<li> prepare your newly deployed 0.11.0.3 application instances for a second round of rolling bounces; make sure to remove the value for config <code>upgrade.mode</code> </li>
<li> bounce each instance of your application once more to complete the upgrade </li>
</ul>
<p> Upgrading from 0.10.0.x to 0.11.0.0, 0.11.0.1, or 0.11.0.2 requires an offline upgrade (rolling bounce upgrade is not supported) </p>
<ul>
<li> stop all old (0.10.0.x) application instances </li>
<li> update your code and swap old code and jar file with new code and new jar file </li>
<li> restart all new (0.11.0.0, 0.11.0.1, or 0.11.0.2) application instances </li>
</ul>

<h3><a id="streams_api_changes_0110" href="#streams_api_changes_0110">Streams API changes in 0.11.0.0</a></h3>

Expand Down
54 changes: 52 additions & 2 deletions docs/upgrade.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,12 @@ <h4><a id="upgrade_11_0_0" href="#upgrade_11_0_0">Upgrading from 0.8.x, 0.9.x, 0
before you switch to 0.11.0.</li>
</ol>

<h5><a id="upgrade_1103_notable" href="#upgrade_1103_notable">Notable changes in 0.11.0.3</a></h5>
<ul>
<li> New Kafka Streams configuration parameter <code>upgrade.from</code> added that allows rolling bounce upgrade from version 0.10.0.x </li>
<li> See the <a href="/{{version}}/documentation/streams/upgrade-guide.html"><b>Kafka Streams upgrade guide</b></a> for details about this new config.
</ul>

<h5><a id="upgrade_1100_notable" href="#upgrade_1100_notable">Notable changes in 0.11.0.0</a></h5>
<ul>
<li>Unclean leader election is now disabled by default. The new default favors durability over availability. Users who wish to
Expand Down Expand Up @@ -214,14 +220,41 @@ <h5><a id="upgrade_1020_streams" href="#upgrade_1020_streams">Upgrading a 0.10.1
<li> See <a href="/{{version}}/documentation/streams#streams_api_changes_0102">Streams API changes in 0.10.2</a> for more details. </li>
</ul>

<h5><a id="upgrade_1020_streams_from_0100" href="#upgrade_1020_streams_from_0100">Upgrading a 0.10.0 Kafka Streams Application</a></h5>
<ul>
<li> Upgrading your Streams application from 0.10.0 to 0.10.2 does require a <a href="#upgrade_10_1">broker upgrade</a> because a Kafka Streams 0.10.2 application can only connect to 0.10.2 or 0.10.1 brokers. </li>
<li> There are couple of API changes, that are not backward compatible (cf. <a href="/{{version}}/documentation/streams#streams_api_changes_0102">Streams API changes in 0.10.2</a> for more details).

@guozhangwang guozhangwang Mar 26, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we just ref-link the streams/upgrade-guide.html here than copying the content again in the main upgrade section?

@mjsax mjsax Mar 27, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can. It's "break" the established pattern though. Up to now, those older sections are just the unchanged release notes from the version before -- I just followed the structure of the existing docs. Let me know what you think. Also, I am wondering, it this changes anyway if we rework the web page from scratch as anticipated.

@guozhangwang guozhangwang Mar 27, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good. We can leave it as is then.

Thus, you need to update and recompile your code. Just swapping the Kafka Streams library jar file will not work and will break your application. </li>
<li> Upgrading from 0.10.0.x to 0.10.2.2 requires two rolling bounces with config <code>upgrade.from="0.10.0"</code> set for first upgrade phase
(cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade">KIP-268</a>).
As an alternative, an offline upgrade is also possible.
<ul>
<li> prepare your application instances for a rolling bounce and make sure that config <code>upgrade.from</code> is set to <code>"0.10.0"</code> for new version 0.10.2.2 </li>
<li> bounce each instance of your application once </li>
<li> prepare your newly deployed 0.10.2.2 application instances for a second round of rolling bounces; make sure to remove the value for config <code>upgrade.mode</code> </li>
<li> bounce each instance of your application once more to complete the upgrade </li>
</ul>
</li>
<li> Upgrading from 0.10.0.x to 0.10.2.0 or 0.10.2.1 requires an offline upgrade (rolling bounce upgrade is not supported)
<ul>
<li> stop all old (0.10.0.x) application instances </li>
<li> update your code and swap old code and jar file with new code and new jar file </li>
<li> restart all new (0.10.2.0 or 0.10.2.1) application instances </li>
</ul>
</li>
</ul>

<h5><a id="upgrade_10202_notable" href="#upgrade_10202_notable">Notable changes in 0.10.2.2</a></h5>
<ul>
<li> New configuration parameter <code>upgrade.from</code> added that allows rolling bounce upgrade from version 0.10.0.x </li>
</ul>

<h5><a id="upgrade_10201_notable" href="#upgrade_10201_notable">Notable changes in 0.10.2.1</a></h5>
<ul>
<li> The default values for two configurations of the StreamsConfig class were changed to improve the resiliency of Kafka Streams applications. The internal Kafka Streams producer <code>retries</code> default value was changed from 0 to 10. The internal Kafka Streams consumer <code>max.poll.interval.ms</code> default value was changed from 300000 to <code>Integer.MAX_VALUE</code>.
</li>
</ul>



<h5><a id="upgrade_1020_notable" href="#upgrade_1020_notable">Notable changes in 0.10.2.0</a></h5>
<ul>
<li>The Java clients (producer and consumer) have acquired the ability to communicate with older brokers. Version 0.10.2 clients
Expand Down Expand Up @@ -294,6 +327,23 @@ <h5><a id="upgrade_1010_streams" href="#upgrade_1010_streams">Upgrading a 0.10.0
<li> Upgrading your Streams application from 0.10.0 to 0.10.1 does require a <a href="#upgrade_10_1">broker upgrade</a> because a Kafka Streams 0.10.1 application can only connect to 0.10.1 brokers. </li>
<li> There are couple of API changes, that are not backward compatible (cf. <a href="/{{version}}/documentation/streams#streams_api_changes_0101">Streams API changes in 0.10.1</a> for more details).
Thus, you need to update and recompile your code. Just swapping the Kafka Streams library jar file will not work and will break your application. </li>
<li> Upgrading from 0.10.0.x to 0.10.1.2 requires two rolling bounces with config <code>upgrade.from="0.10.0"</code> set for first upgrade phase

@guozhangwang guozhangwang Mar 26, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Strictly speaking we cannot yet talk about 0.10.1.2 in 0.11.0.3 yet since we may not release 0.10.1.2 before releasing 0.11.0.3. I think it may not be of a big issue still leaving this here, but just try to point it out.

@mjsax mjsax Mar 27, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right -- not sure how to resolve this... Maybe we can only fix this manually when an older bug fix release is done... It's a little annoying :(

(cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade">KIP-268</a>).
As an alternative, an offline upgrade is also possible.
<ul>
<li> prepare your application instances for a rolling bounce and make sure that config <code>upgrade.from</code> is set to <code>"0.10.0"</code> for new version 0.10.1.2 </li>
<li> bounce each instance of your application once </li>
<li> prepare your newly deployed 0.10.1.2 application instances for a second round of rolling bounces; make sure to remove the value for config <code>upgrade.mode</code> </li>
<li> bounce each instance of your application once more to complete the upgrade </li>
</ul>
</li>
<li> Upgrading from 0.10.0.x to 0.10.1.0 or 0.10.1.1 requires an offline upgrade (rolling bounce upgrade is not supported)
<ul>
<li> stop all old (0.10.0.x) application instances </li>
<li> update your code and swap old code and jar file with new code and new jar file </li>
<li> restart all new (0.10.1.0 or 0.10.1.1) application instances </li>
</ul>
</li>
</ul>

<h5><a id="upgrade_1010_notable" href="#upgrade_1010_notable">Notable changes in 0.10.1.0</a></h5>
Expand Down
6 changes: 6 additions & 0 deletions gradle/dependencies.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ versions += [
jackson: "2.8.5",
jetty: "9.2.22.v20170606",
jersey: "2.24",
kafka_0100: "0.10.0.1",
kafka_0101: "0.10.1.1",
kafka_0102: "0.10.2.1",
log4j: "1.2.17",
jopt: "5.0.3",
junit: "4.12",
Expand Down Expand Up @@ -96,6 +99,9 @@ libs += [
junit: "junit:junit:$versions.junit",
log4j: "log4j:log4j:$versions.log4j",
joptSimple: "net.sf.jopt-simple:jopt-simple:$versions.jopt",
kafkaStreams_0100: "org.apache.kafka:kafka-streams:$versions.kafka_0100",
kafkaStreams_0101: "org.apache.kafka:kafka-streams:$versions.kafka_0101",
kafkaStreams_0102: "org.apache.kafka:kafka-streams:$versions.kafka_0102",
lz4: "net.jpountz.lz4:lz4:$versions.lz4",
metrics: "com.yammer.metrics:metrics-core:$versions.metrics",
powermock: "org.powermock:powermock-module-junit4:$versions.powermock",
Expand Down
3 changes: 2 additions & 1 deletion settings.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.

include 'core', 'examples', 'clients', 'tools', 'streams', 'streams:examples', 'log4j-appender',
include 'core', 'examples', 'clients', 'tools', 'streams', 'streams:examples', 'streams:upgrade-system-tests-0100',
'streams:upgrade-system-tests-0101', 'streams:upgrade-system-tests-0102', 'log4j-appender',
'connect:api', 'connect:transforms', 'connect:runtime', 'connect:json', 'connect:file', 'jmh-benchmarks'
17 changes: 17 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,11 @@ public class StreamsConfig extends AbstractConfig {
*/
public static final String PRODUCER_PREFIX = "producer.";

/**
* Config value for parameter {@link #UPGRADE_FROM_CONFIG "upgrade.from"} for upgrading an application from version {@code 0.10.0.x}.
*/
public static final String UPGRADE_FROM_0100 = "0.10.0";

/**
* Config value for parameter {@link #PROCESSING_GUARANTEE_CONFIG "processing.guarantee"} for at-least-once processing guarantees.
*/
Expand Down Expand Up @@ -247,6 +252,11 @@ public class StreamsConfig extends AbstractConfig {
public static final String TIMESTAMP_EXTRACTOR_CLASS_CONFIG = "timestamp.extractor";
private static final String TIMESTAMP_EXTRACTOR_CLASS_DOC = "Timestamp extractor class that implements the <code>TimestampExtractor</code> interface. This config is deprecated, use <code>" + DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG + "</code> instead";

/** {@code upgrade.from} */
public static final String UPGRADE_FROM_CONFIG = "upgrade.from";
public static final String UPGRADE_FROM_DOC = "Allows upgrading from version 0.10.0 to version 0.10.1 (or newer) in a backward compatible way. " +
"Default is null. Accepted values are \"" + UPGRADE_FROM_0100 + "\" (for upgrading from 0.10.0.x).";

/**
* {@code value.serde}
* @deprecated Use {@link #DEFAULT_VALUE_SERDE_CLASS_CONFIG} instead.
Expand Down Expand Up @@ -466,6 +476,12 @@ public class StreamsConfig extends AbstractConfig {
null,
Importance.LOW,
TIMESTAMP_EXTRACTOR_CLASS_DOC)
.define(UPGRADE_FROM_CONFIG,
ConfigDef.Type.STRING,
null,
in(null, UPGRADE_FROM_0100),
Importance.LOW,
UPGRADE_FROM_DOC)
.define(VALUE_SERDE_CLASS_CONFIG,
Type.CLASS,
null,
Expand Down Expand Up @@ -632,6 +648,7 @@ public Map<String, Object> getConsumerConfigs(final StreamThread streamThread,
consumerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-consumer");

// add configs required for stream partition assignor
consumerProps.put(UPGRADE_FROM_CONFIG, getString(UPGRADE_FROM_CONFIG));
consumerProps.put(InternalConfig.STREAM_THREAD_INSTANCE, streamThread);
consumerProps.put(REPLICATION_FACTOR_CONFIG, getInt(REPLICATION_FACTOR_CONFIG));
consumerProps.put(NUM_STANDBY_REPLICAS_CONFIG, getInt(NUM_STANDBY_REPLICAS_CONFIG));
Expand Down
Loading