You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by choojoyq <gi...@git.apache.org> on 2018/07/07 09:17:28 UTC
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
GitHub user choojoyq opened a pull request:
https://github.com/apache/storm/pull/2756
STORM-3090 - Fix bug when different topics use the same offset for a partition
This pull request contains the same changes from https://github.com/apache/storm/pull/2726 that was merged in 1.x branch. As there are some conflicts during cherry-picking of these changes into 1.0.x/1.1.x branch this pull request is created.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/choojoyq/storm STORM-3090
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/2756.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2756
----
commit 2bebe35cd15cd814db1b1587add60a056a00672b
Author: Nikita Gorbachevsky <ni...@...>
Date: 2018-07-06T17:32:50Z
STORM-3090 - use topic together with partition number during recreation of partition managers
----
---
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
Posted by choojoyq <gi...@git.apache.org>.
Github user choojoyq commented on a diff in the pull request:
https://github.com/apache/storm/pull/2756#discussion_r200964011
--- Diff: external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java ---
@@ -80,22 +81,24 @@ public void refresh() {
List<Partition> mine = KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);
Set<Partition> curr = _managers.keySet();
- Set<Partition> newPartitions = new HashSet<Partition>(mine);
+ Set<Partition> newPartitions = new HashSet<>(mine);
newPartitions.removeAll(curr);
- Set<Partition> deletedPartitions = new HashSet<Partition>(curr);
+ Set<Partition> deletedPartitions = new HashSet<>(curr);
deletedPartitions.removeAll(mine);
- LOG.info(taskId(_taskIndex, _totalTasks) + "Deleted partition managers: " + deletedPartitions.toString());
+ LOG.info("{}Deleted partition managers: {}",
--- End diff --
This space is added in ``taskId`` method.
``return "Task [" + (taskIndex + 1) + "/" + totalTasks + "] ";``
---
[GitHub] storm issue #2756: STORM-3090 - Fix bug when different topics use the same o...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2756
+1
Yes, likely. There's no reason people can't just upgrade to 1.1.x/1.2.x instead, and it's adding a bunch of work for us to have to keep backporting to old branches.
---
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2756#discussion_r200972001
--- Diff: external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java ---
@@ -80,22 +81,24 @@ public void refresh() {
List<Partition> mine = KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);
Set<Partition> curr = _managers.keySet();
- Set<Partition> newPartitions = new HashSet<Partition>(mine);
+ Set<Partition> newPartitions = new HashSet<>(mine);
newPartitions.removeAll(curr);
- Set<Partition> deletedPartitions = new HashSet<Partition>(curr);
+ Set<Partition> deletedPartitions = new HashSet<>(curr);
deletedPartitions.removeAll(mine);
- LOG.info(taskId(_taskIndex, _totalTasks) + "Deleted partition managers: " + deletedPartitions.toString());
+ LOG.info("{}Deleted partition managers: {}",
--- End diff --
Ok, thanks.
---
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2756#discussion_r200818671
--- Diff: external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java ---
@@ -80,22 +81,24 @@ public void refresh() {
List<Partition> mine = KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);
Set<Partition> curr = _managers.keySet();
- Set<Partition> newPartitions = new HashSet<Partition>(mine);
+ Set<Partition> newPartitions = new HashSet<>(mine);
newPartitions.removeAll(curr);
- Set<Partition> deletedPartitions = new HashSet<Partition>(curr);
+ Set<Partition> deletedPartitions = new HashSet<>(curr);
deletedPartitions.removeAll(mine);
- LOG.info(taskId(_taskIndex, _totalTasks) + "Deleted partition managers: " + deletedPartitions.toString());
+ LOG.info("{}Deleted partition managers: {}",
--- End diff --
Missing space
---
[GitHub] storm issue #2756: STORM-3090 - Fix bug when different topics use the same o...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/2756
Could you please point this to at least 1.1.x-branch? We are trying to maintain 3 version lines and due to this, 1.0.x version is no longer supported.
---
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2756#discussion_r200818673
--- Diff: external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java ---
@@ -80,22 +81,24 @@ public void refresh() {
List<Partition> mine = KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);
Set<Partition> curr = _managers.keySet();
- Set<Partition> newPartitions = new HashSet<Partition>(mine);
+ Set<Partition> newPartitions = new HashSet<>(mine);
newPartitions.removeAll(curr);
- Set<Partition> deletedPartitions = new HashSet<Partition>(curr);
+ Set<Partition> deletedPartitions = new HashSet<>(curr);
deletedPartitions.removeAll(mine);
- LOG.info(taskId(_taskIndex, _totalTasks) + "Deleted partition managers: " + deletedPartitions.toString());
+ LOG.info("{}Deleted partition managers: {}",
+ taskId(_taskIndex, _totalTasks), deletedPartitions);
- Map<Integer, PartitionManager> deletedManagers = new HashMap<>();
+ Map<TopicAndPartition, PartitionManager> deletedManagers = new HashMap<>();
for (Partition id : deletedPartitions) {
- deletedManagers.put(id.partition, _managers.remove(id));
+ PartitionManager manager = _managers.remove(id);
+ manager.close();
+ deletedManagers.put(new TopicAndPartition(id.topic, id.partition), manager);
}
- for (PartitionManager manager : deletedManagers.values()) {
- if (manager != null) manager.close();
- }
- LOG.info(taskId(_taskIndex, _totalTasks) + "New partition managers: " + newPartitions.toString());
+
+ LOG.info("{}New partition managers: {}",
--- End diff --
Missing space
---
[GitHub] storm issue #2756: STORM-3090 - Fix bug when different topics use the same o...
Posted by choojoyq <gi...@git.apache.org>.
Github user choojoyq commented on the issue:
https://github.com/apache/storm/pull/2756
Got it, thanks for the answer.
---
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
Posted by choojoyq <gi...@git.apache.org>.
Github user choojoyq commented on a diff in the pull request:
https://github.com/apache/storm/pull/2756#discussion_r200963984
--- Diff: external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java ---
@@ -80,22 +81,24 @@ public void refresh() {
List<Partition> mine = KafkaUtils.calculatePartitionsForTask(brokerInfo, _totalTasks, _taskIndex);
Set<Partition> curr = _managers.keySet();
- Set<Partition> newPartitions = new HashSet<Partition>(mine);
+ Set<Partition> newPartitions = new HashSet<>(mine);
newPartitions.removeAll(curr);
- Set<Partition> deletedPartitions = new HashSet<Partition>(curr);
+ Set<Partition> deletedPartitions = new HashSet<>(curr);
deletedPartitions.removeAll(mine);
- LOG.info(taskId(_taskIndex, _totalTasks) + "Deleted partition managers: " + deletedPartitions.toString());
+ LOG.info("{}Deleted partition managers: {}",
+ taskId(_taskIndex, _totalTasks), deletedPartitions);
- Map<Integer, PartitionManager> deletedManagers = new HashMap<>();
+ Map<TopicAndPartition, PartitionManager> deletedManagers = new HashMap<>();
for (Partition id : deletedPartitions) {
- deletedManagers.put(id.partition, _managers.remove(id));
+ PartitionManager manager = _managers.remove(id);
+ manager.close();
+ deletedManagers.put(new TopicAndPartition(id.topic, id.partition), manager);
}
- for (PartitionManager manager : deletedManagers.values()) {
- if (manager != null) manager.close();
- }
- LOG.info(taskId(_taskIndex, _totalTasks) + "New partition managers: " + newPartitions.toString());
+
+ LOG.info("{}New partition managers: {}",
--- End diff --
This space is added in ``taskId`` method.
``return "Task [" + (taskIndex + 1) + "/" + totalTasks + "] ";``
---
[GitHub] storm issue #2756: STORM-3090 - Fix bug when different topics use the same o...
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on the issue:
https://github.com/apache/storm/pull/2756
Thanks @choojoyq, merged.
---
[GitHub] storm issue #2756: STORM-3090 - Fix bug when different topics use the same o...
Posted by choojoyq <gi...@git.apache.org>.
Github user choojoyq commented on the issue:
https://github.com/apache/storm/pull/2756
base branch is changed to ``1.1.x-branch``. Am i understand correctly that the last release version of 1.0.x-branch was 1.0.6 and newer version will never be released?
---
[GitHub] storm pull request #2756: STORM-3090 - Fix bug when different topics use the...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/2756
---