SENTRY-2276 Sentry-Kafka integration does not support Kafka's Alter/DescribeConfigs and IdempotentWrite operations (Gergo Wilder reviewed by Kalyan Kumar Kalvagadda)
Change-Id: Ie0b7add60affe9901765339344abaa3944b8fc7a
diff --git a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/ConvertUtilTest.java b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/ConvertUtilTest.java
index e08d442..494e212 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/ConvertUtilTest.java
+++ b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/ConvertUtilTest.java
@@ -82,4 +82,22 @@
}
Assert.assertEquals(authorizables.size(), 2);
}
+
+ @Test
+ public void testTransactionalId() {
+ String hostname = "localhost";
+ String transactionalId = "t1";
+ Resource transactionalIdResource = new Resource(ResourceType$.MODULE$.fromString("transactionalId"), transactionalId);
+ List<Authorizable> authorizables = ConvertUtil.convertResourceToAuthorizable(hostname, transactionalIdResource);
+ for (Authorizable auth : authorizables) {
+ if (auth.getTypeName().equalsIgnoreCase(KafkaAuthorizable.AuthorizableType.TRANSACTIONALID.name())) {
+ Assert.assertEquals(auth.getName(), transactionalId);
+ } else if (auth.getTypeName().equalsIgnoreCase(KafkaAuthorizable.AuthorizableType.HOST.name())) {
+ Assert.assertEquals(auth.getName(), hostname);
+ } else {
+ Assert.fail("Unexpected type found: " + auth.getTypeName());
+ }
+ }
+ Assert.assertEquals(authorizables.size(), 2);
+ }
}
diff --git a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/SentryKafkaAuthorizerTest.java b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/SentryKafkaAuthorizerTest.java
index f40d8c2..84aa5b1 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/SentryKafkaAuthorizerTest.java
+++ b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/kafka/authorizer/SentryKafkaAuthorizerTest.java
@@ -40,6 +40,7 @@
private String resourceName;
private Resource clusterResource;
private Resource topic1Resource;
+ private Resource transactionalIdResource;
private KafkaConfig config;
public SentryKafkaAuthorizerTest() throws UnknownHostException {
@@ -49,6 +50,7 @@
resourceName = Resource$.MODULE$.ClusterResourceName();
clusterResource = new Resource(ResourceType$.MODULE$.fromString("cluster"), resourceName);
topic1Resource = new Resource(ResourceType$.MODULE$.fromString("topic"), "t1");
+ transactionalIdResource = new Resource(ResourceType$.MODULE$.fromString("transactionalId"), "tid1");
}
@Before
@@ -66,7 +68,7 @@
@Test
public void testAdmin() {
- KafkaPrincipal admin = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "admin");
+ KafkaPrincipal admin = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "admin_group");
RequestChannel.Session host1Session = new RequestChannel.Session(admin, testHostName1);
RequestChannel.Session host2Session = new RequestChannel.Session(admin, testHostName2);
@@ -79,7 +81,12 @@
Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Delete"), topic1Resource));
Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Alter"), topic1Resource));
Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Describe"), topic1Resource));
- Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("ClusterAction"),topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("ClusterAction"), topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("IdempotentWrite"), clusterResource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("AlterConfigs"), topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("DescribeConfigs"), clusterResource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Write"), transactionalIdResource));
+
Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Create"), clusterResource));
Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Describe"), clusterResource));
@@ -91,11 +98,15 @@
Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Alter"), topic1Resource));
Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Describe"), topic1Resource));
Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("ClusterAction"), topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("IdempotentWrite"), clusterResource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("AlterConfigs"), topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("DescribeConfigs"), clusterResource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Write"), transactionalIdResource));
}
@Test
public void testSubAdmin() {
- KafkaPrincipal admin = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "subadmin");
+ KafkaPrincipal admin = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "subadmin_group2");
RequestChannel.Session host1Session = new RequestChannel.Session(admin, testHostName1);
RequestChannel.Session host2Session = new RequestChannel.Session(admin, testHostName2);
@@ -108,7 +119,11 @@
Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Delete"), topic1Resource));
Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Alter"), topic1Resource));
Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Describe"), topic1Resource));
- Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("ClusterAction"),topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("ClusterAction"), topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("IdempotentWrite"), clusterResource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("AlterConfigs"), topic1Resource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("DescribeConfigs"), clusterResource));
+ Assert.assertTrue("Test failed.", authorizer.authorize(host1Session, Operation$.MODULE$.fromString("Write"), transactionalIdResource));
Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Create"), clusterResource));
Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Describe"), clusterResource));
@@ -120,6 +135,9 @@
Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Alter"), topic1Resource));
Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Describe"), topic1Resource));
Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("ClusterAction"), topic1Resource));
-
+ Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("IdempotentWrite"), clusterResource));
+ Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("AlterConfigs"), topic1Resource));
+ Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("DescribeConfigs"), clusterResource));
+ Assert.assertFalse("Test failed.", authorizer.authorize(host2Session, Operation$.MODULE$.fromString("Write"), transactionalIdResource));
}
}
diff --git a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/AbstractTestKafkaPolicyEngine.java b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/AbstractTestKafkaPolicyEngine.java
index 086b707..1ec2b19 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/AbstractTestKafkaPolicyEngine.java
+++ b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/AbstractTestKafkaPolicyEngine.java
@@ -43,6 +43,11 @@
private static final String PRODUCER_T1_ALL = "host=*->topic=t1->action=write";
private static final String PRODUCER_T1_HOST1 = "host=host1->topic=t1->action=write";
private static final String PRODUCER_T2_HOST2 = "host=host2->topic=t2->action=write";
+ private static final String PRODUCER_TI1_HOST1 = "host=host1->transactionalid=ti1->action=write";
+ private static final String PRODUCER_TI2_HOST2 = "host=host2->transactionalid=ti2->action=write";
+ private static final String PRODUCER_IDEMPOTENTWRITE = "host=host1->cluster=kafka-cluster->action=idempotentwrite";
+ private static final String CONFIG_ADMIN_HOST1 = "host=host1->cluster=kafka-cluster->action=describeconfigs";
+ private static final String CONFIG_ADMIN_T1_HOST2 = "host=host2->topic=t1->action=alterconfigs";
private static final String CONSUMER_PRODUCER_T1 = "host=host1->topic=t1->action=all";
private PolicyEngine policy;
@@ -133,6 +138,46 @@
}
@Test
+ public void testProducer3() throws Exception {
+ Set<String> expected = Sets.newTreeSet(Sets.newHashSet(PRODUCER_TI1_HOST1));
+ Assert.assertEquals(expected.toString(),
+ new TreeSet<String>(policy.getPrivileges(set("producer_group3"), ActiveRoleSet.ALL))
+ .toString());
+ }
+
+ @Test
+ public void testProducer4() throws Exception {
+ Set<String> expected = Sets.newTreeSet(Sets.newHashSet(PRODUCER_TI2_HOST2));
+ Assert.assertEquals(expected.toString(),
+ new TreeSet<String>(policy.getPrivileges(set("producer_group4"), ActiveRoleSet.ALL))
+ .toString());
+ }
+
+ @Test
+ public void testProducer5() throws Exception {
+ Set<String> expected = Sets.newTreeSet(Sets.newHashSet(PRODUCER_IDEMPOTENTWRITE));
+ Assert.assertEquals(expected.toString(),
+ new TreeSet<String>(policy.getPrivileges(set("producer_group5"), ActiveRoleSet.ALL))
+ .toString());
+ }
+
+ @Test
+ public void testConfigAdmin1() throws Exception {
+ Set<String> expected = Sets.newTreeSet(Sets.newHashSet(CONFIG_ADMIN_HOST1));
+ Assert.assertEquals(expected.toString(),
+ new TreeSet<String>(policy.getPrivileges(set("config_admin_group1"), ActiveRoleSet.ALL))
+ .toString());
+ }
+
+ @Test
+ public void testConfigAdmin2() throws Exception {
+ Set<String> expected = Sets.newTreeSet(Sets.newHashSet(CONFIG_ADMIN_T1_HOST2));
+ Assert.assertEquals(expected.toString(),
+ new TreeSet<String>(policy.getPrivileges(set("config_admin_group2"), ActiveRoleSet.ALL))
+ .toString());
+ }
+
+ @Test
public void testConsumerProducer0() throws Exception {
Set<String> expected = Sets.newTreeSet(Sets.newHashSet(CONSUMER_PRODUCER_T1));
Assert.assertEquals(expected.toString(),
@@ -144,7 +189,7 @@
public void testSubAdmin() throws Exception {
Set<String> expected = Sets.newTreeSet(Sets.newHashSet(ADMIN_HOST1));
Assert.assertEquals(expected.toString(),
- new TreeSet<String>(policy.getPrivileges(set("subadmin_group"), ActiveRoleSet.ALL))
+ new TreeSet<String>(policy.getPrivileges(set("subadmin_group1"), ActiveRoleSet.ALL))
.toString());
}
diff --git a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaAuthorizationProviderGeneralCases.java b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaAuthorizationProviderGeneralCases.java
index af92659..5ac3b0c 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaAuthorizationProviderGeneralCases.java
+++ b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaAuthorizationProviderGeneralCases.java
@@ -38,6 +38,7 @@
import org.apache.sentry.core.model.kafka.Host;
import org.apache.sentry.core.model.kafka.KafkaPrivilegeModel;
import org.apache.sentry.core.model.kafka.Topic;
+import org.apache.sentry.core.model.kafka.TransactionalId; ;
import org.apache.sentry.provider.common.HadoopGroupResourceAuthorizationProvider;
import org.apache.sentry.provider.common.ResourceAuthorizationProvider;
import org.apache.sentry.core.common.utils.PolicyFiles;
@@ -60,6 +61,8 @@
private static final Topic topic2 = new Topic("t2");
private static final ConsumerGroup cgroup1 = new ConsumerGroup("cg1");
private static final ConsumerGroup cgroup2 = new ConsumerGroup("cg2");
+ private static final TransactionalId transactionalId1 = new TransactionalId("ti1");
+ private static final TransactionalId transactionalId2 = new TransactionalId("ti2");
private static final KafkaAction ALL = new KafkaAction(KafkaActionConstant.ALL);
private static final KafkaAction READ = new KafkaAction(KafkaActionConstant.READ);
@@ -68,10 +71,13 @@
private static final KafkaAction DELETE = new KafkaAction(KafkaActionConstant.DELETE);
private static final KafkaAction ALTER = new KafkaAction(KafkaActionConstant.ALTER);
private static final KafkaAction DESCRIBE = new KafkaAction(KafkaActionConstant.DESCRIBE);
- private static final KafkaAction CLUSTER_ACTION = new KafkaAction(
- KafkaActionConstant.CLUSTER_ACTION);
+ private static final KafkaAction CLUSTER_ACTION = new KafkaAction(KafkaActionConstant.CLUSTER_ACTION);
+ private static final KafkaAction ALTER_CONFIGS = new KafkaAction(KafkaActionConstant.ALTER_CONFIGS);
+ private static final KafkaAction DESCRIBE_CONFIGS = new KafkaAction(KafkaActionConstant.DESCRIBE_CONFIGS);
+ private static final KafkaAction IDEMPOTENT_WRITE = new KafkaAction(KafkaActionConstant.IDEMPOTENT_WRITE);
- private static final Set<KafkaAction> allActions = Sets.newHashSet(ALL, READ, WRITE, CREATE, DELETE, ALTER, DESCRIBE, CLUSTER_ACTION);
+ private static final Set<KafkaAction> allActions = Sets.newHashSet(ALL, READ, WRITE, CREATE, DELETE, ALTER, DESCRIBE,
+ CLUSTER_ACTION, ALTER_CONFIGS, DESCRIBE_CONFIGS, IDEMPOTENT_WRITE);
private static final Subject ADMIN = new Subject("admin1");
private static final Subject SUB_ADMIN = new Subject("subadmin1");
@@ -81,16 +87,27 @@
private static final Subject PRODUCER0 = new Subject("producer0");
private static final Subject PRODUCER1 = new Subject("producer1");
private static final Subject PRODUCER2 = new Subject("producer2");
+ private static final Subject PRODUCER3 = new Subject("producer3");
+ private static final Subject PRODUCER4 = new Subject("producer4");
+ private static final Subject PRODUCER5 = new Subject("producer5");
+ private static final Subject CONFIG_ADMIN1 = new Subject("config_admin1");
+ private static final Subject CONFIG_ADMIN2 = new Subject("config_admin2");
private static final Subject CONSUMER_PRODUCER0 = new Subject("consumer_producer0");
private static final String ADMIN_GROUP = "admin_group";
- private static final String SUBADMIN_GROUP = "subadmin_group";
+ private static final String SUBADMIN_GROUP = "subadmin_group1";
private static final String CONSUMER_GROUP0 = "consumer_group0";
private static final String CONSUMER_GROUP1 = "consumer_group1";
private static final String CONSUMER_GROUP2 = "consumer_group2";
private static final String PRODUCER_GROUP0 = "producer_group0";
private static final String PRODUCER_GROUP1 = "producer_group1";
private static final String PRODUCER_GROUP2 = "producer_group2";
+ private static final String PRODUCER_GROUP3 = "producer_group3";
+ private static final String PRODUCER_GROUP4 = "producer_group4";
+ private static final String PRODUCER_GROUP5 = "producer_group5";
+ private static final String CONFIG_ADMIN_GROUP1 = "config_admin_group1";
+ private static final String CONFIG_ADMIN_GROUP2 = "config_admin_group2";
+
private static final String CONSUMER_PRODUCER_GROUP0 = "consumer_producer_group0";
static {
@@ -102,6 +119,11 @@
USER_TO_GROUP_MAP.putAll(PRODUCER0.getName(), Arrays.asList(PRODUCER_GROUP0));
USER_TO_GROUP_MAP.putAll(PRODUCER1.getName(), Arrays.asList(PRODUCER_GROUP1));
USER_TO_GROUP_MAP.putAll(PRODUCER2.getName(), Arrays.asList(PRODUCER_GROUP2));
+ USER_TO_GROUP_MAP.putAll(PRODUCER3.getName(), Arrays.asList(PRODUCER_GROUP3));
+ USER_TO_GROUP_MAP.putAll(PRODUCER4.getName(), Arrays.asList(PRODUCER_GROUP4));
+ USER_TO_GROUP_MAP.putAll(PRODUCER5.getName(), Arrays.asList(PRODUCER_GROUP5));
+ USER_TO_GROUP_MAP.putAll(CONFIG_ADMIN1.getName(), Arrays.asList(CONFIG_ADMIN_GROUP1));
+ USER_TO_GROUP_MAP.putAll(CONFIG_ADMIN2.getName(), Arrays.asList(CONFIG_ADMIN_GROUP2));
USER_TO_GROUP_MAP.putAll(CONSUMER_PRODUCER0.getName(), Arrays.asList(CONSUMER_PRODUCER_GROUP0));
}
@@ -171,12 +193,14 @@
Sets.newHashSet(action), READ.equals(action));
}
}
+
for (KafkaAction action : allActions) {
for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
doTestResourceAuthorizationProvider(CONSUMER1, Arrays.asList(host, topic1),
Sets.newHashSet(action), HOST_1.equals(host) && READ.equals(action));
}
}
+
for (KafkaAction action : allActions) {
for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
doTestResourceAuthorizationProvider(CONSUMER2, Arrays.asList(host, topic2),
@@ -193,18 +217,58 @@
Sets.newHashSet(action), WRITE.equals(action));
}
}
+
for (KafkaAction action : allActions) {
for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
doTestResourceAuthorizationProvider(PRODUCER1, Arrays.asList(host, topic1),
Sets.newHashSet(action), HOST_1.equals(host) && WRITE.equals(action));
}
}
+
for (KafkaAction action : allActions) {
for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
doTestResourceAuthorizationProvider(PRODUCER2, Arrays.asList(host, topic2),
Sets.newHashSet(action), HOST_2.equals(host) && WRITE.equals(action));
}
}
+
+ for (KafkaAction action : allActions) {
+ for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
+ doTestResourceAuthorizationProvider(PRODUCER3, Arrays.asList(host, transactionalId1),
+ Sets.newHashSet(action), HOST_1.equals(host) && WRITE.equals(action));
+ }
+ }
+
+ for (KafkaAction action : allActions) {
+ for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
+ doTestResourceAuthorizationProvider(PRODUCER4, Arrays.asList(host, transactionalId2),
+ Sets.newHashSet(action), HOST_2.equals(host) && WRITE.equals(action));
+ }
+ }
+
+ for (KafkaAction action : allActions) {
+ for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
+ doTestResourceAuthorizationProvider(PRODUCER5, Arrays.asList(host, cluster1),
+ Sets.newHashSet(action), HOST_1.equals(host) && IDEMPOTENT_WRITE.equals(action));
+ }
+ }
+ }
+
+ @Test
+ public void testConfigAdmin() throws Exception {
+ for (KafkaAction action : allActions) {
+ for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
+ doTestResourceAuthorizationProvider(CONFIG_ADMIN1, Arrays.asList(host, cluster1),
+ Sets.newHashSet(action), HOST_1.equals(host) && DESCRIBE_CONFIGS.equals(action));
+ }
+ }
+
+ for (KafkaAction action : allActions) {
+ for (Host host : Sets.newHashSet(HOST_1, HOST_2)) {
+ doTestResourceAuthorizationProvider(CONFIG_ADMIN2, Arrays.asList(host, topic1),
+ Sets.newHashSet(action), HOST_2.equals(host) && ALTER_CONFIGS.equals(action));
+ }
+ }
}
@Test
diff --git a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaModelAuthorizables.java b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaModelAuthorizables.java
index 62fbea7..2c2e2c6 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaModelAuthorizables.java
+++ b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaModelAuthorizables.java
@@ -27,6 +27,7 @@
import org.apache.sentry.core.model.kafka.Host;
import org.apache.sentry.core.model.kafka.KafkaModelAuthorizables;
import org.apache.sentry.core.model.kafka.Topic;
+import org.apache.sentry.core.model.kafka.TransactionalId;
import org.apache.shiro.config.ConfigurationException;
import org.junit.Test;
@@ -71,6 +72,9 @@
ConsumerGroup consumergroup1 = (ConsumerGroup)KafkaModelAuthorizables.from("ConsumerGroup=CG1");
assertEquals("CG1", consumergroup1.getName());
+
+ TransactionalId transactionalId1 = (TransactionalId) KafkaModelAuthorizables.from("TransactionalId=tRaNs1");
+ assertEquals("tRaNs1", transactionalId1.getName());
}
@Test
diff --git a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaPrivilegeValidator.java b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaPrivilegeValidator.java
index ba66d43..61d5554 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaPrivilegeValidator.java
+++ b/sentry-binding/sentry-binding-kafka/src/test/java/org/apache/sentry/policy/kafka/TestKafkaPrivilegeValidator.java
@@ -71,6 +71,11 @@
} catch (ConfigurationException ex) {
Assert.fail("Not expected ConfigurationException");
}
+ try {
+ kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->transactionalid=t1->action=write"));
+ } catch (ConfigurationException ex) {
+ Assert.fail("Not expected ConfigurationException");
+ }
}
@Test
@@ -104,6 +109,16 @@
}
@Test
+ public void testInvalidTransactionalIdResource() throws Exception {
+ KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator();
+ try {
+ kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->transationalid=t1->action=write"));
+ Assert.fail("Expected ConfigurationException");
+ } catch (ConfigurationException ex) {
+ }
+ }
+
+ @Test
public void testInvalidConsumerGroupResource() throws Exception {
KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator();
try {
@@ -144,6 +159,12 @@
} catch (ConfigurationException ex) {
Assert.assertEquals(KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage());
}
+ try {
+ kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->topic=t1->transactionalid=t1->action=read"));
+ Assert.fail("Kafka privilege can have one Host authorizable, at most one non Host authorizable and one action.");
+ } catch (ConfigurationException ex) {
+ Assert.assertEquals(KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage());
+ }
}
@Test
diff --git a/sentry-binding/sentry-binding-kafka/src/test/resources/kafka-policy-test-authz-provider.ini b/sentry-binding/sentry-binding-kafka/src/test/resources/kafka-policy-test-authz-provider.ini
index 1951aba..1effd42 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/resources/kafka-policy-test-authz-provider.ini
+++ b/sentry-binding/sentry-binding-kafka/src/test/resources/kafka-policy-test-authz-provider.ini
@@ -17,22 +17,34 @@
[groups]
admin_group = admin_all
-subadmin_group = admin_host1
+subadmin_group1 = admin_host1
+subadmin_group2 = admin_host1234
consumer_group0 = consumer_t1_all
consumer_group1 = consumer_t1_host1
consumer_group2 = consumer_t2_host2
producer_group0 = producer_t1_all
producer_group1 = producer_t1_host1
producer_group2 = producer_t2_host2
+producer_group3 = producer_ti1_host1
+producer_group4 = producer_ti2_host2
+producer_group5 = producer_idempotentwrite_host1
+config_admin_group1 = config_admin_host1
+config_admin_group2 = config_admin_t1_host2
consumer_producer_group0 = consumer_producer_t1
[roles]
admin_all = host=*->action=all
admin_host1 = host=host1->action=all
+admin_host1234 = host=1.2.3.4->action=all
consumer_t1_all = host=*->topic=t1->action=read
consumer_t1_host1 = host=host1->topic=t1->action=read
consumer_t2_host2 = host=host2->topic=t2->action=read
producer_t1_all = host=*->topic=t1->action=write
producer_t1_host1 = host=host1->topic=t1->action=write
producer_t2_host2 = host=host2->topic=t2->action=write
+producer_ti1_host1 = host=host1->transactionalid=ti1->action=write
+producer_ti2_host2 = host=host2->transactionalid=ti2->action=write
+producer_idempotentwrite_host1 = host=host1->cluster=kafka-cluster->action=idempotentwrite
+config_admin_host1 = host=host1->cluster=kafka-cluster->action=describeconfigs
+config_admin_t1_host2 = host=host2->topic=t1->action=alterconfigs
consumer_producer_t1 = host=host1->topic=t1->action=all
diff --git a/sentry-binding/sentry-binding-kafka/src/test/resources/sentry-site.xml b/sentry-binding/sentry-binding-kafka/src/test/resources/sentry-site.xml
index 6383481..adc9239 100644
--- a/sentry-binding/sentry-binding-kafka/src/test/resources/sentry-site.xml
+++ b/sentry-binding/sentry-binding-kafka/src/test/resources/sentry-site.xml
@@ -28,7 +28,7 @@
</property>
<property>
<name>sentry.kafka.provider.resource</name>
- <value>classpath:test-authz-provider.ini</value>
+ <value>classpath:kafka-policy-test-authz-provider.ini</value>
</property>
<property>
<name>sentry.kafka.policy.engine</name>
diff --git a/sentry-binding/sentry-binding-kafka/src/test/resources/test-authz-provider.ini b/sentry-binding/sentry-binding-kafka/src/test/resources/test-authz-provider.ini
deleted file mode 100644
index 520e1d0..0000000
--- a/sentry-binding/sentry-binding-kafka/src/test/resources/test-authz-provider.ini
+++ /dev/null
@@ -1,38 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-[groups]
-admin = admin_all
-subadmin = admin_host1
-consumer0 = consumer_t1_all
-consumer1 = consumer_t1_host1
-consumer2 = consumer_t2_host2
-producer0 = producer_t1_all
-producer1 = producer_t1_host1
-producer2 = producer_t2_host2
-consumer_producer0 = consumer_producer_t1
-
-[roles]
-admin_all = host=*->action=all
-admin_host1 = host=1.2.3.4->action=all
-consumer_t1_all = host=*->topic=t1->action=read
-consumer_t1_host1 = host=host1->topic=t1->action=read
-consumer_t2_host2 = host=host2->topic=t2->action=read
-producer_t1_all = host=*->topic=t1->action=write
-producer_t1_host1 = host=host1->topic=t1->action=write
-producer_t2_host2 = host=host2->topic=t2->action=write
-consumer_producer_t1 = host=host1->topic=t1->action=all
diff --git a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionConstant.java b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionConstant.java
index 17d7fb7..a95469b 100644
--- a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionConstant.java
+++ b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionConstant.java
@@ -29,6 +29,9 @@
public static final String ALTER = "alter";
public static final String DESCRIBE = "describe";
public static final String CLUSTER_ACTION = "clusteraction";
+ public static final String ALTER_CONFIGS = "alterconfigs";
+ public static final String DESCRIBE_CONFIGS = "describeconfigs";
+ public static final String IDEMPOTENT_WRITE = "idempotentwrite";
public static final String actionName = "action";
}
diff --git a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionFactory.java b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionFactory.java
index a1fec1f..1706057 100644
--- a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionFactory.java
+++ b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaActionFactory.java
@@ -53,8 +53,12 @@
ALTER(KafkaActionConstant.ALTER, 16),
DESCRIBE(KafkaActionConstant.DESCRIBE, 32),
CLUSTERACTION(KafkaActionConstant.CLUSTER_ACTION, 64),
+ ALTERCONFIGS(KafkaActionConstant.ALTER_CONFIGS, 128),
+ DESCRIBECONFIGS(KafkaActionConstant.DESCRIBE_CONFIGS, 256),
+ IDEMPOTENTWRITE(KafkaActionConstant.IDEMPOTENT_WRITE, 512),
ALL(KafkaActionConstant.ALL, READ.getCode() | WRITE.getCode() | CREATE.getCode()
- | DELETE.getCode() | ALTER.getCode()| DESCRIBE.getCode() | CLUSTERACTION.getCode());
+ | DELETE.getCode() | ALTER.getCode()| DESCRIBE.getCode() | CLUSTERACTION.getCode()
+ | ALTERCONFIGS.getCode() | DESCRIBECONFIGS.getCode() | IDEMPOTENTWRITE.getCode());
private String name;
private int code;
diff --git a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaAuthorizable.java b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaAuthorizable.java
index 52ae614..6a0c6f7 100644
--- a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaAuthorizable.java
+++ b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaAuthorizable.java
@@ -39,6 +39,9 @@
* CONSUMERGROUP -> Kafka ConsumerGroup resource, users are required to have access to this resource
* in order to perform ConsumerGroup level actions like joining a consumer group,
* querying offset for a partition for a particular consumer group.
+ *
+ * TRANSACTIONALID -> This resource represents actions related to transactions, such as committing.
+ *
*/
public interface KafkaAuthorizable extends Authorizable {
/**
@@ -48,7 +51,8 @@
CLUSTER,
HOST,
TOPIC,
- CONSUMERGROUP
+ CONSUMERGROUP,
+ TRANSACTIONALID
};
/**
diff --git a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaModelAuthorizables.java b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaModelAuthorizables.java
index 45a1148..7a0ecf5 100644
--- a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaModelAuthorizables.java
+++ b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaModelAuthorizables.java
@@ -50,6 +50,8 @@
return new Topic(name);
case CONSUMERGROUP:
return new ConsumerGroup(name);
+ case TRANSACTIONALID:
+ return new TransactionalId(name);
default:
return null;
}
diff --git a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaPrivilegeModel.java b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaPrivilegeModel.java
index e460874..cbf741b 100644
--- a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaPrivilegeModel.java
+++ b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/KafkaPrivilegeModel.java
@@ -46,6 +46,8 @@
ImplyMethodType.STRING_CASE_SENSITIVE);
implyMethodMap.put(KafkaAuthorizable.AuthorizableType.CONSUMERGROUP.name().toLowerCase(),
ImplyMethodType.STRING_CASE_SENSITIVE);
+ implyMethodMap.put(KafkaAuthorizable.AuthorizableType.TRANSACTIONALID.name().toLowerCase(),
+ ImplyMethodType.STRING_CASE_SENSITIVE);
}
@Override
diff --git a/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/TransactionalId.java b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/TransactionalId.java
new file mode 100644
index 0000000..2a91c49
--- /dev/null
+++ b/sentry-core/sentry-core-model-kafka/src/main/java/org/apache/sentry/core/model/kafka/TransactionalId.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.sentry.core.model.kafka;
+
+/**
+ * Represents transactional ID authorizable in Kafka model.
+ */
+public class TransactionalId implements KafkaAuthorizable {
+ private String name;
+
+ /**
+ * Create a transactional ID authorizable for Kafka cluster of a given name.
+ *
+ * @param name Name of Kafka transactional ID.
+ */
+ public TransactionalId(String name) {
+ this.name = name;
+ }
+
+ /**
+ * Get type of Kafka's transactional ID authorizable.
+ *
+ * @return Type of Kafka's transactional ID authorizable.
+ */
+ @Override
+ public AuthorizableType getAuthzType() { return AuthorizableType.TRANSACTIONALID; }
+
+ /**
+ * Get name of Kafka's transactional ID.
+ *
+ * @return Name of Kafka's transactional ID.
+ */
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Get type name of Kafka's transactional ID authorizable.
+ *
+ * @return Type name of Kafka's transactional ID authorizable.
+ */
+ @Override
+ public String getTypeName() {
+ return getAuthzType().name();
+ }
+}
\ No newline at end of file
diff --git a/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAction.java b/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAction.java
index dcab5d5..f450d21 100644
--- a/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAction.java
+++ b/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAction.java
@@ -30,101 +30,84 @@
private KafkaActionFactory factory = KafkaActionFactory.getInstance();
@Test
- public void testImpliesAction() {
- KafkaAction readAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.READ);
- KafkaAction writeAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.WRITE);
- KafkaAction createAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.CREATE);
- KafkaAction deleteAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.DELETE);
- KafkaAction alterAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.ALTER);
- KafkaAction describeAction =
- (KafkaAction) factory.getActionByName(KafkaActionConstant.DESCRIBE);
- KafkaAction adminAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION);
- KafkaAction allAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.ALL);
+ public void testAllActionImpliesAll() {
+ KafkaAction allAction = factory.getActionByName(KafkaActionConstant.ALL);
- assertTrue(allAction.implies(readAction));
- assertTrue(allAction.implies(writeAction));
- assertTrue(allAction.implies(createAction));
- assertTrue(allAction.implies(deleteAction));
- assertTrue(allAction.implies(alterAction));
- assertTrue(allAction.implies(describeAction));
- assertTrue(allAction.implies(adminAction));
- assertTrue(allAction.implies(allAction));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.READ)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.WRITE)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.CREATE)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.DELETE)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.ALTER)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.DESCRIBE)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.ALTER_CONFIGS)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.DESCRIBE_CONFIGS)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.IDEMPOTENT_WRITE)));
+ assertTrue(allAction.implies(factory.getActionByName(KafkaActionConstant.ALL)));
+ }
- assertTrue(readAction.implies(readAction));
- assertFalse(readAction.implies(writeAction));
- assertFalse(readAction.implies(createAction));
- assertFalse(readAction.implies(deleteAction));
- assertFalse(readAction.implies(alterAction));
- assertFalse(readAction.implies(describeAction));
- assertFalse(readAction.implies(adminAction));
- assertFalse(readAction.implies(allAction));
+ @Test
+ public void testActionImpliesSelf() {
+ KafkaAction[] actions = new KafkaAction[]{
+ factory.getActionByName(KafkaActionConstant.READ),
+ factory.getActionByName(KafkaActionConstant.WRITE),
+ factory.getActionByName(KafkaActionConstant.CREATE),
+ factory.getActionByName(KafkaActionConstant.DELETE),
+ factory.getActionByName(KafkaActionConstant.ALTER),
+ factory.getActionByName(KafkaActionConstant.DESCRIBE),
+ factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION),
+ factory.getActionByName(KafkaActionConstant.ALTER_CONFIGS),
+ factory.getActionByName(KafkaActionConstant.DESCRIBE_CONFIGS),
+ factory.getActionByName(KafkaActionConstant.IDEMPOTENT_WRITE),
+ factory.getActionByName(KafkaActionConstant.ALL)
+ };
- assertFalse(writeAction.implies(readAction));
- assertTrue(writeAction.implies(writeAction));
- assertFalse(writeAction.implies(createAction));
- assertFalse(writeAction.implies(deleteAction));
- assertFalse(writeAction.implies(alterAction));
- assertFalse(writeAction.implies(describeAction));
- assertFalse(writeAction.implies(adminAction));
- assertFalse(writeAction.implies(allAction));
+ for(KafkaAction action : actions){
+ assertTrue(action.implies(action));
+ }
+ }
- assertFalse(createAction.implies(readAction));
- assertFalse(createAction.implies(writeAction));
- assertTrue(createAction.implies(createAction));
- assertFalse(createAction.implies(deleteAction));
- assertFalse(createAction.implies(alterAction));
- assertFalse(createAction.implies(describeAction));
- assertFalse(createAction.implies(adminAction));
- assertFalse(createAction.implies(allAction));
+ @Test
+ public void testNonAllActionDoesNotImplyOthers() {
+ KafkaAction allAction = factory.getActionByName(KafkaActionConstant.ALL);
- assertFalse(deleteAction.implies(readAction));
- assertFalse(deleteAction.implies(writeAction));
- assertFalse(deleteAction.implies(createAction));
- assertTrue(deleteAction.implies(deleteAction));
- assertFalse(deleteAction.implies(alterAction));
- assertFalse(deleteAction.implies(describeAction));
- assertFalse(deleteAction.implies(adminAction));
- assertFalse(deleteAction.implies(allAction));
+ KafkaAction[] actions = new KafkaAction[]{
+ factory.getActionByName(KafkaActionConstant.READ),
+ factory.getActionByName(KafkaActionConstant.WRITE),
+ factory.getActionByName(KafkaActionConstant.CREATE),
+ factory.getActionByName(KafkaActionConstant.DELETE),
+ factory.getActionByName(KafkaActionConstant.ALTER),
+ factory.getActionByName(KafkaActionConstant.DESCRIBE),
+ factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION),
+ factory.getActionByName(KafkaActionConstant.ALTER_CONFIGS),
+ factory.getActionByName(KafkaActionConstant.DESCRIBE_CONFIGS),
+ factory.getActionByName(KafkaActionConstant.IDEMPOTENT_WRITE)
+ };
- assertFalse(alterAction.implies(readAction));
- assertFalse(alterAction.implies(writeAction));
- assertFalse(alterAction.implies(createAction));
- assertFalse(alterAction.implies(deleteAction));
- assertTrue(alterAction.implies(alterAction));
- assertFalse(alterAction.implies(describeAction));
- assertFalse(alterAction.implies(adminAction));
- assertFalse(alterAction.implies(allAction));
+ for(KafkaAction action : actions) {
+ for(KafkaAction action2 : actions) {
+ if (action != action2) {
+ assertFalse(action.implies(action2));
+ }
+ }
- assertFalse(describeAction.implies(readAction));
- assertFalse(describeAction.implies(writeAction));
- assertFalse(describeAction.implies(createAction));
- assertFalse(describeAction.implies(deleteAction));
- assertFalse(describeAction.implies(alterAction));
- assertTrue(describeAction.implies(describeAction));
- assertFalse(describeAction.implies(adminAction));
- assertFalse(describeAction.implies(allAction));
-
- assertFalse(adminAction.implies(readAction));
- assertFalse(adminAction.implies(writeAction));
- assertFalse(adminAction.implies(createAction));
- assertFalse(adminAction.implies(deleteAction));
- assertFalse(adminAction.implies(alterAction));
- assertFalse(adminAction.implies(describeAction));
- assertTrue(adminAction.implies(adminAction));
- assertFalse(adminAction.implies(allAction));
+ assertFalse(action.implies(allAction));
+ }
}
@Test
public void testGetActionByName() throws Exception {
- KafkaAction readAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.READ);
- KafkaAction writeAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.WRITE);
- KafkaAction createAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.CREATE);
- KafkaAction deleteAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.DELETE);
- KafkaAction alterAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.ALTER);
- KafkaAction describeAction =
- (KafkaAction) factory.getActionByName(KafkaActionConstant.DESCRIBE);
- KafkaAction adminAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION);
- KafkaAction allAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.ALL);
+ KafkaAction readAction = factory.getActionByName(KafkaActionConstant.READ);
+ KafkaAction writeAction = factory.getActionByName(KafkaActionConstant.WRITE);
+ KafkaAction createAction = factory.getActionByName(KafkaActionConstant.CREATE);
+ KafkaAction deleteAction = factory.getActionByName(KafkaActionConstant.DELETE);
+ KafkaAction alterAction = factory.getActionByName(KafkaActionConstant.ALTER);
+ KafkaAction describeAction = factory.getActionByName(KafkaActionConstant.DESCRIBE);
+ KafkaAction adminAction = factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION);
+ KafkaAction alterConfigsAction = factory.getActionByName(KafkaActionConstant.ALTER_CONFIGS);
+ KafkaAction describeConfigsAction = factory.getActionByName(KafkaActionConstant.DESCRIBE_CONFIGS);
+ KafkaAction idempotentWriteAction = factory.getActionByName(KafkaActionConstant.IDEMPOTENT_WRITE);
+ KafkaAction allAction = factory.getActionByName(KafkaActionConstant.ALL);
assertTrue(readAction.equals(new KafkaAction(KafkaActionConstant.READ)));
assertTrue(writeAction.equals(new KafkaAction(KafkaActionConstant.WRITE)));
@@ -133,20 +116,25 @@
assertTrue(alterAction.equals(new KafkaAction(KafkaActionConstant.ALTER)));
assertTrue(describeAction.equals(new KafkaAction(KafkaActionConstant.DESCRIBE)));
assertTrue(adminAction.equals(new KafkaAction(KafkaActionConstant.CLUSTER_ACTION)));
+ assertTrue(alterConfigsAction.equals(new KafkaAction(KafkaActionConstant.ALTER_CONFIGS)));
+ assertTrue(describeConfigsAction.equals(new KafkaAction(KafkaActionConstant.DESCRIBE_CONFIGS)));
+ assertTrue(idempotentWriteAction.equals(new KafkaAction(KafkaActionConstant.IDEMPOTENT_WRITE)));
assertTrue(allAction.equals(new KafkaAction(KafkaActionConstant.ALL)));
}
@Test
public void testGetActionsByCode() throws Exception {
- KafkaAction readAction = new KafkaAction(KafkaActionConstant.READ);
- KafkaAction writeAction = new KafkaAction(KafkaActionConstant.WRITE);
- KafkaAction createAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.CREATE);
- KafkaAction deleteAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.DELETE);
- KafkaAction alterAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.ALTER);
- KafkaAction describeAction =
- (KafkaAction) factory.getActionByName(KafkaActionConstant.DESCRIBE);
- KafkaAction adminAction = (KafkaAction) factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION);
- KafkaAction allAction = new KafkaAction(KafkaActionConstant.ALL);
+ KafkaAction readAction = factory.getActionByName(KafkaActionConstant.READ);
+ KafkaAction writeAction = factory.getActionByName(KafkaActionConstant.WRITE);
+ KafkaAction createAction = factory.getActionByName(KafkaActionConstant.CREATE);
+ KafkaAction deleteAction = factory.getActionByName(KafkaActionConstant.DELETE);
+ KafkaAction alterAction = factory.getActionByName(KafkaActionConstant.ALTER);
+ KafkaAction describeAction = factory.getActionByName(KafkaActionConstant.DESCRIBE);
+ KafkaAction adminAction = factory.getActionByName(KafkaActionConstant.CLUSTER_ACTION);
+ KafkaAction alterConfigsAction = factory.getActionByName(KafkaActionConstant.ALTER_CONFIGS);
+ KafkaAction describeConfigsAction = factory.getActionByName(KafkaActionConstant.DESCRIBE_CONFIGS);
+ KafkaAction idempotentWriteAction = factory.getActionByName(KafkaActionConstant.IDEMPOTENT_WRITE);
+ KafkaAction allAction = factory.getActionByName(KafkaActionConstant.ALL);
assertEquals(Lists.newArrayList(readAction),
factory.getActionsByCode(readAction.getActionCode()));
@@ -162,8 +150,15 @@
factory.getActionsByCode(describeAction.getActionCode()));
assertEquals(Lists.newArrayList(adminAction),
factory.getActionsByCode(adminAction.getActionCode()));
+ assertEquals(Lists.newArrayList(alterConfigsAction),
+ factory.getActionsByCode(alterConfigsAction.getActionCode()));
+ assertEquals(Lists.newArrayList(describeConfigsAction),
+ factory.getActionsByCode(describeConfigsAction.getActionCode()));
+ assertEquals(Lists.newArrayList(idempotentWriteAction),
+ factory.getActionsByCode(idempotentWriteAction.getActionCode()));
assertEquals(Lists.newArrayList(readAction, writeAction, createAction, deleteAction,
- alterAction, describeAction, adminAction), factory.getActionsByCode(allAction
+ alterAction, describeAction, adminAction,
+ alterConfigsAction, describeConfigsAction, idempotentWriteAction), factory.getActionsByCode(allAction
.getActionCode()));
}
diff --git a/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAuthorizable.java b/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAuthorizable.java
index 04316f2..e00fbbd 100644
--- a/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAuthorizable.java
+++ b/sentry-core/sentry-core-model-kafka/src/test/java/org/apache/sentry/core/model/kafka/TestKafkaAuthorizable.java
@@ -41,6 +41,9 @@
ConsumerGroup consumerGroup = new ConsumerGroup(name);
Assert.assertEquals(consumerGroup.getName(), name);
+
+ TransactionalId transactionalId = new TransactionalId(name);
+ Assert.assertEquals(transactionalId.getName(), name);
}
@Test
@@ -56,5 +59,8 @@
ConsumerGroup consumerGroup = new ConsumerGroup("consumerGroup1");
Assert.assertEquals(consumerGroup.getAuthzType(), AuthorizableType.CONSUMERGROUP);
+
+ TransactionalId transactionalId = new TransactionalId("transactionalId1");
+ Assert.assertEquals(transactionalId.getAuthzType(), AuthorizableType.TRANSACTIONALID);
}
}
