[jira] [Reopened] (STORM-3090) The same offset value is used by the same partition number of different topics.

["Stig Rohde Døssing (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/STORM-3090?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Stig Rohde Døssing reopened STORM-3090:
---------------------------------------

> The same offset value is used by the same partition number of different topics.
> -------------------------------------------------------------------------------
>
>                 Key: STORM-3090
>                 URL: https://issues.apache.org/jira/browse/STORM-3090
>             Project: Apache Storm
>          Issue Type: Bug
>          Components: storm-kafka
>    Affects Versions: 1.1.0, 1.0.4, 1.2.2
>            Reporter: Nikita Gorbachevski
>            Assignee: Nikita Gorbachevski
>            Priority: Critical
>              Labels: pull-request-available
>             Fix For: 1.2.3, 1.1.4
>
>          Time Spent: 4h 10m
>  Remaining Estimate: 0h
>
> In the current implementation of `ZkCoordinator` deleted partition managers are used as state holders for newly created partition managers. This behaviour was introduced in the scope of [this|https://issues-test.apache.org/jira/browse/STORM-2296] ticket. However existing lookup is based on only on partition number.
> {code:java}
> Map<Integer, PartitionManager> deletedManagers = new HashMap<>();
> for (Partition id : deletedPartitions) {
>  deletedManagers.put(id.partition, _managers.remove(id));
> }
> for (PartitionManager manager : deletedManagers.values()) {
>  if (manager != null) manager.close();
> }
> LOG.info(taskPrefix(_taskIndex, _totalTasks, _taskId) + " New partition managers: " + newPartitions.toString());
> for (Partition id : newPartitions) {
>  PartitionManager man = new PartitionManager(
>  _connections,
>  _topologyInstanceId,
>  _state,
>  _topoConf,
>  _spoutConfig,
>  id,
>  deletedManagers.get(id.partition));
>  _managers.put(id, man);
> {code}
> Which is definitely incorrect as the same task is able to manage multiple partitions with the same number but for different topics. In this case all new partition managers obtain the same offset value from a random deleted partition manager (as `HashMap` is used). And all fetch requests for the new partition managers fail with `TopicOffsetOutOfRangeException`. Some of them are recovered via this logic if assigned offset is smaller than the real one, but other continue to repetitively fail with offset out of range exception preventing fetching messages from Kafka.
> {code:java}
> if (offset > _emittedToOffset) {
>  _lostMessageCount.incrBy(offset - _emittedToOffset);
>  _emittedToOffset = offset;
>  LOG.warn("{} Using new offset: {}", _partition, _emittedToOffset);
> }
> {code}
> I assume that state holder lookup should be based both on topic and partition number.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)