kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ij...@apache.org
Subject kafka git commit: MINOR: Fix ClassCastException in ConsumerGroupCommand
Date Fri, 22 Sep 2017 21:36:23 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.11.0 f16fc68ec -> 3cda63697


MINOR: Fix ClassCastException in ConsumerGroupCommand

Offset and partition are not converted from String to
long and int correctly.

Running the command line with --from-file option causes
the following exception:

java.lang.ClassCastException: java.lang.String cannot be
cast to java.lang.Integer

Reason: asInstanceOf used for the conversion.

Also, unit test is using --to-earliest and --from-file
together when executing the test. This is executing
--to-earliest option only and ignoring --from-file
option. Since the preparation part is also using
--to-earliest to create the file, this unit test
passes without testing --from-file option. Fixed
the unit test too.

Author: Erkan Unal <eunal@cisco.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #3938 from eu657/eu657-patch-1

(cherry picked from commit 71a65d95a257d98925b5f0f6f2227504cf5043a2)
Signed-off-by: Ismael Juma <ismael@juma.me.uk>


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3cda6369
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3cda6369
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3cda6369

Branch: refs/heads/0.11.0
Commit: 3cda636973e4c1bd303a7913da55d7d66c201d3d
Parents: f16fc68
Author: Erkan Unal <eunal@cisco.com>
Authored: Fri Sep 22 22:34:05 2017 +0100
Committer: Ismael Juma <ismael@juma.me.uk>
Committed: Fri Sep 22 22:36:19 2017 +0100

----------------------------------------------------------------------
 .../src/main/scala/kafka/admin/ConsumerGroupCommand.scala |  4 ++--
 .../unit/kafka/admin/ResetConsumerGroupOffsetTest.scala   | 10 +++++-----
 2 files changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/3cda6369/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
index 6690d89..d7033de 100755
--- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
@@ -564,8 +564,8 @@ object ConsumerGroupCommand extends Logging {
       resetPlanCsv.split("\n")
         .map { line =>
           val Array(topic, partition, offset) = line.split(",").map(_.trim)
-          val topicPartition = new TopicPartition(topic, partition.asInstanceOf[Int])
-          val offsetAndMetadata = new OffsetAndMetadata(offset.asInstanceOf[Long])
+          val topicPartition = new TopicPartition(topic, partition.toInt)
+          val offsetAndMetadata = new OffsetAndMetadata(offset.toLong)
           (topicPartition, offsetAndMetadata)
         }.toMap
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/3cda6369/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala
index effcd92..e8822de 100644
--- a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala
@@ -549,7 +549,7 @@ class ResetConsumerGroupOffsetTest extends KafkaServerTestHarness {
 
   @Test
   def testResetOffsetsExportImportPlan() {
-    val cgcArgs = Array("--bootstrap-server", brokerList, "--reset-offsets", "--group", group,
"--all-topics", "--to-earliest", "--export")
+    val cgcArgs = Array("--bootstrap-server", brokerList, "--reset-offsets", "--group", group,
"--all-topics", "--to-offset","2", "--export")
     val opts = new ConsumerGroupCommandOptions(cgcArgs)
     val consumerGroupCommand = createConsumerGroupService(opts)
 
@@ -564,18 +564,18 @@ class ResetConsumerGroupOffsetTest extends KafkaServerTestHarness {
       val bw = new BufferedWriter(new FileWriter(file))
       bw.write(consumerGroupCommand.exportOffsetsToReset(assignmentsToReset))
       bw.close()
-      assignmentsToReset.exists { assignment => assignment._2.offset() == 0 } &&
file.exists()
+      assignmentsToReset.exists { assignment => assignment._2.offset() == 2 } &&
file.exists()
     }, "Expected the consume all messages and save reset offsets plan to file")
 
 
-    val cgcArgsExec = Array("--bootstrap-server", brokerList, "--reset-offsets", "--group",
group, "--all-topics", "--to-earliest", "--from-file", file.getCanonicalPath)
+    val cgcArgsExec = Array("--bootstrap-server", brokerList, "--reset-offsets", "--group",
group, "--all-topics", "--from-file", file.getCanonicalPath)
     val optsExec = new ConsumerGroupCommandOptions(cgcArgsExec)
     val consumerGroupCommandExec = createConsumerGroupService(optsExec)
 
     TestUtils.waitUntilTrue(() => {
         val assignmentsToReset = consumerGroupCommandExec.resetOffsets()
-        assignmentsToReset.exists { assignment => assignment._2.offset() == 0 }
-    }, "Expected the consumer group to reset to offset 0 (earliest) by file.")
+        assignmentsToReset.exists { assignment => assignment._2.offset() == 2 }
+    }, "Expected the consumer group to reset to offset 2 according to the plan in the file.")
 
     file.deleteOnExit()
 


Mime
View raw message