@Test public void testFindRetryPolicyByTopicAndGroup() throws Exception { RetryPolicy policy = m_metaService.findRetryPolicyByTopicAndGroup("test_broker", "group1"); assertTrue(policy instanceof FrequencySpecifiedRetryPolicy); assertEquals(4, ((FrequencySpecifiedRetryPolicy) policy).getRetryTimes()); policy = m_metaService.findRetryPolicyByTopicAndGroup("test_broker", "group2"); assertTrue(policy instanceof FrequencySpecifiedRetryPolicy); assertEquals(2, ((FrequencySpecifiedRetryPolicy) policy).getRetryTimes()); policy = m_metaService.findRetryPolicyByTopicAndGroup("test_broker", "group3"); assertTrue(policy instanceof FrequencySpecifiedRetryPolicy); assertEquals(3, ((FrequencySpecifiedRetryPolicy) policy).getRetryTimes()); try { m_metaService.findRetryPolicyByTopicAndGroup("topic_not_found", "group3"); fail(); } catch (RuntimeException e) { } catch (Exception e) { fail(); } try { m_metaService.findRetryPolicyByTopicAndGroup("test_broker", "group_not_found"); fail(); } catch (RuntimeException e) { } catch (Exception e) { fail(); } }
@Override public RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId) { Topic topic = findTopic(topicName, getMeta()); if (containsConsumerGroup(topicName, groupId)) { ConsumerGroup consumerGroup = topic.findConsumerGroup(groupId); String retryPolicyValue = consumerGroup.getRetryPolicy(); if (StringUtils.isBlank(retryPolicyValue)) { retryPolicyValue = topic.getConsumerRetryPolicy(); } return RetryPolicyFactory.create(retryPolicyValue); } else { throw new RuntimeException(String.format("Consumer group %s for topic %s not found", groupId, topicName)); } }
DefaultMetaService implements MetaService, Initializable { @Override public RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId) { Topic topic = findTopic(topicName, getMeta()); if (containsConsumerGroup(topicName, groupId)) { ConsumerGroup consumerGroup = topic.findConsumerGroup(groupId); String retryPolicyValue = consumerGroup.getRetryPolicy(); if (StringUtils.isBlank(retryPolicyValue)) { retryPolicyValue = topic.getConsumerRetryPolicy(); } return RetryPolicyFactory.create(retryPolicyValue); } else { throw new RuntimeException(String.format("Consumer group %s for topic %s not found", groupId, topicName)); } } }
DefaultMetaService implements MetaService, Initializable { @Override public RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId) { Topic topic = findTopic(topicName, getMeta()); if (containsConsumerGroup(topicName, groupId)) { ConsumerGroup consumerGroup = topic.findConsumerGroup(groupId); String retryPolicyValue = consumerGroup.getRetryPolicy(); if (StringUtils.isBlank(retryPolicyValue)) { retryPolicyValue = topic.getConsumerRetryPolicy(); } return RetryPolicyFactory.create(retryPolicyValue); } else { throw new RuntimeException(String.format("Consumer group %s for topic %s not found", groupId, topicName)); } }  }
DefaultMetaService implements MetaService, Initializable { @Override public RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId) { Topic topic = findTopic(topicName, getMeta()); if (containsConsumerGroup(topicName, groupId)) { ConsumerGroup consumerGroup = topic.findConsumerGroup(groupId); String retryPolicyValue = consumerGroup.getRetryPolicy(); if (StringUtils.isBlank(retryPolicyValue)) { retryPolicyValue = topic.getConsumerRetryPolicy(); } return RetryPolicyFactory.create(retryPolicyValue); } else { throw new RuntimeException(String.format("Consumer group %s for topic %s not found", groupId, topicName)); } }  @Override List<Partition> listPartitionsByTopic(String topicName); @Override Storage findStorageByTopic(String topicName); @Override Partition findPartitionByTopicAndPartition(String topicName, int partitionId); @Override List<Topic> listTopicsByPattern(String topicPattern); @Override Topic findTopicByName(String topicName); @Override int translateToIntGroupId(String topicName, String groupName); @Override List<Datasource> listAllMysqlDataSources(); @Override int getAckTimeoutSecondsByTopicAndConsumerGroup(String topicName, String groupId); @Override LeaseAcquireResponse tryAcquireConsumerLease(Tpg tpg, String sessionId); @Override LeaseAcquireResponse tryRenewConsumerLease(Tpg tpg, Lease lease, String sessionId); @Override LeaseAcquireResponse tryRenewBrokerLease(String topic, int partition, Lease lease, String sessionId,
	      int brokerPort); @Override LeaseAcquireResponse tryAcquireBrokerLease(String topic, int partition, String sessionId, int brokerPort); @Override void initialize(); @Override synchronized Pair<Endpoint, Long> findEndpointByTopicAndPartition(String topic, int partition); @Override RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId); @Override boolean containsEndpoint(Endpoint endpoint); @Override boolean containsConsumerGroup(String topicName, String groupId); @Override Offset findMessageOffsetByTime(String topic, int partition, long time); @Override Map<Integer, Offset> findMessageOffsetByTime(String topic, long time); @Override List<ZookeeperEnsemble> listAllZookeeperEnsemble(); @Override Idc getPrimaryIdc(); }
DefaultMetaService implements MetaService, Initializable { @Override public RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId) { Topic topic = findTopic(topicName, getMeta()); if (containsConsumerGroup(topicName, groupId)) { ConsumerGroup consumerGroup = topic.findConsumerGroup(groupId); String retryPolicyValue = consumerGroup.getRetryPolicy(); if (StringUtils.isBlank(retryPolicyValue)) { retryPolicyValue = topic.getConsumerRetryPolicy(); } return RetryPolicyFactory.create(retryPolicyValue); } else { throw new RuntimeException(String.format("Consumer group %s for topic %s not found", groupId, topicName)); } }  @Override List<Partition> listPartitionsByTopic(String topicName); @Override Storage findStorageByTopic(String topicName); @Override Partition findPartitionByTopicAndPartition(String topicName, int partitionId); @Override List<Topic> listTopicsByPattern(String topicPattern); @Override Topic findTopicByName(String topicName); @Override int translateToIntGroupId(String topicName, String groupName); @Override List<Datasource> listAllMysqlDataSources(); @Override int getAckTimeoutSecondsByTopicAndConsumerGroup(String topicName, String groupId); @Override LeaseAcquireResponse tryAcquireConsumerLease(Tpg tpg, String sessionId); @Override LeaseAcquireResponse tryRenewConsumerLease(Tpg tpg, Lease lease, String sessionId); @Override LeaseAcquireResponse tryRenewBrokerLease(String topic, int partition, Lease lease, String sessionId,
	      int brokerPort); @Override LeaseAcquireResponse tryAcquireBrokerLease(String topic, int partition, String sessionId, int brokerPort); @Override void initialize(); @Override synchronized Pair<Endpoint, Long> findEndpointByTopicAndPartition(String topic, int partition); @Override RetryPolicy findRetryPolicyByTopicAndGroup(String topicName, String groupId); @Override boolean containsEndpoint(Endpoint endpoint); @Override boolean containsConsumerGroup(String topicName, String groupId); @Override Offset findMessageOffsetByTime(String topic, int partition, long time); @Override Map<Integer, Offset> findMessageOffsetByTime(String topic, long time); @Override List<ZookeeperEnsemble> listAllZookeeperEnsemble(); @Override Idc getPrimaryIdc();  }