Skip to content

KAFKA-5520: KIP-171 - Extend Consumer Group Reset Offset for Stream Application #3831

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

Closed
jeqo wants to merge 13 commits into from
Closed

KAFKA-5520: KIP-171 - Extend Consumer Group Reset Offset for Stream Application #3831

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
c330445
removing zookeeper dependency
jeqo Sep 1, 2017
0a836a6
replace seek offset with call to ConsumerGroupCommand
jeqo Sep 1, 2017
2e5bb3e
add dependency from streams to core
jeqo Sep 1, 2017
3795672
move back to core module.
jeqo Sep 11, 2017
de1f536
change execute option for dry-run
jeqo Sep 11, 2017
01a6b50
align dry-run options
jeqo Sep 11, 2017
8025153
Merge remote-tracking branch 'origin/trunk' into features/kip-171
jeqo Sep 11, 2017
88b0168
add scenario params
jeqo Sep 11, 2017
cb4ab37
map scenarios
jeqo Sep 11, 2017
bb782f8
rename integration test
jeqo Sep 11, 2017
060c897
add import control for clients.admin on core
jeqo Sep 12, 2017
4d4758b
rollback bootstrapServer option
jeqo Sep 12, 2017
cc89be1
fix test bootstrap server param
jeqo Sep 12, 2017
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
1 change: 1 addition & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -883,6 +883,7 @@ project(':streams') {
compile project(':connect:json') // this dependency should be removed after we unify data API
compile libs.slf4jApi
compile libs.rocksDBJni
compile libs.joptSimple

testCompile project(':clients').sourceSets.test.output
testCompile project(':core')
Expand Down
1 change: 1 addition & 0 deletions checkstyle/import-control-core.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
<allow pkg="kafka.producer" />
<allow pkg="kafka.consumer" />
<allow pkg="joptsimple" />
<allow pkg="org.apache.kafka.clients.admin" />
<allow pkg="org.apache.kafka.clients.consumer" />
</subpackage>

Expand Down
8 changes: 4 additions & 4 deletions core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -528,8 +528,8 @@ object ConsumerGroupCommand extends Logging {
case "Empty" | "Dead" =>
val partitionsToReset = getPartitionsToReset(groupId)
val preparedOffsets = prepareOffsetsToReset(groupId, partitionsToReset)
val execute = opts.options.has(opts.executeOpt)
if (execute)
val dryRun = opts.options.has(opts.dryRunOpt)
if (!dryRun)
getConsumer().commitSync(preparedOffsets.asJava)
preparedOffsets
case currentState =>
Expand Down Expand Up @@ -726,7 +726,7 @@ object ConsumerGroupCommand extends Logging {
"Has 3 execution options: (default) to plan which offsets to reset, --execute to execute the reset-offsets process, and --export to export the results to a CSV format." + nl +
"Has the following scenarios to choose: --to-datetime, --by-period, --to-earliest, --to-latest, --shift-by, --from-file, --to-current. One scenario must be choose" + nl +
"To define the scope use: --all-topics or --topic. . One scope must be choose, unless you use '--from-file' scenario"
val ExecuteDoc = "Execute operation. Supported operations: reset-offsets."
val DryRunDoc = "Only show results without executing changes on Consumer Groups. Supported operations: reset-offsets."
val ExportDoc = "Export operation execution to a CSV file. Supported operations: reset-offsets."
val ResetToOffsetDoc = "Reset offsets to a specific offset."
val ResetFromFileDoc = "Reset offsets to values defined in CSV file."
Expand Down Expand Up @@ -769,7 +769,7 @@ object ConsumerGroupCommand extends Logging {
.describedAs("command config property file")
.ofType(classOf[String])
val resetOffsetsOpt = parser.accepts("reset-offsets", ResetOffsetsDoc)
val executeOpt = parser.accepts("execute", ExecuteDoc)
val dryRunOpt = parser.accepts("dry-run", DryRunDoc)
val exportOpt = parser.accepts("export", ExportDoc)
val resetToOffsetOpt = parser.accepts("to-offset", ResetToOffsetDoc)
.withRequiredArg()
Expand Down
Loading