You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/01/27 17:45:04 UTC
kafka git commit: MINOR: Replace for within for each;
replace if-elseif with match
Repository: kafka
Updated Branches:
refs/heads/trunk 662c2f893 -> 9213de8f1
MINOR: Replace for within for each; replace if-elseif with match
In some Places for the loop was used but it can be replaced by the for each.
In One file if else if else was used so I replaced the same with match.
Author: Akash Sethi <ak...@knoldus.in>
Reviewers: Guozhang Wang <wa...@gmail.com>
Closes #2435 from akashsethi24/trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/9213de8f
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/9213de8f
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/9213de8f
Branch: refs/heads/trunk
Commit: 9213de8f14dfd2754d913667ae9a703f76277022
Parents: 662c2f8
Author: Akash Sethi <ak...@knoldus.in>
Authored: Fri Jan 27 09:45:01 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Fri Jan 27 09:45:01 2017 -0800
----------------------------------------------------------------------
.../kafka/common/network/ByteBufferReceive.java | 4 +--
.../kafka/admin/ConsumerGroupCommand.scala | 31 ++++++++++----------
.../kstream/internals/KTableKTableJoinTest.java | 16 +++++-----
3 files changed, 26 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/9213de8f/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
index 159c301..307eadf 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ByteBufferReceive.java
@@ -33,8 +33,8 @@ public class ByteBufferReceive implements Receive {
super();
this.source = source;
this.buffers = buffers;
- for (int i = 0; i < buffers.length; i++)
- remaining += buffers[i].remaining();
+ for (ByteBuffer buffer : buffers)
+ remaining += buffer.remaining();
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/9213de8f/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 d4e407a..c3d9f24 100755
--- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
@@ -306,22 +306,23 @@ object ConsumerGroupCommand extends Logging {
val offsetFetchResponse = OffsetFetchResponse.readFrom(channel.receive().payload())
offsetFetchResponse.requestInfo.foreach { case (topicAndPartition, offsetAndMetadata) =>
- if (offsetAndMetadata == OffsetMetadataAndError.NoOffset) {
- val topicDirs = new ZKGroupTopicDirs(group, topicAndPartition.topic)
- // this group may not have migrated off zookeeper for offsets storage (we don't expose the dual-commit option in this tool
- // (meaning the lag may be off until all the consumers in the group have the same setting for offsets storage)
- try {
- val offset = zkUtils.readData(topicDirs.consumerOffsetDir + "/" + topicAndPartition.partition)._1.toLong
- offsetMap.put(topicAndPartition, offset)
- } catch {
- case z: ZkNoNodeException =>
- printError(s"Could not fetch offset from zookeeper for group '$group' partition '$topicAndPartition' due to missing offset data in zookeeper.", Some(z))
- }
+ offsetAndMetadata match {
+ case OffsetMetadataAndError.NoOffset =>
+ val topicDirs = new ZKGroupTopicDirs(group, topicAndPartition.topic)
+ // this group may not have migrated off zookeeper for offsets storage (we don't expose the dual-commit option in this tool
+ // (meaning the lag may be off until all the consumers in the group have the same setting for offsets storage)
+ try {
+ val offset = zkUtils.readData(topicDirs.consumerOffsetDir + "/" + topicAndPartition.partition)._1.toLong
+ offsetMap.put(topicAndPartition, offset)
+ } catch {
+ case z: ZkNoNodeException =>
+ printError(s"Could not fetch offset from zookeeper for group '$group' partition '$topicAndPartition' due to missing offset data in zookeeper.", Some(z))
+ }
+ case offsetAndMetaData if offsetAndMetaData.error == Errors.NONE.code =>
+ offsetMap.put(topicAndPartition, offsetAndMetadata.offset)
+ case _ =>
+ printError(s"Could not fetch offset from kafka for group '$group' partition '$topicAndPartition' due to ${Errors.forCode(offsetAndMetadata.error).exception}.")
}
- else if (offsetAndMetadata.error == Errors.NONE.code)
- offsetMap.put(topicAndPartition, offsetAndMetadata.offset)
- else
- printError(s"Could not fetch offset from kafka for group '$group' partition '$topicAndPartition' due to ${Errors.forCode(offsetAndMetadata.error).exception}.")
}
channel.disconnect()
offsetMap.toMap
http://git-wip-us.apache.org/repos/asf/kafka/blob/9213de8f/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
index 4bb64b8..a938605 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinTest.java
@@ -119,8 +119,8 @@ public class KTableKTableJoinTest {
// push all four items to the primary stream. this should produce two items.
- for (int i = 0; i < expectedKeys.length; i++) {
- driver.process(topic1, expectedKeys[i], "XX" + expectedKeys[i]);
+ for (int expectedKey : expectedKeys) {
+ driver.process(topic1, expectedKey, "XX" + expectedKey);
}
driver.flushState();
@@ -128,8 +128,8 @@ public class KTableKTableJoinTest {
checkJoinedValues(getter, kv(0, "XX0+Y0"), kv(1, "XX1+Y1"));
// push all items to the other stream. this should produce four items.
- for (int i = 0; i < expectedKeys.length; i++) {
- driver.process(topic2, expectedKeys[i], "YY" + expectedKeys[i]);
+ for (int expectedKey : expectedKeys) {
+ driver.process(topic2, expectedKey, "YY" + expectedKey);
}
driver.flushState();
@@ -138,8 +138,8 @@ public class KTableKTableJoinTest {
// push all four items to the primary stream. this should produce four items.
- for (int i = 0; i < expectedKeys.length; i++) {
- driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
+ for (int expectedKey : expectedKeys) {
+ driver.process(topic1, expectedKey, "X" + expectedKey);
}
driver.flushState();
@@ -311,8 +311,8 @@ public class KTableKTableJoinTest {
// push all four items to the primary stream. this should produce four items.
- for (int i = 0; i < expectedKeys.length; i++) {
- driver.process(topic1, expectedKeys[i], "X" + expectedKeys[i]);
+ for (int expectedKey : expectedKeys) {
+ driver.process(topic1, expectedKey, "X" + expectedKey);
}
driver.flushState();
proc.checkAndClearProcessResult("0:(X0+YY0<-XX0+YY0)", "1:(X1+YY1<-XX1+YY1)", "2:(X2+YY2<-XX2+YY2)", "3:(X3+YY3<-XX3+YY3)");