@Test public void testResetOwner() { String ownerPath = zkCoordinator.ownerPath(topic1, 1); zkCoordinator.takePartition(topic1, 1); zkCoordinator.resetOwner(topic1, 1); assertTrue(zkConnector.exists(ownerPath)); String owner = zkConnector.readString(ownerPath); assertEquals(consumerId, owner); }
public synchronized boolean resetOwner(String topic, int partition) { String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath)) { m_zk.delete(ownerPath); } if (m_zk.create(ownerPath, true)) { if (LOGGER.isInfoEnabled()) LOGGER.info( "Owner path " + ownerPath + " is rewritten."); m_zk.writeString(ownerPath, m_consumerId); return true; } else return false; } catch (Exception e) { LOGGER.error( "Fail to reset owner " + ownerPath, e); return false; } }
ZkCoordinator implements PartitionCoordinator { public synchronized boolean resetOwner(String topic, int partition) { String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath)) { m_zk.delete(ownerPath); } if (m_zk.create(ownerPath, true)) { if (LOGGER.isInfoEnabled()) LOGGER.info( "Owner path " + ownerPath + " is rewritten."); m_zk.writeString(ownerPath, m_consumerId); return true; } else return false; } catch (Exception e) { LOGGER.error( "Fail to reset owner " + ownerPath, e); return false; } } }
ZkCoordinator implements PartitionCoordinator { public synchronized boolean resetOwner(String topic, int partition) { String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath)) { m_zk.delete(ownerPath); } if (m_zk.create(ownerPath, true)) { if (LOGGER.isInfoEnabled()) LOGGER.info( "Owner path " + ownerPath + " is rewritten."); m_zk.writeString(ownerPath, m_consumerId); return true; } else return false; } catch (Exception e) { LOGGER.error( "Fail to reset owner " + ownerPath, e); return false; } } ZkCoordinator(List<String> topics, String groupId,
			String consumerId, KafkaController.ZkConnector zk); }
ZkCoordinator implements PartitionCoordinator { public synchronized boolean resetOwner(String topic, int partition) { String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath)) { m_zk.delete(ownerPath); } if (m_zk.create(ownerPath, true)) { if (LOGGER.isInfoEnabled()) LOGGER.info( "Owner path " + ownerPath + " is rewritten."); m_zk.writeString(ownerPath, m_consumerId); return true; } else return false; } catch (Exception e) { LOGGER.error( "Fail to reset owner " + ownerPath, e); return false; } } ZkCoordinator(List<String> topics, String groupId,
			String consumerId, KafkaController.ZkConnector zk); @Override synchronized void registerConsumer(); @Override synchronized void unregisterConsumer(); synchronized void reRegisterConsumer(boolean force); @Override synchronized boolean isNewPartition(String topic, int partition); @Override synchronized boolean takePartition(String topic, int partition); synchronized boolean resetOwner(String topic, int partition); @Override synchronized void releasePartition(String topic, int partition); @Override synchronized Set<Integer> getMyPartitions(String topic); synchronized boolean isZkOwnerExist(String topic, int partition); }
ZkCoordinator implements PartitionCoordinator { public synchronized boolean resetOwner(String topic, int partition) { String ownerPath = ownerPath(topic, partition); try { if (m_zk.exists(ownerPath)) { m_zk.delete(ownerPath); } if (m_zk.create(ownerPath, true)) { if (LOGGER.isInfoEnabled()) LOGGER.info( "Owner path " + ownerPath + " is rewritten."); m_zk.writeString(ownerPath, m_consumerId); return true; } else return false; } catch (Exception e) { LOGGER.error( "Fail to reset owner " + ownerPath, e); return false; } } ZkCoordinator(List<String> topics, String groupId,
			String consumerId, KafkaController.ZkConnector zk); @Override synchronized void registerConsumer(); @Override synchronized void unregisterConsumer(); synchronized void reRegisterConsumer(boolean force); @Override synchronized boolean isNewPartition(String topic, int partition); @Override synchronized boolean takePartition(String topic, int partition); synchronized boolean resetOwner(String topic, int partition); @Override synchronized void releasePartition(String topic, int partition); @Override synchronized Set<Integer> getMyPartitions(String topic); synchronized boolean isZkOwnerExist(String topic, int partition);  }