You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/07/05 14:32:52 UTC
[flink] 03/03: [FLINK-27199][Connector/Pulsar] Move mocked Pulsar runtime to new local memory based metastore. Drop embedded Pulsar runtime. Add a bootstrap script for Pulsar docker runtime.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f629b3ece44c3a3a011ed96f08bd9f1870363bcf
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Jul 5 11:10:48 2022 +0800
[FLINK-27199][Connector/Pulsar] Move mocked Pulsar runtime to new local memory based metastore. Drop embedded Pulsar runtime. Add a bootstrap script for Pulsar docker runtime.
---
.../runtime/mock/MockBookKeeperClientFactory.java | 6 +-
.../testutils/runtime/mock/MockPulsarService.java | 30 ---------
.../runtime/mock/MockZooKeeperClientFactory.java | 73 ----------------------
.../testutils/runtime/mock/PulsarMockRuntime.java | 6 +-
4 files changed, 7 insertions(+), 108 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java
index 41fad547a99..d35bd669ab0 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java
@@ -25,7 +25,7 @@ import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.pulsar.broker.BookKeeperClientFactory;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.zookeeper.ZooKeeper;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import java.io.IOException;
import java.util.Map;
@@ -42,7 +42,7 @@ public class MockBookKeeperClientFactory implements BookKeeperClientFactory {
@Override
public BookKeeper create(
ServiceConfiguration conf,
- ZooKeeper zkClient,
+ MetadataStoreExtended store,
EventLoopGroup eventLoopGroup,
Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
Map<String, Object> ensemblePlacementPolicyProperties)
@@ -53,7 +53,7 @@ public class MockBookKeeperClientFactory implements BookKeeperClientFactory {
@Override
public BookKeeper create(
ServiceConfiguration conf,
- ZooKeeper zkClient,
+ MetadataStoreExtended store,
EventLoopGroup eventLoopGroup,
Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
Map<String, Object> ensemblePlacementPolicyProperties,
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java
index 6b6c4120afa..09015a7db86 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java
@@ -24,49 +24,23 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.namespace.NamespaceService;
-import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
-import org.apache.pulsar.metadata.impl.ZKMetadataStore;
-import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
-import org.apache.zookeeper.MockZooKeeperSession;
import java.util.function.Supplier;
/** A Mock pulsar service which would use the mocked zookeeper and bookkeeper. */
public class MockPulsarService extends PulsarService {
- private final int brokerServicePort;
-
- private final MockZooKeeperClientFactory zooKeeperClientFactory =
- new MockZooKeeperClientFactory();
-
- private final MockZooKeeperSession zooKeeperSession =
- MockZooKeeperSession.newInstance(zooKeeperClientFactory.getZooKeeper());
-
private final SameThreadOrderedSafeExecutor orderedExecutor =
new SameThreadOrderedSafeExecutor();
public MockPulsarService(ServiceConfiguration config) {
super(config);
- this.brokerServicePort =
- config.getBrokerServicePort().orElseThrow(IllegalArgumentException::new);
- }
-
- public ZooKeeperClientFactory getZooKeeperClientFactory() {
- return zooKeeperClientFactory;
}
public BookKeeperClientFactory newBookKeeperClientFactory() {
return new MockBookKeeperClientFactory();
}
- public MetadataStoreExtended createLocalMetadataStore() {
- return new ZKMetadataStore(zooKeeperSession);
- }
-
- public MetadataStoreExtended createConfigurationMetadataStore() {
- return new ZKMetadataStore(zooKeeperSession);
- }
-
public Supplier<NamespaceService> getNamespaceServiceProvider() {
return () -> new NamespaceService(this);
}
@@ -80,8 +54,4 @@ public class MockPulsarService extends PulsarService {
public BrokerInterceptor getBrokerInterceptor() {
return new BlankBrokerInterceptor();
}
-
- public int getBrokerServicePort() {
- return brokerServicePort;
- }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java
deleted file mode 100644
index 3c89484ad37..00000000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java
+++ /dev/null
@@ -1,73 +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.
- */
-
-package org.apache.flink.connector.pulsar.testutils.runtime.mock;
-
-import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.MoreExecutors;
-
-import org.apache.bookkeeper.util.ZkUtils;
-import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.MockZooKeeper;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.ACL;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
-import static org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl.ENCODING_SCHEME;
-import static org.apache.zookeeper.CreateMode.PERSISTENT;
-
-/** A ZooKeeperClientFactory implementation which returns mocked zookeeper instead of normal zk. */
-public class MockZooKeeperClientFactory implements ZooKeeperClientFactory {
-
- private final MockZooKeeper zooKeeper;
-
- public MockZooKeeperClientFactory() {
- this.zooKeeper = MockZooKeeper.newInstance(MoreExecutors.newDirectExecutorService());
- List<ACL> dummyAclList = new ArrayList<>(0);
-
- try {
- ZkUtils.createFullPathOptimistic(
- zooKeeper,
- "/ledgers/available/192.168.1.1:" + 5000,
- "".getBytes(ENCODING_SCHEME),
- dummyAclList,
- PERSISTENT);
-
- zooKeeper.create(
- "/ledgers/LAYOUT",
- "1\nflat:1".getBytes(ENCODING_SCHEME),
- dummyAclList,
- PERSISTENT);
- } catch (KeeperException | InterruptedException e) {
- throw new IllegalStateException(e);
- }
- }
-
- @Override
- public CompletableFuture<ZooKeeper> create(
- String serverList, SessionType sessionType, int zkSessionTimeoutMillis) {
- return CompletableFuture.completedFuture(zooKeeper);
- }
-
- MockZooKeeper getZooKeeper() {
- return zooKeeper;
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java
index e6ff060455a..a86ff5283f5 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockRuntime.java
@@ -87,8 +87,8 @@ public class PulsarMockRuntime implements PulsarRuntime {
configuration.setActiveConsumerFailoverDelayTimeMillis(0);
configuration.setDefaultRetentionTimeInMinutes(7);
configuration.setDefaultNumberOfNamespaceBundles(1);
- configuration.setZookeeperServers("localhost:2181");
- configuration.setConfigurationStoreServers("localhost:3181");
+ configuration.setMetadataStoreUrl("memory:local");
+ configuration.setConfigurationMetadataStoreUrl("memory:local");
configuration.setAuthenticationEnabled(false);
configuration.setAuthorizationEnabled(false);
@@ -101,6 +101,8 @@ public class PulsarMockRuntime implements PulsarRuntime {
configuration.setWebServicePort(Optional.of(0));
// Enable transactions.
+ configuration.setSystemTopicEnabled(true);
+ configuration.setBrokerDeduplicationEnabled(true);
configuration.setTransactionCoordinatorEnabled(true);
configuration.setTransactionMetadataStoreProviderClassName(
"org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider");