AtlasTopicCreator { public void createAtlasTopic(Configuration atlasProperties, String... topicNames) { if (atlasProperties.getBoolean(ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)) { if (!handleSecurity(atlasProperties)) { return; } ZkUtils zkUtils = createZkUtils(atlasProperties); for (String topicName : topicNames) { try { LOG.warn("Attempting to create topic {}", topicName); if (!ifTopicExists(topicName, zkUtils)) { createTopic(atlasProperties, topicName, zkUtils); } else { LOG.warn("Ignoring call to create topic {}, as it already exists.", topicName); } } catch (Throwable t) { LOG.error("Failed while creating topic {}", topicName, t); } } zkUtils.close(); } else { LOG.info("Not creating topics {} as {} is false", StringUtils.join(topicNames, ","), ATLAS_NOTIFICATION_CREATE_TOPICS_KEY); } }  void createAtlasTopic(Configuration atlasProperties, String... topicNames); static void main(String[] args); static final String ATLAS_NOTIFICATION_CREATE_TOPICS_KEY; }
@Test public void shouldNotCreateAtlasTopicIfNotConfiguredToDoSo() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(false); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final boolean[] topicExistsCalled = new boolean[] {false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { topicExistsCalled[0] = true; return false; } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK"); assertFalse(topicExistsCalled[0]); }
@Test public void shouldNotCreateTopicIfItAlreadyExists() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final boolean[] topicExistsCalled = new boolean[]{false}; final boolean[] createTopicCalled = new boolean[]{false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { topicExistsCalled[0] = true; return true; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createTopicCalled[0] = true; } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK"); assertTrue(topicExistsCalled[0]); assertFalse(createTopicCalled[0]); }
@Test public void shouldCreateTopicIfItDoesNotExist() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final boolean[] createdTopic = new boolean[]{false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createdTopic[0] = true; } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK"); assertTrue(createdTopic[0]); }
@Test public void shouldNotFailIfExceptionOccursDuringCreatingTopic() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final boolean[] createTopicCalled = new boolean[]{false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createTopicCalled[0] = true; throw new RuntimeException("Simulating failure during creating topic"); } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK"); assertTrue(createTopicCalled[0]); }
@Test public void shouldCreateMultipleTopics() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final Map<String, Boolean> createdTopics = new HashMap<>(); createdTopics.put("ATLAS_HOOK", false); createdTopics.put("ATLAS_ENTITIES", false); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createdTopics.put(topicName, true); } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK", "ATLAS_ENTITIES"); assertTrue(createdTopics.get("ATLAS_HOOK")); assertTrue(createdTopics.get("ATLAS_ENTITIES")); }
@Test public void shouldCreateTopicEvenIfEarlierOneFails() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final Map<String, Boolean> createdTopics = new HashMap<>(); createdTopics.put("ATLAS_ENTITIES", false); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { if (topicName.equals("ATLAS_HOOK")) { throw new RuntimeException("Simulating failure when creating ATLAS_HOOK topic"); } else { createdTopics.put(topicName, true); } } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK", "ATLAS_ENTITIES"); assertTrue(createdTopics.get("ATLAS_ENTITIES")); }
@Test public void shouldCloseResources() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK", "ATLAS_ENTITIES"); verify(zookeeperUtils, times(1)).close(); }
@Test public void shouldNotProcessTopicCreationIfSecurityFails() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final Map<String, Boolean> createdTopics = new HashMap<>(); createdTopics.put("ATLAS_HOOK", false); createdTopics.put("ATLAS_ENTITIES", false); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createdTopics.put(topicName, true); } @Override protected boolean handleSecurity(Configuration atlasProperties) { return false; } }; atlasTopicCreator.createAtlasTopic(configuration, "ATLAS_HOOK", "ATLAS_ENTITIES"); assertFalse(createdTopics.get("ATLAS_HOOK")); assertFalse(createdTopics.get("ATLAS_ENTITIES")); }
@Test public void shouldNotCreateAtlasTopicIfNotConfiguredToDoSo() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(false); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final boolean[] topicExistsCalled = new boolean[] {false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { topicExistsCalled[0] = true; return false; } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC); assertFalse(topicExistsCalled[0]); }
@Test public void shouldNotCreateTopicIfItAlreadyExists() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final boolean[] topicExistsCalled = new boolean[]{false}; final boolean[] createTopicCalled = new boolean[]{false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { topicExistsCalled[0] = true; return true; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createTopicCalled[0] = true; } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC); assertTrue(topicExistsCalled[0]); assertFalse(createTopicCalled[0]); }
@Test public void shouldCreateTopicIfItDoesNotExist() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final boolean[] createdTopic = new boolean[]{false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createdTopic[0] = true; } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC); assertTrue(createdTopic[0]); }
@Test public void shouldNotFailIfExceptionOccursDuringCreatingTopic() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final boolean[] createTopicCalled = new boolean[]{false}; AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createTopicCalled[0] = true; throw new RuntimeException("Simulating failure during creating topic"); } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC); assertTrue(createTopicCalled[0]); }
@Test public void shouldCreateMultipleTopics() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final Map<String, Boolean> createdTopics = new HashMap<>(); createdTopics.put(ATLAS_HOOK_TOPIC, false); createdTopics.put(ATLAS_ENTITIES_TOPIC, false); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createdTopics.put(topicName, true); } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC, ATLAS_ENTITIES_TOPIC); assertTrue(createdTopics.get(ATLAS_HOOK_TOPIC)); assertTrue(createdTopics.get(ATLAS_ENTITIES_TOPIC)); }
@Test public void shouldCreateTopicEvenIfEarlierOneFails() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final Map<String, Boolean> createdTopics = new HashMap<>(); createdTopics.put(ATLAS_ENTITIES_TOPIC, false); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { if (topicName.equals(ATLAS_HOOK_TOPIC)) { throw new RuntimeException("Simulating failure when creating ATLAS_HOOK topic"); } else { createdTopics.put(topicName, true); } } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC, ATLAS_ENTITIES_TOPIC); assertTrue(createdTopics.get(ATLAS_ENTITIES_TOPIC)); }
@Test public void shouldCloseResources() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); when(configuration.getString("atlas.authentication.method.kerberos")).thenReturn("false"); final ZkUtils zookeeperUtils = mock(ZkUtils.class); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC, ATLAS_ENTITIES_TOPIC); verify(zookeeperUtils, times(1)).close(); }
@Test public void shouldNotProcessTopicCreationIfSecurityFails() { Configuration configuration = mock(Configuration.class); when(configuration.getBoolean(AtlasTopicCreator.ATLAS_NOTIFICATION_CREATE_TOPICS_KEY, true)). thenReturn(true); final ZkUtils zookeeperUtils = mock(ZkUtils.class); final Map<String, Boolean> createdTopics = new HashMap<>(); createdTopics.put(ATLAS_HOOK_TOPIC, false); createdTopics.put(ATLAS_ENTITIES_TOPIC, false); AtlasTopicCreator atlasTopicCreator = new AtlasTopicCreator() { @Override protected boolean ifTopicExists(String topicName, ZkUtils zkUtils) { return false; } @Override protected ZkUtils createZkUtils(Configuration atlasProperties) { return zookeeperUtils; } @Override protected void createTopic(Configuration atlasProperties, String topicName, ZkUtils zkUtils) { createdTopics.put(topicName, true); } @Override protected boolean handleSecurity(Configuration atlasProperties) { return false; } }; atlasTopicCreator.createAtlasTopic(configuration, ATLAS_HOOK_TOPIC, ATLAS_ENTITIES_TOPIC); assertFalse(createdTopics.get(ATLAS_HOOK_TOPIC)); assertFalse(createdTopics.get(ATLAS_ENTITIES_TOPIC)); }