001/* 002 * (C) Copyright 2006-2016 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * 017 * Contributors: 018 * anechaev 019 */ 020package org.nuxeo.runtime.kafka; 021 022import java.util.HashMap; 023import java.util.Map; 024import java.util.Properties; 025import java.util.Set; 026 027import org.apache.commons.logging.Log; 028import org.apache.commons.logging.LogFactory; 029import org.apache.kafka.clients.consumer.ConsumerConfig; 030import org.apache.kafka.clients.producer.ProducerConfig; 031import org.nuxeo.lib.stream.log.kafka.KafkaUtils; 032import org.nuxeo.runtime.model.ComponentContext; 033import org.nuxeo.runtime.model.ComponentInstance; 034import org.nuxeo.runtime.model.DefaultComponent; 035 036public class KafkaConfigServiceImpl extends DefaultComponent implements KafkaConfigService { 037 public static final String KAFKA_CONFIG_XP = "kafkaConfig"; 038 039 public static final int APPLICATION_STARTED_ORDER = -600; 040 041 private static final Log log = LogFactory.getLog(KafkaConfigServiceImpl.class); 042 043 protected static final String DEFAULT_ZK_SERVERS = "DEFAULT_TEST"; 044 045 protected static final String DEFAULT_BOOTSTRAP_SERVERS = "DEFAULT_TEST"; 046 047 protected final Map<String, KafkaConfigDescriptor> configs = new HashMap<>(); 048 049 @Override 050 public void registerContribution(Object contribution, String extensionPoint, ComponentInstance contributor) { 051 if (extensionPoint.equals(KAFKA_CONFIG_XP)) { 052 KafkaConfigDescriptor descriptor = (KafkaConfigDescriptor) contribution; 053 configs.put(descriptor.name, descriptor); 054 log.info(String.format("Register Kafka contribution: %s", descriptor.name)); 055 } 056 } 057 058 @Override 059 public int getApplicationStartedOrder() { 060 // since there is no dependencies, let's start before main nuxeo core services 061 return APPLICATION_STARTED_ORDER; 062 } 063 064 @Override 065 public void deactivate(ComponentContext context) { 066 super.deactivate(context); 067 log.debug("Deactivating service"); 068 } 069 070 @Override 071 public void activate(ComponentContext context) { 072 super.activate(context); 073 log.debug("Activating service"); 074 } 075 076 @Override 077 public Set<String> listConfigNames() { 078 return configs.keySet(); 079 } 080 081 @Override 082 public String getZkServers(String configName) { 083 checkConfigName(configName); 084 String ret = configs.get(configName).zkServers; 085 if (DEFAULT_ZK_SERVERS.equals(ret)) { 086 return KafkaUtils.getZkServers(); 087 } 088 return ret; 089 } 090 091 protected void checkConfigName(String configName) { 092 if (!configs.containsKey(configName)) { 093 throw new IllegalArgumentException("Unknown configuration name: " + configName); 094 } 095 } 096 097 @Override 098 public Properties getProducerProperties(String configName) { 099 checkConfigName(configName); 100 Properties ret = configs.get(configName).getProducerProperties(); 101 if (DEFAULT_BOOTSTRAP_SERVERS.equals(ret.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { 102 ret.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaUtils.getBootstrapServers()); 103 } 104 return ret; 105 } 106 107 @Override 108 public Properties getConsumerProperties(String configName) { 109 checkConfigName(configName); 110 Properties ret = configs.get(configName).getConsumerProperties(); 111 if (DEFAULT_BOOTSTRAP_SERVERS.equals(ret.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { 112 ret.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaUtils.getBootstrapServers()); 113 } 114 return ret; 115 } 116 117 @Override 118 public String getTopicPrefix(String configName) { 119 checkConfigName(configName); 120 return configs.get(configName).getTopicPrefix(); 121 } 122}