Upgrade pulsar to 2.8.0-rc-202106071430

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java

@@ -369,10 +369,12 @@ public void initGroupCoordinator(BrokerService service) throws Exception {
             .build();
 
         PulsarAdmin pulsarAdmin = service.pulsar().getAdminClient();
-        ClusterData clusterData = new ClusterData(service.getPulsar().getWebServiceAddress(),
-                                                  service.getPulsar().getWebServiceAddressTls(),
-                                                  service.getPulsar().getBrokerServiceUrl(),
-                                                  service.getPulsar().getBrokerServiceUrlTls());
+        final ClusterData clusterData = ClusterData.builder()
+                .serviceUrl(brokerService.getPulsar().getWebServiceAddress())
+                .serviceUrlTls(brokerService.getPulsar().getWebServiceAddressTls())
+                .brokerServiceUrl(brokerService.getPulsar().getBrokerServiceUrl())
+                .brokerServiceUrlTls(brokerService.getPulsar().getBrokerServiceUrlTls())
+                .build();
         MetadataUtils.createOffsetMetadataIfMissing(pulsarAdmin, clusterData, kafkaConfig);
 
 
@@ -404,10 +406,12 @@ public void initTransactionCoordinator() throws Exception {
                 .build();
 
         PulsarAdmin pulsarAdmin = brokerService.getPulsar().getAdminClient();
-        ClusterData clusterData = new ClusterData(brokerService.getPulsar().getWebServiceAddress(),
-                                                  brokerService.getPulsar().getWebServiceAddressTls(),
-                                                  brokerService.getPulsar().getBrokerServiceUrl(),
-                                                  brokerService.getPulsar().getBrokerServiceUrlTls());
+        final ClusterData clusterData = ClusterData.builder()
+                .serviceUrl(brokerService.getPulsar().getWebServiceAddress())
+                .serviceUrlTls(brokerService.getPulsar().getWebServiceAddressTls())
+                .brokerServiceUrl(brokerService.getPulsar().getBrokerServiceUrl())
+                .brokerServiceUrlTls(brokerService.getPulsar().getBrokerServiceUrlTls())
+                .build();
 
         MetadataUtils.createTxnMetadataIfMissing(pulsarAdmin, clusterData, kafkaConfig);
 

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtils.java

@@ -15,6 +15,7 @@
 
 import com.google.common.collect.Sets;
 import io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration;
+import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
@@ -114,7 +115,10 @@ private static void createKafkaMetadataIfMissing(PulsarAdmin pulsarAdmin,
             if (!tenants.getTenants().contains(kafkaMetadataTenant)) {
                 log.info("Tenant: {} does not exist, creating it ...", kafkaMetadataTenant);
                 tenants.createTenant(kafkaMetadataTenant,
-                        new TenantInfo(Sets.newHashSet(conf.getSuperUserRoles()), Sets.newHashSet(cluster)));
+                        TenantInfo.builder()
+                                .adminRoles(conf.getSuperUserRoles())
+                                .allowedClusters(Collections.singleton(cluster))
+                                .build());
             } else {
                 TenantInfo kafkaMetadataTenantInfo = tenants.getTenantInfo(kafkaMetadataTenant);
                 Set<String> allowedClusters = kafkaMetadataTenantInfo.getAllowedClusters();

kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtilsTest.java

@@ -48,7 +48,7 @@ public class MetadataUtilsTest {
     public void testCreateKafkaMetadataIfMissing() throws Exception {
         KopTopic.initialize("public/default");
         KafkaServiceConfiguration conf = new KafkaServiceConfiguration();
-        ClusterData clusterData = new ClusterData();
+        ClusterData clusterData = ClusterData.builder().build();
         conf.setClusterName("test");
         conf.setKafkaMetadataTenant("public");
         conf.setKafkaMetadataNamespace("default");
@@ -82,7 +82,7 @@ public void testCreateKafkaMetadataIfMissing() throws Exception {
         doReturn(mockNamespaces).when(mockPulsarAdmin).namespaces();
         doReturn(mockTopics).when(mockPulsarAdmin).topics();
 
-        TenantInfo partialTenant = new TenantInfo();
+        TenantInfo partialTenant = TenantInfo.builder().build();
         doReturn(partialTenant).when(mockTenants).getTenantInfo(eq(conf.getKafkaMetadataTenant()));
 
         MetadataUtils.createOffsetMetadataIfMissing(mockPulsarAdmin, clusterData, conf);
@@ -120,8 +120,10 @@ public void testCreateKafkaMetadataIfMissing() throws Exception {
 
         doReturn(Lists.newArrayList("public")).when(mockTenants).getTenants();
 
-        partialTenant = new TenantInfo(Sets.newHashSet(conf.getSuperUserRoles()),
-            Sets.newHashSet("other-cluster"));
+        partialTenant = TenantInfo.builder()
+                .adminRoles(conf.getSuperUserRoles())
+                .allowedClusters(Sets.newHashSet("other-cluster"))
+                .build();
         doReturn(partialTenant).when(mockTenants).getTenantInfo(eq(conf.getKafkaMetadataTenant()));
 
         doReturn(Lists.newArrayList("test")).when(mockNamespaces).getNamespaces("public");

pom.xml

@@ -48,7 +48,7 @@
     <lombok.version>1.18.4</lombok.version>
     <mockito.version>2.22.0</mockito.version>
     <pulsar.group.id>io.streamnative</pulsar.group.id>
-    <pulsar.version>2.8.0-rc-202105291235</pulsar.version>
+    <pulsar.version>2.8.0-rc-202106071430</pulsar.version>
     <slf4j.version>1.7.25</slf4j.version>
     <spotbugs-annotations.version>3.1.8</spotbugs-annotations.version>
     <testcontainers.version>1.15.1</testcontainers.version>

tests/src/test/java/io/streamnative/pulsar/handlers/kop/DifferentNamespaceTestBase.java

@@ -16,7 +16,6 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
-import com.google.common.collect.Sets;
 import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -73,7 +72,10 @@ protected void setup() throws Exception {
         super.internalSetup();
 
         admin.tenants().createTenant(ANOTHER_TENANT,
-                new TenantInfo(Sets.newHashSet("admin_user"), Sets.newHashSet(super.configClusterName)));
+                TenantInfo.builder()
+                        .adminRoles(Collections.singleton("admin_user"))
+                        .allowedClusters(Collections.singleton(configClusterName))
+                        .build());
         admin.namespaces().createNamespace(ANOTHER_TENANT + "/" + ANOTHER_NAMESPACE);
     }
 

tests/src/test/java/io/streamnative/pulsar/handlers/kop/DistributedClusterTest.java

@@ -36,9 +36,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.pulsar.broker.PulsarService;
-import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.junit.Assert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -157,22 +155,6 @@ protected void stopBroker() throws Exception {
     public void setup() throws Exception {
         super.internalSetup();
 
-        if (!admin.clusters().getClusters().contains(configClusterName)) {
-            // so that clients can test short names
-            admin.clusters().createCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        } else {
-            admin.clusters().updateCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        }
-
-        if (!admin.tenants().getTenants().contains("public")) {
-            admin.tenants().createTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        } else {
-            admin.tenants().updateTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        }
         if (!admin.namespaces().getNamespaces("public").contains("public/default")) {
             admin.namespaces().createNamespace("public/default");
             admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));

tests/src/test/java/io/streamnative/pulsar/handlers/kop/EntryPublishTimeKafkaFormatTest.java

@@ -49,7 +49,7 @@
 public class EntryPublishTimeKafkaFormatTest extends EntryPublishTimeTest {
     private static final Logger log = LoggerFactory.getLogger(EntryPublishTimeKafkaFormatTest.class);
 
-    public EntryPublishTimeKafkaFormatTest(String format) {
+    public EntryPublishTimeKafkaFormatTest() {
         super("kafka");
     }
 

tests/src/test/java/io/streamnative/pulsar/handlers/kop/EntryPublishTimeTest.java

@@ -16,7 +16,6 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 
-import com.google.common.collect.Sets;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupCoordinator;
@@ -25,9 +24,6 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import org.apache.pulsar.broker.protocol.ProtocolHandler;
-import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.AfterMethod;
@@ -51,36 +47,6 @@ public EntryPublishTimeTest(String format) {
     @Override
     protected void setup() throws Exception {
         super.internalSetup();
-        if (!admin.clusters().getClusters().contains(configClusterName)) {
-            // so that clients can test short names
-            admin.clusters().createCluster(configClusterName,
-                    new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        } else {
-            admin.clusters().updateCluster(configClusterName,
-                    new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        }
-
-        if (!admin.tenants().getTenants().contains("public")) {
-            admin.tenants().createTenant("public",
-                    new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        } else {
-            admin.tenants().updateTenant("public",
-                    new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        }
-        if (!admin.namespaces().getNamespaces("public").contains("public/default")) {
-            admin.namespaces().createNamespace("public/default");
-            admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));
-            admin.namespaces().setRetention("public/default",
-                    new RetentionPolicies(60, 1000));
-        }
-        if (!admin.namespaces().getNamespaces("public").contains("public/__kafka")) {
-            admin.namespaces().createNamespace("public/__kafka");
-            admin.namespaces().setNamespaceReplicationClusters("public/__kafka", Sets.newHashSet("test"));
-            admin.namespaces().setRetention("public/__kafka",
-                    new RetentionPolicies(-1, -1));
-        }
-
-        log.info("created namespaces, init handler");
 
         ProtocolHandler handler = pulsar.getProtocolHandlers().protocol("kafka");
         GroupCoordinator groupCoordinator = ((KafkaProtocolHandler) handler).getGroupCoordinator();

tests/src/test/java/io/streamnative/pulsar/handlers/kop/InnerTopicProtectionTest.java

@@ -26,12 +26,10 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.testng.Assert;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
 /**
@@ -97,28 +95,12 @@ protected void resetConfig() {
             brokerPort, brokerWebservicePort, kafkaBrokerPort);
     }
 
-    @BeforeMethod
+    @BeforeClass
     @Override
     protected void setup() throws Exception {
         super.internalSetup();
         log.info("success internal setup");
 
-        if (!admin.clusters().getClusters().contains(configClusterName)) {
-            // so that clients can test short names
-            admin.clusters().createCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        } else {
-            admin.clusters().updateCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        }
-
-        if (!admin.tenants().getTenants().contains("public")) {
-            admin.tenants().createTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        } else {
-            admin.tenants().updateTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        }
         if (!admin.namespaces().getNamespaces("public").contains("public/default")) {
             admin.namespaces().createNamespace("public/default");
             admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));
@@ -133,7 +115,7 @@ protected void setup() throws Exception {
         }
     }
 
-    @AfterMethod
+    @AfterClass
     @Override
     protected void cleanup() throws Exception {
         super.internalCleanup();

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

@@ -71,9 +71,7 @@
 import org.apache.kafka.common.requests.RequestHeader;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.pulsar.broker.protocol.ProtocolHandler;
-import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Ignore;
@@ -102,22 +100,6 @@ protected void setup() throws Exception {
         super.internalSetup();
         log.info("success internal setup");
 
-        if (!admin.clusters().getClusters().contains(configClusterName)) {
-            // so that clients can test short names
-            admin.clusters().createCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        } else {
-            admin.clusters().updateCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        }
-
-        if (!admin.tenants().getTenants().contains("public")) {
-            admin.tenants().createTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        } else {
-            admin.tenants().updateTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        }
         if (!admin.namespaces().getNamespaces("public").contains("public/default")) {
             admin.namespaces().createNamespace("public/default");
             admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaIntegrationTest.java

@@ -38,9 +38,7 @@
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.testcontainers.Testcontainers;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.output.WaitingConsumer;
@@ -161,29 +159,6 @@ protected void setup() throws Exception {
                 + "," + SSL_PREFIX + "127.0.0.1:" + kafkaBrokerPortTls);
         super.internalSetup();
 
-
-        if (!this.admin.clusters().getClusters().contains(this.configClusterName)) {
-            // so that clients can test short names
-            this.admin.clusters().createCluster(this.configClusterName,
-                    new ClusterData("http://127.0.0.1:" + this.brokerWebservicePort));
-        } else {
-            this.admin.clusters().updateCluster(this.configClusterName,
-                    new ClusterData("http://127.0.0.1:" + this.brokerWebservicePort));
-        }
-
-        if (!this.admin.tenants().getTenants().contains("public")) {
-            this.admin.tenants().createTenant("public",
-                    new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        } else {
-            this.admin.tenants().updateTenant("public",
-                    new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        }
-        if (!this.admin.namespaces().getNamespaces("public").contains("public/default")) {
-            this.admin.namespaces().createNamespace("public/default");
-            this.admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));
-            this.admin.namespaces().setRetention("public/default",
-                    new RetentionPolicies(60, 1000));
-        }
         if (!this.admin.namespaces().getNamespaces("public").contains("public/__kafka")) {
             this.admin.namespaces().createNamespace("public/__kafka");
             this.admin.namespaces().setNamespaceReplicationClusters("public/__kafka", Sets.newHashSet("test"));

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaMessageOrderTestBase.java

@@ -38,9 +38,7 @@
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.impl.BatchMessageIdImpl;
 import org.apache.pulsar.client.impl.TopicMessageIdImpl;
-import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
-import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.DataProvider;
@@ -69,28 +67,6 @@ protected void setup() throws Exception {
         super.internalSetup();
         log.info("success internal setup");
 
-        if (!admin.clusters().getClusters().contains(configClusterName)) {
-            // so that clients can test short names
-            admin.clusters().createCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        } else {
-            admin.clusters().updateCluster(configClusterName,
-                new ClusterData("http://127.0.0.1:" + brokerWebservicePort));
-        }
-
-        if (!admin.tenants().getTenants().contains("public")) {
-            admin.tenants().createTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        } else {
-            admin.tenants().updateTenant("public",
-                new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
-        }
-        if (!admin.namespaces().getNamespaces("public").contains("public/default")) {
-            admin.namespaces().createNamespace("public/default");
-            admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));
-            admin.namespaces().setRetention("public/default",
-                new RetentionPolicies(60, 1000));
-        }
         if (!admin.namespaces().getNamespaces("public").contains("public/__kafka")) {
             admin.namespaces().createNamespace("public/__kafka");
             admin.namespaces().setNamespaceReplicationClusters("public/__kafka", Sets.newHashSet("test"));