如题所述
Kafkaæ¯ç±LinkedIn设计çä¸ä¸ªé«ååéãåå¸å¼ãåºäºåå¸è®¢é
模å¼çæ¶æ¯ç³»ç»ï¼ä½¿ç¨Scalaç¼åï¼å®ä»¥å¯æ°´å¹³æ©å±ãå¯é æ§ãå¼æ¥éä¿¡åé«ååççç¹æ§è被广æ³ä½¿ç¨ãç®åè¶æ¥è¶å¤çå¼æºåå¸å¼å¤çç³»ç»é½æ¯æä¸Kafkaéæï¼å
¶ä¸Spark Streamingä½ä¸ºå端æµå¼æé
åKafkaä½ä¸ºå端æ¶æ¯ç³»ç»æ£æ为å½åæµå¤çç³»ç»ç主æµæ¶æä¹ä¸ã
ç¶èï¼å½ä¸è¶æ¥è¶å¤çå®å ¨æ¼æ´ãæ°æ®æ³é²çé®é¢ççåï¼å®å ¨æ£æ为系ç»éåä¸å¾ä¸èèçé®é¢ï¼Kafkaç±äºå ¶å®å ¨æºå¶çå®ä¹ï¼ä¹å¯¼è´å ¶å¨æ°æ®ææè¡ä¸çé¨ç½²åå¨ä¸¥éçå®å ¨éæ£ãæ¬æå°å´ç»Kafkaï¼å ä»ç»å ¶æ´ä½æ¶æåå ³é®æ¦å¿µï¼åæ·±å ¥åæå ¶æ¶æä¹ä¸åå¨çå®å ¨é®é¢ï¼æåå享ä¸Transwarpå¨Kafkaå®å ¨æ§ä¸æåçå·¥ä½åå ¶ä½¿ç¨æ¹æ³ã
Kafkaæ¶æä¸å®å ¨
é¦å ï¼æ们æ¥äºè§£ä¸æå ³Kafkaçå 个åºæ¬æ¦å¿µï¼
Topicï¼Kafkaææ¥æ¶çæ¶æ¯æç§ç±»ååï¼æ¯ä¸ªç§ç±»é½ç§°ä¹ä¸ºTopicï¼ç±å¯ä¸çTopic Nameæ è¯ã
Producerï¼åTopicåå¸æ¶æ¯çè¿ç¨ç§°ä¸ºProducerã
Consumerï¼ä»Topic订é æ¶æ¯çè¿ç¨ç§°ä¸ºConsumerã
Brokerï¼Kafkaé群å å«ä¸ä¸ªæå¤ä¸ªæå¡å¨ï¼è¿ç§æå¡å¨è¢«ç§°ä¸ºBrokerã
Kafkaçæ´ä½æ¶æå¦ä¸å¾æ示ï¼å ¸åçKafkaé群å å«ä¸ç»åå¸æ¶æ¯çProducerï¼ä¸ç»ç®¡çTopicçBrokerï¼åä¸ç»è®¢é æ¶æ¯çConsumerãTopicå¯ä»¥æå¤ä¸ªååºï¼æ¯ä¸ªååºåªåå¨äºä¸ä¸ªBrokerãProducerå¯ä»¥æç §ä¸å®ççç¥å°æ¶æ¯ååç»æå®çååºï¼å¦ç®åç轮询å个ååºæè æç §ç¹å®å段çHashå¼æå®ååºãBrokeréè¦éè¿ZooKeeperè®°å½é群çææBrokerãé举ååºçLeaderï¼è®°å½Consumerçæ¶è´¹æ¶æ¯çå移éï¼ä»¥åå¨Consumer Groupåçååæ¶è¿è¡relalance. Brokeræ¥æ¶ååéæ¶æ¯æ¯è¢«å¨çï¼ç±Producer主å¨åéæ¶æ¯ï¼Consumer主å¨æåæ¶æ¯ã
ç¶èï¼åæKafkaæ¡æ¶ï¼æ们ä¼åç°ä»¥ä¸ä¸¥éçå®å ¨é®é¢ï¼
1.ç½ç»ä¸çä»»ä½ä¸å°ä¸»æºï¼é½å¯ä»¥éè¿å¯å¨Brokerè¿ç¨èå å ¥Kafkaé群ï¼è½å¤æ¥æ¶Producerçæ¶æ¯ï¼è½å¤ç¯¡æ¹æ¶æ¯å¹¶åéç»Consumerã
2.ç½ç»ä¸çä»»ä½ä¸å°ä¸»æºï¼é½å¯ä»¥å¯å¨æ¶æçProducer/Consumerè¿æ¥å°Brokerï¼åééæ³æ¶æ¯ææåéç§æ¶æ¯æ°æ®ã
3.Brokerä¸æ¯æè¿æ¥å°å¯ç¨Kerberos认è¯çZooKeeperé群ï¼æ²¡æ对åæ¾å¨ZooKeeperä¸çæ°æ®è®¾ç½®æéãä»»æç¨æ·é½è½å¤ç´æ¥è®¿é®ZooKeeperé群ï¼å¯¹è¿äºæ°æ®è¿è¡ä¿®æ¹æå é¤ã
4.Kafkaä¸çTopicä¸æ¯æ设置访é®æ§å¶å表ï¼ä»»æè¿æ¥å°Kafkaé群çConsumerï¼æProducerï¼é½è½å¯¹ä»»æTopic读åï¼æåéï¼æ¶æ¯ã
éçKafkaåºç¨åºæ¯è¶æ¥è¶å¹¿æ³ï¼ç¹å«æ¯ä¸äºæ°æ®éç§ç¨åº¦è¾é«çé¢åï¼å¦é路交éçè§é¢çæ§ï¼ï¼ä¸è¿°å®å ¨é®é¢çåå¨ç¹å¦ä¸é¢å®æ¶ç¸å¼¹ï¼ä¸æ¦å ç½è¢«é»å®¢å ¥ä¾µæè å é¨åºç°æ¶æç¨æ·ï¼ææçéç§æ°æ®ï¼å¦è½¦è¾åºè¡è®°å½ï¼é½è½å¤è½»æå°è¢«çªåï¼èæ éæ»ç ´Brokeræå¨çæå¡å¨ã
Kafkaå®å ¨è®¾è®¡
åºäºä¸è¿°åæï¼Transwarpä»ä»¥ä¸ä¸¤ä¸ªæ¹é¢å¢å¼ºKafkaçå®å ¨æ§ï¼
身份认è¯ï¼Authenticationï¼ï¼è®¾è®¡å¹¶å®ç°äºåºäºKerberosååºäºIPç两ç§èº«ä»½è®¤è¯æºå¶ãåè 为强身份认è¯ï¼ç¸æ¯äºåè å ·ææ´å¥½çå®å ¨æ§ï¼åè éç¨äºIPå°åå¯ä¿¡çç½ç»ç¯å¢ï¼ç¸æ¯äºåè é¨ç½²æ´ä¸ºç®ä¾¿ã
æéæ§å¶ï¼Authorizationï¼ï¼è®¾è®¡å¹¶å®ç°äºTopic级å«çæé模åãTopicçæéå为READï¼ä»Topicæåæ°æ®ï¼ãWRITEï¼åTopicä¸ç产æ°æ®ï¼ãCREATEï¼å建Topicï¼åDELETEï¼å é¤Topicï¼ã
åºäºKerberosç身份æºå¶å¦ä¸å¾æ示ï¼
Brokerå¯å¨æ¶ï¼éè¦ä½¿ç¨é ç½®æ件ä¸ç身份åå¯é¥æ件åKDCï¼Kerberosæå¡å¨ï¼è®¤è¯ï¼è®¤è¯éè¿åå å ¥Kafkaé群ï¼å¦åæ¥ééåºã
Producerï¼æConsumerï¼å¯å¨åéè¦ç»è¿å¦ä¸æ¥éª¤ä¸Broker建ç«å®å ¨çSocketè¿æ¥ï¼
1.ProduceråKDC认è¯èº«ä»½ï¼éè¿åå¾å°TGTï¼ç¥¨è¯è¯·æ±ç¥¨è¯ï¼ï¼å¦åæ¥ééåº
2.Producer使ç¨TGTåKDC请æ±Kafkaæå¡ï¼KDCéªè¯TGT并åProducerè¿åSessionKeyï¼ä¼è¯å¯é¥ï¼åServiceTicketï¼æå¡ç¥¨è¯ï¼
3.Producer使ç¨SessionKeyåServiceTicketä¸Broker建ç«è¿æ¥ï¼Broker使ç¨èªèº«çå¯é¥è§£å¯ServiceTicketï¼è·å¾ä¸Produceréä¿¡çSessionKeyï¼ç¶å使ç¨SessionKeyéªè¯Producerç身份ï¼éè¿å建ç«è¿æ¥ï¼å¦åæç»è¿æ¥ã
ZooKeeperéè¦å¯ç¨Kerberos认è¯æ¨¡å¼ï¼ä¿è¯BrokeræConsumerä¸å ¶çè¿æ¥æ¯å®å ¨çã
Topicç访é®æ§å¶å表ï¼ACLï¼åå¨äºZooKeeperä¸ï¼åå¨èç¹çè·¯å¾ä¸º/acl/<topic>/<user>ï¼èç¹æ°æ®ä¸ºR(ead)ãW(rite)ãC(reate)ãD(elete)æéçéåï¼å¦/acl/transaction/jackèç¹çæ°æ®ä¸ºRWï¼å表示ç¨æ·jackè½å¤å¯¹transactionè¿ä¸ªtopicè¿è¡è¯»ååã
å¦å¤ï¼kafka为ç¹æç¨æ·ï¼åªækafkaç¨æ·è½å¤èµäº/åæ¶æéãå æ¤ï¼ACLç¸å ³çZooKeeperèç¹æé为kafkaå ·ææææéï¼å ¶ä»ç¨æ·ä¸å ·æä»»ä½æéã
æ建å®å ¨çKafkaæå¡
é¦å ï¼æ们为Brokerå¯ç¨Kerberos认è¯æ¨¡å¼ï¼é ç½®æ件为/etc/kafka/conf/server.propertiesï¼å®å ¨ç¸å ³çåæ°å¦ä¸æ示ï¼
å ¶ä¸ï¼authenticationåæ°è¡¨ç¤ºè®¤è¯æ¨¡å¼ï¼å¯éé 置项为simple, kerberosåipaddressï¼é»è®¤ä¸ºsimpleãå½è®¤è¯æ¨¡å¼ä¸ºkerberosæ¶ï¼éè¦é¢å¤é 置账æ·å±æ§principalå对åºçå¯é¥æ件路å¾keytab.
认è¯æ¨¡å¼ä¸ºipaddressæ¶ï¼ProduceråConsumerå建æ¶ä¸éè¦åä»»ä½æ¹åãè认è¯æ¨¡å¼ä¸ºkerberosæ¶ï¼éè¦é¢å å建好ç¸åºçprincipalåkeytabï¼å¹¶ä½¿ç¨APIè¿è¡ç»å½ï¼æ ·ä¾ä»£ç å¦ä¸æ示ï¼
public class SecureProducer extends Thread {
private final kafka.javaapi.producer.Producer<Integer, String> producer;
private final String topic;
private final Properties props = new Properties();
public SecureProducer(String topic) {
AuthenticationManager.setAuthMethod(âkerberosâ);
AuthenticationManager.login(âproducer1â³, â/etc/producer1.keytabâ);
props.put(âserializer.classâ, âkafka.serializer.StringEncoderâ);
props.put(âmetadata.broker.listâ,
â172.16.1.190:9092,172.16.1.192:9092,172.16.1.193:9092â³);
// Use random partitioner. Donât need the key type. Just set it to Integer.
// The message is of type String.
producer = new kafka.javaapi.producer.Producer<Integer, String>(
new ProducerConfig(props));
this.topic = topic;
}
. . .
Topicæé管ç
Topicçæé管ç主è¦æ¯éè¿AuthorizationManagerè¿ä¸ªç±»æ¥å®æçï¼å ¶ç±»ç»æå¦ä¸å¾æ示ï¼
å ¶ä¸ï¼resetPermission(user, Permissions, topic) 为éç½®user对topicçæéã
grant(user, Permissions, topic) 为èµäºuser对topicæéã
revoke(user, Permissions, topic) 为åæ¶user对topicæéã
isPermitted(user, Permissions, topic) 为æ£æ¥user对topicæ¯å¦å ·ææå®æéã
è°ç¨grantærevokeè¿è¡æé设置å®æåï¼éè¦commitå½ä»¤æ交修æ¹å°ZooKeeper
Kerberos模å¼ä¸ï¼AuthorizationManageréè¦å 使ç¨AuthenticationManager.loginæ¹æ³ç»å½ï¼ä¸ZooKeeper建ç«å®å ¨çè¿æ¥ï¼åè¿è¡æé设置ã示ä¾ä»£ç å¦ä¸æ示ï¼
public class AuthzTest {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(âauthenticationâ, âkerberosâ);
props.setProperty(âzookeeper.connectâ, â172.16.2.116:2181,172.16.2.117:2181,172.16.2.118:2181â³);
props.setProperty(âprincipalâ, âkafka/host1@TDHâ);
props.setProperty(âkeytabâ, â/usr/lib/kafka/config/kafka.keytabâ);
ZKConfig config = new ZKConfig(props);
AuthenticationManager.setAuthMethod(config.authentication());
AuthenticationManager.login(config.principal(), config.keytab());
AuthorizationManager authzManager = new AuthorizationManager(config);
// reset permission READ and WRITE to ip 172.16.1.87 on topic test
authzManager.resetPermission(â172.16.1.87â³,
new Permissions(Permissions.READ, Permissions.WRITE), âtestâ);
// grant permission WRITE to ip 172.16.1.87 on topic test
authzManager.grant(â172.16.1.87â³, new Permissions(Permissions.CREATE), âtestâ);
// revoke permission READ from ip 172.16.1.87 on topic test
authzManager.revoke(â172.16.1.87â³, new Permissions(Permissions.READ), âtestâ);
// commit the permission settings
authzManager.commit();
authzManager.close();
}
}
ipaddress认è¯æ¨¡å¼ä¸ï¼åæ¶åèµäºæéçæä½å¦ä¸æ示ï¼
public class AuthzTest {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(âauthenticationâ, âipaddressâ);
props.setProperty(âzookeeper.connectâ,
â172.16.1.87:2181,172.16.1.88:2181,172.16.1.89:2181â³);
ZKConfig config = new ZKConfig(props);
// new authorization manager
AuthorizationManager authzManager = new AuthorizationManager(config);
// reset permission READ and WRITE to ip 172.16.1.87 on topic test
authzManager.resetPermission(â172.16.1.87â³,
new Permissions(Permissions.READ, Permissions.WRITE), âtestâ);
// grant permission WRITE to ip 172.16.1.87 on topic test
authzManager.grant(â172.16.1.87â³, new Permissions(Permissions.CREATE), âtestâ);
// revoke permission READ from ip 172.16.1.87 on topic test
authzManager.revoke(â172.16.1.87â³, new Permissions(Permissions.READ), âtestâ);
// commit the permission settings
authzManager.commit();
authzManager.close();
}
}
æ»ç»ä¸å±æ
æ¬æéè¿ä»ç»Kafkaç°ææ¶æï¼æ·±å ¥ææå ¶ä¸åå¨çå®å ¨é®é¢ï¼å¹¶ç»åºTranswarpå¨Kafkaå®å ¨ä¸æåçå·¥ä½åå ¶ä½¿ç¨æ¹å¼ãç¶èï¼çºµè§Hadoop & Sparkçæç³»ç»ï¼å®å ¨åè½è¿åå¨å¾å¤é®é¢ï¼åç»ä»¶çæéç³»ç»ç¬ç«æ··ä¹±ï¼ç¼ºå°éä¸æç¨çè´¦æ·ç®¡çç³»ç»ãæäºç»ä»¶çæé管çè¿å¾ä¸æçï¼å¦Sparkçè°åº¦å¨ç¼ºå°ç¨æ·çæ¦å¿µï¼ä¸è½éå¶å ·ä½ç¨æ·ä½¿ç¨èµæºçå¤å°ãTranswarpåºäºå¼æºçæ¬ï¼å¨å®å ¨æ¹é¢å·²æç¸å½å¤ç积累ï¼å¹¶æç»æ¹è¿å¼åï¼è´åäºä¸ºä¼ä¸ç¨æ·æä¾ä¸ä¸ªæç¨ãé«æãå®å ¨å稳å®çåºç¡æ°æ®å¹³å°ã
ç¶èï¼å½ä¸è¶æ¥è¶å¤çå®å ¨æ¼æ´ãæ°æ®æ³é²çé®é¢ççåï¼å®å ¨æ£æ为系ç»éåä¸å¾ä¸èèçé®é¢ï¼Kafkaç±äºå ¶å®å ¨æºå¶çå®ä¹ï¼ä¹å¯¼è´å ¶å¨æ°æ®ææè¡ä¸çé¨ç½²åå¨ä¸¥éçå®å ¨éæ£ãæ¬æå°å´ç»Kafkaï¼å ä»ç»å ¶æ´ä½æ¶æåå ³é®æ¦å¿µï¼åæ·±å ¥åæå ¶æ¶æä¹ä¸åå¨çå®å ¨é®é¢ï¼æåå享ä¸Transwarpå¨Kafkaå®å ¨æ§ä¸æåçå·¥ä½åå ¶ä½¿ç¨æ¹æ³ã
Kafkaæ¶æä¸å®å ¨
é¦å ï¼æ们æ¥äºè§£ä¸æå ³Kafkaçå 个åºæ¬æ¦å¿µï¼
Topicï¼Kafkaææ¥æ¶çæ¶æ¯æç§ç±»ååï¼æ¯ä¸ªç§ç±»é½ç§°ä¹ä¸ºTopicï¼ç±å¯ä¸çTopic Nameæ è¯ã
Producerï¼åTopicåå¸æ¶æ¯çè¿ç¨ç§°ä¸ºProducerã
Consumerï¼ä»Topic订é æ¶æ¯çè¿ç¨ç§°ä¸ºConsumerã
Brokerï¼Kafkaé群å å«ä¸ä¸ªæå¤ä¸ªæå¡å¨ï¼è¿ç§æå¡å¨è¢«ç§°ä¸ºBrokerã
Kafkaçæ´ä½æ¶æå¦ä¸å¾æ示ï¼å ¸åçKafkaé群å å«ä¸ç»åå¸æ¶æ¯çProducerï¼ä¸ç»ç®¡çTopicçBrokerï¼åä¸ç»è®¢é æ¶æ¯çConsumerãTopicå¯ä»¥æå¤ä¸ªååºï¼æ¯ä¸ªååºåªåå¨äºä¸ä¸ªBrokerãProducerå¯ä»¥æç §ä¸å®ççç¥å°æ¶æ¯ååç»æå®çååºï¼å¦ç®åç轮询å个ååºæè æç §ç¹å®å段çHashå¼æå®ååºãBrokeréè¦éè¿ZooKeeperè®°å½é群çææBrokerãé举ååºçLeaderï¼è®°å½Consumerçæ¶è´¹æ¶æ¯çå移éï¼ä»¥åå¨Consumer Groupåçååæ¶è¿è¡relalance. Brokeræ¥æ¶ååéæ¶æ¯æ¯è¢«å¨çï¼ç±Producer主å¨åéæ¶æ¯ï¼Consumer主å¨æåæ¶æ¯ã
ç¶èï¼åæKafkaæ¡æ¶ï¼æ们ä¼åç°ä»¥ä¸ä¸¥éçå®å ¨é®é¢ï¼
1.ç½ç»ä¸çä»»ä½ä¸å°ä¸»æºï¼é½å¯ä»¥éè¿å¯å¨Brokerè¿ç¨èå å ¥Kafkaé群ï¼è½å¤æ¥æ¶Producerçæ¶æ¯ï¼è½å¤ç¯¡æ¹æ¶æ¯å¹¶åéç»Consumerã
2.ç½ç»ä¸çä»»ä½ä¸å°ä¸»æºï¼é½å¯ä»¥å¯å¨æ¶æçProducer/Consumerè¿æ¥å°Brokerï¼åééæ³æ¶æ¯ææåéç§æ¶æ¯æ°æ®ã
3.Brokerä¸æ¯æè¿æ¥å°å¯ç¨Kerberos认è¯çZooKeeperé群ï¼æ²¡æ对åæ¾å¨ZooKeeperä¸çæ°æ®è®¾ç½®æéãä»»æç¨æ·é½è½å¤ç´æ¥è®¿é®ZooKeeperé群ï¼å¯¹è¿äºæ°æ®è¿è¡ä¿®æ¹æå é¤ã
4.Kafkaä¸çTopicä¸æ¯æ设置访é®æ§å¶å表ï¼ä»»æè¿æ¥å°Kafkaé群çConsumerï¼æProducerï¼é½è½å¯¹ä»»æTopic读åï¼æåéï¼æ¶æ¯ã
éçKafkaåºç¨åºæ¯è¶æ¥è¶å¹¿æ³ï¼ç¹å«æ¯ä¸äºæ°æ®éç§ç¨åº¦è¾é«çé¢åï¼å¦é路交éçè§é¢çæ§ï¼ï¼ä¸è¿°å®å ¨é®é¢çåå¨ç¹å¦ä¸é¢å®æ¶ç¸å¼¹ï¼ä¸æ¦å ç½è¢«é»å®¢å ¥ä¾µæè å é¨åºç°æ¶æç¨æ·ï¼ææçéç§æ°æ®ï¼å¦è½¦è¾åºè¡è®°å½ï¼é½è½å¤è½»æå°è¢«çªåï¼èæ éæ»ç ´Brokeræå¨çæå¡å¨ã
Kafkaå®å ¨è®¾è®¡
åºäºä¸è¿°åæï¼Transwarpä»ä»¥ä¸ä¸¤ä¸ªæ¹é¢å¢å¼ºKafkaçå®å ¨æ§ï¼
身份认è¯ï¼Authenticationï¼ï¼è®¾è®¡å¹¶å®ç°äºåºäºKerberosååºäºIPç两ç§èº«ä»½è®¤è¯æºå¶ãåè 为强身份认è¯ï¼ç¸æ¯äºåè å ·ææ´å¥½çå®å ¨æ§ï¼åè éç¨äºIPå°åå¯ä¿¡çç½ç»ç¯å¢ï¼ç¸æ¯äºåè é¨ç½²æ´ä¸ºç®ä¾¿ã
æéæ§å¶ï¼Authorizationï¼ï¼è®¾è®¡å¹¶å®ç°äºTopic级å«çæé模åãTopicçæéå为READï¼ä»Topicæåæ°æ®ï¼ãWRITEï¼åTopicä¸ç产æ°æ®ï¼ãCREATEï¼å建Topicï¼åDELETEï¼å é¤Topicï¼ã
åºäºKerberosç身份æºå¶å¦ä¸å¾æ示ï¼
Brokerå¯å¨æ¶ï¼éè¦ä½¿ç¨é ç½®æ件ä¸ç身份åå¯é¥æ件åKDCï¼Kerberosæå¡å¨ï¼è®¤è¯ï¼è®¤è¯éè¿åå å ¥Kafkaé群ï¼å¦åæ¥ééåºã
Producerï¼æConsumerï¼å¯å¨åéè¦ç»è¿å¦ä¸æ¥éª¤ä¸Broker建ç«å®å ¨çSocketè¿æ¥ï¼
1.ProduceråKDC认è¯èº«ä»½ï¼éè¿åå¾å°TGTï¼ç¥¨è¯è¯·æ±ç¥¨è¯ï¼ï¼å¦åæ¥ééåº
2.Producer使ç¨TGTåKDC请æ±Kafkaæå¡ï¼KDCéªè¯TGT并åProducerè¿åSessionKeyï¼ä¼è¯å¯é¥ï¼åServiceTicketï¼æå¡ç¥¨è¯ï¼
3.Producer使ç¨SessionKeyåServiceTicketä¸Broker建ç«è¿æ¥ï¼Broker使ç¨èªèº«çå¯é¥è§£å¯ServiceTicketï¼è·å¾ä¸Produceréä¿¡çSessionKeyï¼ç¶å使ç¨SessionKeyéªè¯Producerç身份ï¼éè¿å建ç«è¿æ¥ï¼å¦åæç»è¿æ¥ã
ZooKeeperéè¦å¯ç¨Kerberos认è¯æ¨¡å¼ï¼ä¿è¯BrokeræConsumerä¸å ¶çè¿æ¥æ¯å®å ¨çã
Topicç访é®æ§å¶å表ï¼ACLï¼åå¨äºZooKeeperä¸ï¼åå¨èç¹çè·¯å¾ä¸º/acl/<topic>/<user>ï¼èç¹æ°æ®ä¸ºR(ead)ãW(rite)ãC(reate)ãD(elete)æéçéåï¼å¦/acl/transaction/jackèç¹çæ°æ®ä¸ºRWï¼å表示ç¨æ·jackè½å¤å¯¹transactionè¿ä¸ªtopicè¿è¡è¯»ååã
å¦å¤ï¼kafka为ç¹æç¨æ·ï¼åªækafkaç¨æ·è½å¤èµäº/åæ¶æéãå æ¤ï¼ACLç¸å ³çZooKeeperèç¹æé为kafkaå ·ææææéï¼å ¶ä»ç¨æ·ä¸å ·æä»»ä½æéã
æ建å®å ¨çKafkaæå¡
é¦å ï¼æ们为Brokerå¯ç¨Kerberos认è¯æ¨¡å¼ï¼é ç½®æ件为/etc/kafka/conf/server.propertiesï¼å®å ¨ç¸å ³çåæ°å¦ä¸æ示ï¼
å ¶ä¸ï¼authenticationåæ°è¡¨ç¤ºè®¤è¯æ¨¡å¼ï¼å¯éé 置项为simple, kerberosåipaddressï¼é»è®¤ä¸ºsimpleãå½è®¤è¯æ¨¡å¼ä¸ºkerberosæ¶ï¼éè¦é¢å¤é 置账æ·å±æ§principalå对åºçå¯é¥æ件路å¾keytab.
认è¯æ¨¡å¼ä¸ºipaddressæ¶ï¼ProduceråConsumerå建æ¶ä¸éè¦åä»»ä½æ¹åãè认è¯æ¨¡å¼ä¸ºkerberosæ¶ï¼éè¦é¢å å建好ç¸åºçprincipalåkeytabï¼å¹¶ä½¿ç¨APIè¿è¡ç»å½ï¼æ ·ä¾ä»£ç å¦ä¸æ示ï¼
public class SecureProducer extends Thread {
private final kafka.javaapi.producer.Producer<Integer, String> producer;
private final String topic;
private final Properties props = new Properties();
public SecureProducer(String topic) {
AuthenticationManager.setAuthMethod(âkerberosâ);
AuthenticationManager.login(âproducer1â³, â/etc/producer1.keytabâ);
props.put(âserializer.classâ, âkafka.serializer.StringEncoderâ);
props.put(âmetadata.broker.listâ,
â172.16.1.190:9092,172.16.1.192:9092,172.16.1.193:9092â³);
// Use random partitioner. Donât need the key type. Just set it to Integer.
// The message is of type String.
producer = new kafka.javaapi.producer.Producer<Integer, String>(
new ProducerConfig(props));
this.topic = topic;
}
. . .
Topicæé管ç
Topicçæé管ç主è¦æ¯éè¿AuthorizationManagerè¿ä¸ªç±»æ¥å®æçï¼å ¶ç±»ç»æå¦ä¸å¾æ示ï¼
å ¶ä¸ï¼resetPermission(user, Permissions, topic) 为éç½®user对topicçæéã
grant(user, Permissions, topic) 为èµäºuser对topicæéã
revoke(user, Permissions, topic) 为åæ¶user对topicæéã
isPermitted(user, Permissions, topic) 为æ£æ¥user对topicæ¯å¦å ·ææå®æéã
è°ç¨grantærevokeè¿è¡æé设置å®æåï¼éè¦commitå½ä»¤æ交修æ¹å°ZooKeeper
Kerberos模å¼ä¸ï¼AuthorizationManageréè¦å 使ç¨AuthenticationManager.loginæ¹æ³ç»å½ï¼ä¸ZooKeeper建ç«å®å ¨çè¿æ¥ï¼åè¿è¡æé设置ã示ä¾ä»£ç å¦ä¸æ示ï¼
public class AuthzTest {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(âauthenticationâ, âkerberosâ);
props.setProperty(âzookeeper.connectâ, â172.16.2.116:2181,172.16.2.117:2181,172.16.2.118:2181â³);
props.setProperty(âprincipalâ, âkafka/host1@TDHâ);
props.setProperty(âkeytabâ, â/usr/lib/kafka/config/kafka.keytabâ);
ZKConfig config = new ZKConfig(props);
AuthenticationManager.setAuthMethod(config.authentication());
AuthenticationManager.login(config.principal(), config.keytab());
AuthorizationManager authzManager = new AuthorizationManager(config);
// reset permission READ and WRITE to ip 172.16.1.87 on topic test
authzManager.resetPermission(â172.16.1.87â³,
new Permissions(Permissions.READ, Permissions.WRITE), âtestâ);
// grant permission WRITE to ip 172.16.1.87 on topic test
authzManager.grant(â172.16.1.87â³, new Permissions(Permissions.CREATE), âtestâ);
// revoke permission READ from ip 172.16.1.87 on topic test
authzManager.revoke(â172.16.1.87â³, new Permissions(Permissions.READ), âtestâ);
// commit the permission settings
authzManager.commit();
authzManager.close();
}
}
ipaddress认è¯æ¨¡å¼ä¸ï¼åæ¶åèµäºæéçæä½å¦ä¸æ示ï¼
public class AuthzTest {
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(âauthenticationâ, âipaddressâ);
props.setProperty(âzookeeper.connectâ,
â172.16.1.87:2181,172.16.1.88:2181,172.16.1.89:2181â³);
ZKConfig config = new ZKConfig(props);
// new authorization manager
AuthorizationManager authzManager = new AuthorizationManager(config);
// reset permission READ and WRITE to ip 172.16.1.87 on topic test
authzManager.resetPermission(â172.16.1.87â³,
new Permissions(Permissions.READ, Permissions.WRITE), âtestâ);
// grant permission WRITE to ip 172.16.1.87 on topic test
authzManager.grant(â172.16.1.87â³, new Permissions(Permissions.CREATE), âtestâ);
// revoke permission READ from ip 172.16.1.87 on topic test
authzManager.revoke(â172.16.1.87â³, new Permissions(Permissions.READ), âtestâ);
// commit the permission settings
authzManager.commit();
authzManager.close();
}
}
æ»ç»ä¸å±æ
æ¬æéè¿ä»ç»Kafkaç°ææ¶æï¼æ·±å ¥ææå ¶ä¸åå¨çå®å ¨é®é¢ï¼å¹¶ç»åºTranswarpå¨Kafkaå®å ¨ä¸æåçå·¥ä½åå ¶ä½¿ç¨æ¹å¼ãç¶èï¼çºµè§Hadoop & Sparkçæç³»ç»ï¼å®å ¨åè½è¿åå¨å¾å¤é®é¢ï¼åç»ä»¶çæéç³»ç»ç¬ç«æ··ä¹±ï¼ç¼ºå°éä¸æç¨çè´¦æ·ç®¡çç³»ç»ãæäºç»ä»¶çæé管çè¿å¾ä¸æçï¼å¦Sparkçè°åº¦å¨ç¼ºå°ç¨æ·çæ¦å¿µï¼ä¸è½éå¶å ·ä½ç¨æ·ä½¿ç¨èµæºçå¤å°ãTranswarpåºäºå¼æºçæ¬ï¼å¨å®å ¨æ¹é¢å·²æç¸å½å¤ç积累ï¼å¹¶æç»æ¹è¿å¼åï¼è´åäºä¸ºä¼ä¸ç¨æ·æä¾ä¸ä¸ªæç¨ãé«æãå®å ¨å稳å®çåºç¡æ°æ®å¹³å°ã
温馨提示:答案为网友推荐,仅供参考