diff --git a/pom.xml b/pom.xml index 7df9da7152..f8b6c660f2 100644 --- a/pom.xml +++ b/pom.xml @@ -39,7 +39,7 @@ 1.18.4 2.22.0 4.1.32.Final - 2.4.0 + 2.5.0-2cc34afc0 1.7.25 3.1.8 1.11.2 @@ -196,6 +196,30 @@ test + + org.mockito + mockito-core + ${mockito.version} + test + + + + org.apache.pulsar + pulsar-broker + ${pulsar.version} + test-jar + test + + + + org.apache.pulsar + managed-ledger-original + ${pulsar.version} + test-jar + test + + + @@ -264,6 +288,11 @@ maven-surefire-plugin ${maven-surefire-plugin.version} + -Xmx2G + -Dpulsar.allocator.pooled=false + -Dpulsar.allocator.leak_detection=Advanced + -Dlog4j.configurationFile="log4j2.xml" + false 1 ${redirectTestOutputToFile} diff --git a/src/main/java/io/streamnative/kop/KafkaBrokerService.java b/src/main/java/io/streamnative/kop/KafkaBrokerService.java index fdb5ec3059..1ce89bb661 100644 --- a/src/main/java/io/streamnative/kop/KafkaBrokerService.java +++ b/src/main/java/io/streamnative/kop/KafkaBrokerService.java @@ -14,8 +14,9 @@ package io.streamnative.kop; import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.AdaptiveRecvByteBufAllocator; import io.netty.channel.ChannelOption; -import io.streamnative.kop.utils.ReflectionUtils; +import io.netty.handler.ssl.SslContext; import java.io.IOException; import java.net.InetSocketAddress; import java.util.Optional; @@ -23,6 +24,7 @@ import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.BrokerServiceUtil; import org.apache.pulsar.broker.service.DistributedIdGenerator; +import org.apache.pulsar.broker.service.PulsarChannelInitializer; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; import org.apache.pulsar.common.util.netty.EventLoopUtil; @@ -48,17 +50,18 @@ public void start() throws Exception { kafkaService.getZkClient(), "/counters/producer-name", kafkaService.getConfiguration().getClusterName()); - ReflectionUtils.setField(this, "producerNameGenerator", producerNameGenerator); + + setProducerNameGenerator(producerNameGenerator); ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT); bootstrap.group( - ReflectionUtils.getField(this, "acceptorGroup"), - ReflectionUtils.getField(this, "workerGroup")); + getAcceptorGroup(), + getWorkerGroup()); bootstrap.childOption(ChannelOption.TCP_NODELAY, true); bootstrap.channel(EventLoopUtil.getServerSocketChannelClass( - ReflectionUtils.getField(this, "workerGroup") + getWorkerGroup() )); EventLoopUtil.enableTriggeredMode(bootstrap); @@ -76,22 +79,52 @@ public void start() throws Exception { log.info("Started Kop Broker service on port {}", port.get()); } + + // start original Pulsar Broker service + ServerBootstrap pulsarBootstrap = new ServerBootstrap(); + pulsarBootstrap.childOption(ChannelOption.ALLOCATOR, PulsarByteBufAllocator.DEFAULT); + pulsarBootstrap.group( + getAcceptorGroup(), + getWorkerGroup()); + pulsarBootstrap.childOption(ChannelOption.TCP_NODELAY, true); + pulsarBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, + new AdaptiveRecvByteBufAllocator(1024, 16 * 1024, 1 * 1024 * 1024)); + + pulsarBootstrap.channel(EventLoopUtil.getServerSocketChannelClass(getWorkerGroup())); + EventLoopUtil.enableTriggeredMode(pulsarBootstrap); + + pulsarBootstrap.childHandler(new PulsarChannelInitializer(kafkaService, false)); + + Optional pulsarPort = serviceConfig.getBrokerServicePort(); + if (port.isPresent()) { + // Bind and start to accept incoming connections. + InetSocketAddress addr = new InetSocketAddress(kafkaService.getBindAddress(), pulsarPort.get()); + try { + pulsarBootstrap.bind(addr).sync(); + } catch (Exception e) { + throw new IOException("Failed to bind Pulsar broker on " + addr, e); + } + log.info("Started Pulsar Broker service on port {}", pulsarPort.get()); + } + + Optional tlsPort = serviceConfig.getBrokerServicePortTls(); + if (tlsPort.isPresent()) { + ServerBootstrap tlsBootstrap = pulsarBootstrap.clone(); + tlsBootstrap.childHandler(new PulsarChannelInitializer(kafkaService, true)); + tlsBootstrap.bind(new InetSocketAddress(kafkaService.getBindAddress(), tlsPort.get())).sync(); + log.info("Started Pulsar Broker TLS service on port {} - TLS provider: {}", tlsPort.get(), + SslContext.defaultServerProvider()); + } + // start other housekeeping functions BrokerServiceUtil.startStatsUpdater( this, serviceConfig.getStatsUpdateInitialDelayInSecs(), serviceConfig.getStatsUpdateFrequencyInSecs()); - ReflectionUtils.callNoArgVoidMethod( - this, "startInactivityMonitor" - ); - ReflectionUtils.callNoArgVoidMethod( - this, "startMessageExpiryMonitor" - ); - ReflectionUtils.callNoArgVoidMethod( - this, "startCompactionMonitor" - ); - ReflectionUtils.callNoArgVoidMethod( - this, "startBacklogQuotaChecker" - ); + + startInactivityMonitor(); + startMessageExpiryMonitor(); + startCompactionMonitor(); + startBacklogQuotaChecker(); } } diff --git a/src/main/java/io/streamnative/kop/KafkaRequestHandler.java b/src/main/java/io/streamnative/kop/KafkaRequestHandler.java index 40a05b8583..6c866b6f7c 100644 --- a/src/main/java/io/streamnative/kop/KafkaRequestHandler.java +++ b/src/main/java/io/streamnative/kop/KafkaRequestHandler.java @@ -537,7 +537,6 @@ private ByteBuf messageToByteBuf(Message message) { return buf; } - protected void handleFindCoordinatorRequest(KafkaHeaderAndRequest findCoordinator) { throw new NotImplementedException("handleFindCoordinatorRequest"); } @@ -601,7 +600,9 @@ private CompletableFuture findBroker(KafkaService kafkaServic log.debug("Find broker: {} for topicName: {}", uri, topic); } - Node node = newNode(new InetSocketAddress(uri.getHost(), uri.getPort())); + Node node = newNode(new InetSocketAddress( + uri.getHost(), + kafkaService.getKafkaConfig().getKafkaServicePort().get())); resultFuture.complete(newPartitionMetadata(topic, node)); return; } else { diff --git a/src/main/java/io/streamnative/kop/KafkaService.java b/src/main/java/io/streamnative/kop/KafkaService.java index 011e8d7749..58859c0b33 100644 --- a/src/main/java/io/streamnative/kop/KafkaService.java +++ b/src/main/java/io/streamnative/kop/KafkaService.java @@ -14,7 +14,6 @@ package io.streamnative.kop; import com.google.common.collect.Maps; -import io.streamnative.kop.utils.ReflectionUtils; import java.util.Map; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; @@ -26,7 +25,6 @@ import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.loadbalance.LoadManager; -import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.schema.SchemaRegistryService; import org.apache.pulsar.broker.stats.MetricsGenerator; import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsServlet; @@ -51,7 +49,7 @@ public KafkaService(KafkaServiceConfiguration config) { @Override public void start() throws PulsarServerException { - ReentrantLock lock = ReflectionUtils.getField(this, "mutex"); + ReentrantLock lock = getMutex(); lock.lock(); @@ -77,62 +75,33 @@ public void start() throws PulsarServerException { new LocalZooKeeperConnectionService(getZooKeeperClientFactory(), kafkaConfig.getZookeeperServers(), kafkaConfig.getZooKeeperSessionTimeoutMillis()); - ReflectionUtils.setField( - this, - "localZooKeeperConnectionProvider", - localZooKeeperConnectionService - ); + setLocalZooKeeperConnectionProvider(localZooKeeperConnectionService); localZooKeeperConnectionService.start(getShutdownService()); - // Initialize and start service to access configuration repository. - ReflectionUtils.callNoArgVoidMethod( - this, - "startZkCacheService" - ); + startZkCacheService(); BookKeeperClientFactory bkClientFactory = newBookKeeperClientFactory(); - ReflectionUtils.setField( - this, - "bkClientFactory", - bkClientFactory - ); - ReflectionUtils.setField( - this, - "managedLedgerClientFactory", - new ManagedLedgerClientFactory(kafkaConfig, getZkClient(), bkClientFactory) - ); - ReflectionUtils.setField( - this, - "brokerService", - new KafkaBrokerService(this) - ); + setBkClientFactory(bkClientFactory); + setManagedLedgerClientFactory( + new ManagedLedgerClientFactory(kafkaConfig, getZkClient(), bkClientFactory)); + setBrokerService(new KafkaBrokerService(this)); // Start load management service (even if load balancing is disabled) getLoadManager().set(LoadManager.create(this)); // Start the leader election service - ReflectionUtils.callNoArgVoidMethod( - this, - "startLeaderElectionService" - ); + startLeaderElectionService(); // needs load management service - ReflectionUtils.callNoArgVoidMethod( - this, - "startNamespaceService" - ); + startNamespaceService(); - ReflectionUtils.setField( - this, - "offloader", - createManagedLedgerOffloader(kafkaConfig) - ); + setOffloader(createManagedLedgerOffloader(kafkaConfig)); getBrokerService().start(); WebService webService = new WebService(this); - ReflectionUtils.setField(this, "webService", webService); + setWebService(webService); Map attributeMap = Maps.newHashMap(); attributeMap.put(WebService.ATTRIBUTE_PULSAR_NAME, this); Map vipAttributeMap = Maps.newHashMap(); @@ -171,36 +140,22 @@ public Boolean get() { webService.addStaticResources("/static", "/static"); // Register heartbeat and bootstrap namespaces. - ReflectionUtils.getField( - this, "nsService" - ).registerBootstrapNamespaces(); + getNsService().registerBootstrapNamespaces(); - ReflectionUtils.setField( - this, - "schemaRegistryService", - SchemaRegistryService.create(this) - ); + setSchemaRegistryService(SchemaRegistryService.create(this)); webService.start(); - ReflectionUtils.setField( - this, - "metricsGenerator", - new MetricsGenerator(this) - ); + setMetricsGenerator(new MetricsGenerator(this)); // By starting the Load manager service, the broker will also become visible // to the rest of the broker by creating the registration z-node. This needs // to be done only when the broker is fully operative. - ReflectionUtils.callNoArgVoidMethod( - this, - "startLoadManagementService"); + startLoadManagementService(); - reflectSetState(State.Started); + setState(State.Started); - ReflectionUtils.callNoArgVoidMethod( - this, - "acquireSLANamespace"); + acquireSLANamespace(); final String bootstrapMessage = "bootstrap service " + (kafkaConfig.getWebServicePort().isPresent() @@ -221,17 +176,4 @@ public Boolean get() { lock.unlock(); } } - - protected void reflectSetState(State state) { - try { - ReflectionUtils.setField( - this, - "state", - state - ); - } catch (IllegalAccessException | NoSuchFieldException e) { - throw new RuntimeException("Unable to set broker set to " + state, e); - } - } - } diff --git a/src/test/java/io/streamnative/kop/KafkaRequestHandlerTest.java b/src/test/java/io/streamnative/kop/KafkaRequestHandlerTest.java index d602a065e1..25d77faf5a 100644 --- a/src/test/java/io/streamnative/kop/KafkaRequestHandlerTest.java +++ b/src/test/java/io/streamnative/kop/KafkaRequestHandlerTest.java @@ -15,7 +15,7 @@ import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX; -import static org.mockito.Matchers.anyObject; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -155,7 +155,7 @@ public void testChannelRead() throws Exception { handler.channelActive(ctx); handler.channelRead(mock(ChannelHandlerContext.class), inputBuf); - verify(handler, times(1)).handleApiVersionsRequest(anyObject()); + verify(handler, times(1)).handleApiVersionsRequest(any()); } @Test diff --git a/src/test/java/io/streamnative/kop/KafkaRequestTypeTest.java b/src/test/java/io/streamnative/kop/KafkaRequestTypeTest.java new file mode 100644 index 0000000000..1f6d77b90f --- /dev/null +++ b/src/test/java/io/streamnative/kop/KafkaRequestTypeTest.java @@ -0,0 +1,106 @@ +/** + * Licensed 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 io.streamnative.kop; + + +import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; + +import com.google.common.collect.Sets; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.TenantInfo; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * Unit test for Different kafka request type. + */ +@Slf4j +public class KafkaRequestTypeTest extends MockKafkaServiceBaseTest { + + @BeforeMethod + @Override + protected void setup() throws Exception { + super.internalSetup(); + // so that clients can test short names + admin.clusters().createCluster("test", new ClusterData("http://127.0.0.1:" + brokerWebservicePort)); + + admin.tenants().createTenant("public", + new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test"))); + admin.namespaces().createNamespace("public/default"); + admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test")); + } + + @AfterMethod + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + + @Test(timeOut = 20000) + public void testProduceRequest() throws Exception { + String topicName = "kopTopicProduce"; + + // create partitioned topic. + kafkaService.getAdminClient().topics().createPartitionedTopic(topicName, 1); + + Consumer consumer = pulsarClient.newConsumer() + .topic("persistent://public/default/" + topicName + PARTITIONED_TOPIC_SUFFIX + 0) + .subscriptionName("test_producer_sub").subscribe(); + + + // 1. produce message with Kafka producer. + Producer producer = new Producer(topicName, false); + + int messageNo = 0; + int totalMsgs = 10; + + while (messageNo < totalMsgs) { + String messageStr = "Message_Kop_" + messageNo; + + try { + producer.getProducer().send(new ProducerRecord<>(topicName, + messageNo, + messageStr)).get(); + log.info("Sent message: (" + messageNo + ", " + messageStr + ")"); + } catch (Exception e) { + e.printStackTrace(); + } + ++messageNo; + } + + assertEquals(totalMsgs, messageNo); + + Message msg = null; + + // 2. Consume messages use Pulsar client Consumer. + for (int i = 0; i < totalMsgs; i++) { + msg = consumer.receive(100, TimeUnit.MILLISECONDS); + log.info("Pulsar consumer get message: {}", new String(msg.getData())); + consumer.acknowledge(msg); + } + + // verify have received all messages + msg = consumer.receive(100, TimeUnit.MILLISECONDS); + assertNull(msg); + } +} diff --git a/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java b/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java new file mode 100644 index 0000000000..8172f27b1c --- /dev/null +++ b/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java @@ -0,0 +1,359 @@ +/** + * Licensed 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 io.streamnative.kop; + +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.spy; + +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.lang.reflect.Field; +import java.net.URI; +import java.net.URL; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.function.Supplier; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.client.EnsemblePlacementPolicy; +import org.apache.bookkeeper.client.PulsarMockBookKeeper; +import org.apache.bookkeeper.test.PortManager; +import org.apache.bookkeeper.util.ZkUtils; +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.pulsar.broker.BookKeeperClientFactory; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.auth.SameThreadOrderedSafeExecutor; +import org.apache.pulsar.broker.namespace.NamespaceService; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.compaction.Compactor; +import org.apache.pulsar.zookeeper.ZooKeeperClientFactory; +import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.MockZooKeeper; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.ACL; + +/** + * A test base to start a KafkaService. + */ +@Slf4j +public abstract class MockKafkaServiceBaseTest { + + protected KafkaServiceConfiguration conf; + protected KafkaService kafkaService; + protected PulsarAdmin admin; + protected URL brokerUrl; + protected URL brokerUrlTls; + protected URI lookupUrl; + protected PulsarClient pulsarClient; + + protected final int brokerWebservicePort = PortManager.nextFreePort(); + protected final int brokerWebservicePortTls = PortManager.nextFreePort(); + protected final int brokerPort = PortManager.nextFreePort(); + protected final int kafkaBrokerPort = PortManager.nextFreePort(); + + + protected MockZooKeeper mockZookKeeper; + protected NonClosableMockBookKeeper mockBookKeeper; + protected boolean isTcpLookup = false; + protected final String configClusterName = "test"; + + private SameThreadOrderedSafeExecutor sameThreadOrderedSafeExecutor; + private ExecutorService bkExecutor; + + public MockKafkaServiceBaseTest() { + resetConfig(); + } + + protected void resetConfig() { + this.conf = new KafkaServiceConfiguration(); + this.conf.setKafkaServicePort(Optional.ofNullable(kafkaBrokerPort)); + this.conf.setBrokerServicePort(Optional.ofNullable(brokerPort)); + this.conf.setAdvertisedAddress("localhost"); + this.conf.setWebServicePort(Optional.ofNullable(brokerWebservicePort)); + this.conf.setClusterName(configClusterName); + this.conf.setAdvertisedAddress("localhost"); + this.conf.setManagedLedgerCacheSizeMB(8); + this.conf.setActiveConsumerFailoverDelayTimeMillis(0); + this.conf.setDefaultNumberOfNamespaceBundles(1); + this.conf.setZookeeperServers("localhost:2181"); + this.conf.setConfigurationStoreServers("localhost:3181"); + } + + protected final void internalSetup() throws Exception { + init(); + lookupUrl = new URI(brokerUrl.toString()); + if (isTcpLookup) { + lookupUrl = new URI("broker://localhost:" + brokerPort); + } + pulsarClient = newPulsarClient(lookupUrl.toString(), 0); + } + + protected PulsarClient newPulsarClient(String url, int intervalInSecs) throws PulsarClientException { + return PulsarClient.builder().serviceUrl(url).statsInterval(intervalInSecs, TimeUnit.SECONDS).build(); + } + + protected final void init() throws Exception { + sameThreadOrderedSafeExecutor = new SameThreadOrderedSafeExecutor(); + bkExecutor = Executors.newSingleThreadExecutor( + new ThreadFactoryBuilder().setNameFormat("mock-kafkaService-bk") + .setUncaughtExceptionHandler((thread, ex) -> log.info("Uncaught exception", ex)) + .build()); + + mockZookKeeper = createMockZooKeeper(); + mockBookKeeper = createMockBookKeeper(mockZookKeeper, bkExecutor); + + startBroker(); + + brokerUrl = new URL("http://" + kafkaService.getAdvertisedAddress() + ":" + brokerWebservicePort); + brokerUrlTls = new URL("https://" + kafkaService.getAdvertisedAddress() + ":" + brokerWebservicePortTls); + + admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).build()); + } + + protected final void internalCleanup() throws Exception { + try { + // if init fails, some of these could be null, and if so would throw + // an NPE in shutdown, obscuring the real error + if (admin != null) { + admin.close(); + } + if (pulsarClient != null) { + pulsarClient.close(); + } + if (kafkaService != null) { + kafkaService.close(); + } + if (mockBookKeeper != null) { + mockBookKeeper.reallyShutdown(); + } + if (mockZookKeeper != null) { + mockZookKeeper.shutdown(); + } + if (sameThreadOrderedSafeExecutor != null) { + sameThreadOrderedSafeExecutor.shutdown(); + } + if (bkExecutor != null) { + bkExecutor.shutdown(); + } + } catch (Exception e) { + log.warn("Failed to clean up mocked kafkaService service:", e); + throw e; + } + } + + protected abstract void setup() throws Exception; + + protected abstract void cleanup() throws Exception; + + protected void restartBroker() throws Exception { + stopBroker(); + startBroker(); + } + + protected void stopBroker() throws Exception { + kafkaService.close(); + } + + protected void startBroker() throws Exception { + this.kafkaService = startBroker(conf); + } + + protected KafkaService startBroker(KafkaServiceConfiguration conf) throws Exception { + KafkaService kafkaService = spy(new KafkaService(conf)); + + setupBrokerMocks(kafkaService); + boolean isAuthorizationEnabled = conf.isAuthorizationEnabled(); + // enable authorization to initialize authorization service which is used by grant-permission + conf.setAuthorizationEnabled(true); + kafkaService.start(); + conf.setAuthorizationEnabled(isAuthorizationEnabled); + + Compactor spiedCompactor = spy(kafkaService.getCompactor()); + doReturn(spiedCompactor).when(kafkaService).getCompactor(); + + return kafkaService; + } + + protected void setupBrokerMocks(KafkaService kafkaService) throws Exception { + // Override default providers with mocked ones + doReturn(mockZooKeeperClientFactory).when(kafkaService).getZooKeeperClientFactory(); + doReturn(mockBookKeeperClientFactory).when(kafkaService).newBookKeeperClientFactory(); + + Supplier namespaceServiceSupplier = () -> spy(new NamespaceService(kafkaService)); + doReturn(namespaceServiceSupplier).when(kafkaService).getNamespaceServiceProvider(); + + doReturn(sameThreadOrderedSafeExecutor).when(kafkaService).getOrderedExecutor(); + } + + public static MockZooKeeper createMockZooKeeper() throws Exception { + MockZooKeeper zk = MockZooKeeper.newInstance(MoreExecutors.newDirectExecutorService()); + List dummyAclList = new ArrayList<>(0); + + ZkUtils.createFullPathOptimistic(zk, "/ledgers/available/192.168.1.1:" + 5000, + "".getBytes(ZookeeperClientFactoryImpl.ENCODING_SCHEME), dummyAclList, CreateMode.PERSISTENT); + + zk.create( + "/ledgers/LAYOUT", + "1\nflat:1".getBytes(ZookeeperClientFactoryImpl.ENCODING_SCHEME), dummyAclList, + CreateMode.PERSISTENT); + return zk; + } + + public static NonClosableMockBookKeeper createMockBookKeeper(ZooKeeper zookeeper, + ExecutorService executor) throws Exception { + return spy(new NonClosableMockBookKeeper(zookeeper, executor)); + } + + /** + * Prevent the MockBookKeeper instance from being closed when the broker is restarted within a test. + */ + public static class NonClosableMockBookKeeper extends PulsarMockBookKeeper { + + public NonClosableMockBookKeeper(ZooKeeper zk, ExecutorService executor) throws Exception { + super(zk, executor); + } + + @Override + public void close() { + // no-op + } + + @Override + public void shutdown() { + // no-op + } + + public void reallyShutdown() { + super.shutdown(); + } + } + + protected ZooKeeperClientFactory mockZooKeeperClientFactory = new ZooKeeperClientFactory() { + + @Override + public CompletableFuture create(String serverList, SessionType sessionType, + int zkSessionTimeoutMillis) { + // Always return the same instance + // (so that we don't loose the mock ZK content on broker restart + return CompletableFuture.completedFuture(mockZookKeeper); + } + }; + + private BookKeeperClientFactory mockBookKeeperClientFactory = new BookKeeperClientFactory() { + + @Override + public BookKeeper create(ServiceConfiguration conf, ZooKeeper zkClient, + Optional> ensemblePlacementPolicyClass, + Map properties) { + // Always return the same instance (so that we don't loose the mock BK content on broker restart + return mockBookKeeper; + } + + @Override + public void close() { + // no-op + } + }; + + public static void retryStrategically(Predicate predicate, int retryCount, long intSleepTimeInMillis) + throws Exception { + for (int i = 0; i < retryCount; i++) { + if (predicate.test(null) || i == (retryCount - 1)) { + break; + } + Thread.sleep(intSleepTimeInMillis + (intSleepTimeInMillis * i)); + } + } + + public static void setFieldValue(Class clazz, Object classObj, String fieldName, Object fieldValue) + throws Exception { + Field field = clazz.getDeclaredField(fieldName); + field.setAccessible(true); + field.set(classObj, fieldValue); + } + + /** + * A producer wrapper. + */ + @Getter + public class Producer { + private final KafkaProducer producer; + private final String topic; + private final Boolean isAsync; + + public Producer(String topic, Boolean isAsync) { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost" + ":" + kafkaBrokerPort); + props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer"); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producer = new KafkaProducer<>(props); + this.topic = topic; + this.isAsync = isAsync; + } + } + + /** + * A callback wrapper for produce async. + */ + class DemoCallBack implements Callback { + + private final long startTime; + private final int key; + private final String message; + + public DemoCallBack(long startTime, int key, String message) { + this.startTime = startTime; + this.key = key; + this.message = message; + } + + /** + * A callback method the user can implement to provide asynchronous handling of request completion. + * This method will be called when the record sent to the server has been acknowledged. + * Exactly one of the arguments will be non-null. + * + * @param metadata The metadata for the record that was sent (i.e. the partition and offset). Null if an error + * occurred. + * @param exception The exception thrown during processing of this record. Null if no error occurred. + */ + public void onCompletion(RecordMetadata metadata, Exception exception) { + long elapsedTime = System.currentTimeMillis() - startTime; + if (metadata != null) { + System.out.println( + "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() + + "), " + "offset(" + metadata.offset() + ") in " + elapsedTime + " ms"); + } else { + exception.printStackTrace(); + } + } + } +} diff --git a/src/test/resources/log4j2.xml b/src/test/resources/log4j2.xml new file mode 100644 index 0000000000..8e79ebf8d8 --- /dev/null +++ b/src/test/resources/log4j2.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + + + +