You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/08/12 03:23:32 UTC
[GitHub] [inlong] vernedeng opened a new pull request, #5512: [INLONG-5496][SortSdk] Refactor single topic manager
vernedeng opened a new pull request, #5512:
URL: https://github.com/apache/inlong/pull/5512
### Prepare a Pull Request
*(Change the title refer to the following example)*
- Title Example: [INLONG-XYZ][Component] Title of the pull request
*(The following *XYZ* should be replaced by the actual [GitHub Issue](https://github.com/apache/inlong/issues) number)*
- Fixes #5496
### Motivation
*Explain here the context, and why you're making that change. What is the problem you're trying to solve?*
### Modifications
*Describe the modifications you've done.*
### Verifying this change
*(Please pick either of the following options)*
- [ ] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [ ] This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SortSdk] Refactor single topic manager
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944130799
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java:
##########
@@ -311,6 +328,13 @@ public enum ConsumeStrategy {
lastest_absolutely
}
+ public enum TopicManagerType {
+ // single topic manager
+ single_topic,
Review Comment:
The enum fields should use uppercase letters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r945850673
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManagerFactory.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig.TopicManagerType;
+
+/**
+ * Inlong topic manager factory.
+ * To create single or multi topic fetcher manager according to the {@link TopicManagerType}
+ */
+public class InlongTopicManagerFactory {
+
+ public static InlongTopicManager createInLongTopicManager(
Review Comment:
there are too many naming problem, I will make a new issue to fix these.
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java:
##########
@@ -0,0 +1,495 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.kafka.InLongKafkaFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.pulsar.InLongPulsarFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.InLongTubeFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.TubeConsumerCreater;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Inlong manager that maintain the single topic fetchers.
+ * It is suitable to the cases that each topic has its own configurations.
+ * And each consumer only consume the very one topic.
+ */
+public class InlongSingleTopicManager extends InlongTopicManager {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(InlongSingleTopicManager.class);
+
+ private final ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, PulsarClient> pulsarClients = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, TubeConsumerCreater> tubeFactories = new ConcurrentHashMap<>();
+
+ private final PeriodicTask updateMetaDataWorker;
+ private volatile List<String> toBeSelectFetchers = new ArrayList<>();
+ private boolean stopAssign = false;
+
+ public InlongSingleTopicManager(ClientContext context, QueryConsumeConfig queryConsumeConfig) {
+ super(context, queryConsumeConfig);
+ updateMetaDataWorker = new UpdateMetaDataThread(context.getConfig().getUpdateMetaDataIntervalSec(),
+ TimeUnit.SECONDS);
+ String threadName = "sortsdk_inlongtopic_manager_" + context.getConfig().getSortTaskId()
+ + "_" + StringUtil.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss");
+ updateMetaDataWorker.start(threadName);
+ }
+
+ private void updateToBeSelectFetchers(Collection<String> c) {
+ toBeSelectFetchers = new ArrayList<>(c);
+ }
+
+ private boolean initFetcher(InLongTopicFetcher fetcher, InLongTopic inLongTopic) {
+ if (InlongTopicTypeEnum.PULSAR.getName().equalsIgnoreCase(inLongTopic.getTopicType())) {
+ LOGGER.info("create fetcher topic is pulsar {}", inLongTopic);
+ return fetcher.init(pulsarClients.get(inLongTopic.getInLongCluster().getClusterId()));
+ } else if (InlongTopicTypeEnum.KAFKA.getName().equalsIgnoreCase(inLongTopic.getTopicType())) {
+ LOGGER.info("create fetcher topic is kafka {}", inLongTopic);
+ return fetcher.init(inLongTopic.getInLongCluster().getBootstraps());
+ } else if (InlongTopicTypeEnum.TUBE.getName().equalsIgnoreCase(inLongTopic.getTopicType())) {
+ LOGGER.info("create fetcher topic is tube {}", inLongTopic);
+ return fetcher.init(tubeFactories.get(inLongTopic.getInLongCluster().getClusterId()));
+ } else {
+ LOGGER.error("create fetcher topic type not support " + inLongTopic.getTopicType());
+ return false;
+ }
+ }
+
+ @Override
+ public InLongTopicFetcher addFetcher(InLongTopic inLongTopic) {
+
+ try {
+ InLongTopicFetcher result = fetchers.get(inLongTopic.getTopicKey());
+ if (result == null) {
+ // create fetcher (pulsar,tube,kafka)
+ InLongTopicFetcher inLongTopicFetcher = createInLongTopicFetcher(inLongTopic);
+ InLongTopicFetcher preValue = fetchers.putIfAbsent(inLongTopic.getTopicKey(), inLongTopicFetcher);
+ LOGGER.info("addFetcher :{}", inLongTopic.getTopicKey());
+ if (preValue != null) {
+ result = preValue;
+ if (inLongTopicFetcher != null) {
+ inLongTopicFetcher.close();
+ }
+ LOGGER.info("addFetcher create same fetcher {}", inLongTopic);
+ } else {
+ result = inLongTopicFetcher;
+ if (result != null
+ && !initFetcher(result, inLongTopic)) {
Review Comment:
fixed, thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944170852
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java:
##########
@@ -311,6 +328,13 @@ public enum ConsumeStrategy {
lastest_absolutely
}
+ public enum TopicManagerType {
+ // single topic manager
+ single_topic,
Review Comment:
fixed, thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944313078
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManagerFactory.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig.TopicManagerType;
+
+/**
+ * Inlong topic manager factory.
+ * To create single or multi topic fetcher manager according to the {@link TopicManagerType}
+ */
+public class InlongTopicManagerFactory {
+
+ public static InlongTopicManager createInLongTopicManager(
Review Comment:
`createInLongTopicManager` -> `createInlongTopicManager`.
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java:
##########
@@ -39,7 +40,7 @@ public class SortClientImpl extends SortClient {
private final ClientContext context;
- private final InLongTopicManager inLongTopicManager;
+ private final InlongTopicManager inLongTopicManager;
Review Comment:
`inLongTopicManager` -> `inlongTopicManager`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944312637
##########
inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManagerTest.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig;
+import org.apache.inlong.sdk.sort.entity.CacheZoneCluster;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.ClientContextImpl;
+import org.apache.inlong.sdk.sort.impl.InlongTopicManagerImpl;
+import org.apache.inlong.sdk.sort.impl.QueryConsumeConfigImpl;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClientContext.class})
+public class InlongSingleTopicManagerTest {
+
+ private InLongTopic inLongTopic;
+ private ClientContext clientContext;
+ private QueryConsumeConfig queryConsumeConfig;
+ private InlongTopicManager inLongTopicManager;
Review Comment:
`inLongTopicManager` -> `inlongTopicManager`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944317301
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManager.java:
##########
@@ -0,0 +1,495 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.kafka.InLongKafkaFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.pulsar.InLongPulsarFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.InLongTubeFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.TubeConsumerCreater;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Inlong manager that maintain the single topic fetchers.
+ * It is suitable to the cases that each topic has its own configurations.
+ * And each consumer only consume the very one topic.
+ */
+public class InlongSingleTopicManager extends InlongTopicManager {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(InlongSingleTopicManager.class);
+
+ private final ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, PulsarClient> pulsarClients = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, TubeConsumerCreater> tubeFactories = new ConcurrentHashMap<>();
+
+ private final PeriodicTask updateMetaDataWorker;
+ private volatile List<String> toBeSelectFetchers = new ArrayList<>();
+ private boolean stopAssign = false;
+
+ public InlongSingleTopicManager(ClientContext context, QueryConsumeConfig queryConsumeConfig) {
+ super(context, queryConsumeConfig);
+ updateMetaDataWorker = new UpdateMetaDataThread(context.getConfig().getUpdateMetaDataIntervalSec(),
+ TimeUnit.SECONDS);
+ String threadName = "sortsdk_inlongtopic_manager_" + context.getConfig().getSortTaskId()
+ + "_" + StringUtil.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss");
+ updateMetaDataWorker.start(threadName);
+ }
+
+ private void updateToBeSelectFetchers(Collection<String> c) {
+ toBeSelectFetchers = new ArrayList<>(c);
+ }
+
+ private boolean initFetcher(InLongTopicFetcher fetcher, InLongTopic inLongTopic) {
+ if (InlongTopicTypeEnum.PULSAR.getName().equalsIgnoreCase(inLongTopic.getTopicType())) {
+ LOGGER.info("create fetcher topic is pulsar {}", inLongTopic);
+ return fetcher.init(pulsarClients.get(inLongTopic.getInLongCluster().getClusterId()));
+ } else if (InlongTopicTypeEnum.KAFKA.getName().equalsIgnoreCase(inLongTopic.getTopicType())) {
+ LOGGER.info("create fetcher topic is kafka {}", inLongTopic);
+ return fetcher.init(inLongTopic.getInLongCluster().getBootstraps());
+ } else if (InlongTopicTypeEnum.TUBE.getName().equalsIgnoreCase(inLongTopic.getTopicType())) {
+ LOGGER.info("create fetcher topic is tube {}", inLongTopic);
+ return fetcher.init(tubeFactories.get(inLongTopic.getInLongCluster().getClusterId()));
+ } else {
+ LOGGER.error("create fetcher topic type not support " + inLongTopic.getTopicType());
+ return false;
+ }
+ }
+
+ @Override
+ public InLongTopicFetcher addFetcher(InLongTopic inLongTopic) {
+
+ try {
+ InLongTopicFetcher result = fetchers.get(inLongTopic.getTopicKey());
+ if (result == null) {
+ // create fetcher (pulsar,tube,kafka)
+ InLongTopicFetcher inLongTopicFetcher = createInLongTopicFetcher(inLongTopic);
+ InLongTopicFetcher preValue = fetchers.putIfAbsent(inLongTopic.getTopicKey(), inLongTopicFetcher);
+ LOGGER.info("addFetcher :{}", inLongTopic.getTopicKey());
+ if (preValue != null) {
+ result = preValue;
+ if (inLongTopicFetcher != null) {
+ inLongTopicFetcher.close();
+ }
+ LOGGER.info("addFetcher create same fetcher {}", inLongTopic);
+ } else {
+ result = inLongTopicFetcher;
+ if (result != null
+ && !initFetcher(result, inLongTopic)) {
Review Comment:
Those two lines can be merged into one line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944313078
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InlongTopicManagerFactory.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig.TopicManagerType;
+
+/**
+ * Inlong topic manager factory.
+ * To create single or multi topic fetcher manager according to the {@link TopicManagerType}
+ */
+public class InlongTopicManagerFactory {
+
+ public static InlongTopicManager createInLongTopicManager(
Review Comment:
`inLongTopicManager` -> `inlongTopicManager`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r945851416
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/SortClientImpl.java:
##########
@@ -39,7 +40,7 @@ public class SortClientImpl extends SortClient {
private final ClientContext context;
- private final InLongTopicManager inLongTopicManager;
+ private final InlongTopicManager inLongTopicManager;
Review Comment:
there are too many naming problem, I will make a new issue to fix these.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944170203
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InLongSingleTopicManager.java:
##########
@@ -0,0 +1,490 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InLongTopicManager;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.kafka.InLongKafkaFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.pulsar.InLongPulsarFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.InLongTubeFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.TubeConsumerCreater;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InLongSingleTopicManager extends InLongTopicManager {
+
+ private final Logger logger = LoggerFactory.getLogger(InLongSingleTopicManager.class);
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
dockerzhang merged PR #5512:
URL: https://github.com/apache/inlong/pull/5512
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r945849755
##########
inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManagerTest.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig;
+import org.apache.inlong.sdk.sort.entity.CacheZoneCluster;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.ClientContextImpl;
+import org.apache.inlong.sdk.sort.impl.InlongTopicManagerImpl;
+import org.apache.inlong.sdk.sort.impl.QueryConsumeConfigImpl;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClientContext.class})
+public class InlongSingleTopicManagerTest {
+
+ private InLongTopic inLongTopic;
+ private ClientContext clientContext;
+ private QueryConsumeConfig queryConsumeConfig;
+ private InlongTopicManager inLongTopicManager;
Review Comment:
there are too many naming problem, I will make a new issue to fix these.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944315040
##########
inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManagerTest.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig;
+import org.apache.inlong.sdk.sort.entity.CacheZoneCluster;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.ClientContextImpl;
+import org.apache.inlong.sdk.sort.impl.InlongTopicManagerImpl;
+import org.apache.inlong.sdk.sort.impl.QueryConsumeConfigImpl;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClientContext.class})
+public class InlongSingleTopicManagerTest {
+
+ private InLongTopic inLongTopic;
+ private ClientContext clientContext;
+ private QueryConsumeConfig queryConsumeConfig;
+ private InlongTopicManager inLongTopicManager;
+
+ {
+ System.setProperty("log4j2.disable.jmx", Boolean.TRUE.toString());
+
+ inLongTopic = new InLongTopic();
+ inLongTopic.setTopic("testTopic");
+ inLongTopic.setPartitionId(0);
+ inLongTopic.setTopicType("pulsar");
+ inLongTopic.setProperties(new HashMap<>());
+
+ CacheZoneCluster cacheZoneCluster = new CacheZoneCluster("clusterId", "bootstraps", "token");
+ inLongTopic.setInLongCluster(cacheZoneCluster);
+
+ clientContext = PowerMockito.mock(ClientContextImpl.class);
+
+ SortClientConfig sortClientConfig = PowerMockito.mock(SortClientConfig.class);
+ when(clientContext.getConfig()).thenReturn(sortClientConfig);
+ when(sortClientConfig.getSortTaskId()).thenReturn("test");
+ when(sortClientConfig.getUpdateMetaDataIntervalSec()).thenReturn(60);
+ queryConsumeConfig = PowerMockito.mock(QueryConsumeConfigImpl.class);
+ inLongTopicManager = new InlongTopicManagerImpl(clientContext, queryConsumeConfig);
+ }
+
+ @Test
+ public void testAddFetcher() {
+ InlongTopicManager inLongTopicManager = new InlongTopicManagerImpl(clientContext, queryConsumeConfig);
+
+ InLongTopicFetcher inLongTopicFetcher = inLongTopicManager.addFetcher(inLongTopic);
+ Assert.assertNull(inLongTopicFetcher);
+ }
+
+ @Test
+ public void testRemoveFetcher() {
+
+ InLongTopicFetcher inLongTopicFetcher = inLongTopicManager.removeFetcher(inLongTopic, true);
+ Assert.assertNull(inLongTopicFetcher);
+
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+
+ inLongTopicFetcher = inLongTopicManager.removeFetcher(inLongTopic, true);
+ Assert.assertNotNull(inLongTopicFetcher);
+
+ }
+
+ @Test
+ public void testGetFetcher() {
+ InLongTopicFetcher fetcher = inLongTopicManager.getFetcher(inLongTopic.getTopicKey());
+ Assert.assertNull(fetcher);
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+
+ fetcher = inLongTopicManager.getFetcher(inLongTopic.getTopicKey());
+ Assert.assertNotNull(fetcher);
+
+ }
+
+ @Test
+ public void testGetManagedInLongTopics() {
+ Set<String> managedInLongTopics = inLongTopicManager.getManagedInLongTopics();
+ Assert.assertEquals(0, managedInLongTopics.size());
+
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+ managedInLongTopics = inLongTopicManager.getManagedInLongTopics();
+ Assert.assertEquals(1, managedInLongTopics.size());
+
+ }
+
+ @Test
+ public void testGetAllFetchers() {
+ Collection<InLongTopicFetcher> allFetchers = inLongTopicManager.getAllFetchers();
+ Assert.assertEquals(0, allFetchers.size());
+
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+ allFetchers = inLongTopicManager.getAllFetchers();
+ Assert.assertEquals(1, allFetchers.size());
+ }
+
+ @Test
+ public void testOfflineAllTp() {
Review Comment:
What does the `Tp` mean?
If it is `topic`, then using the `Topics` is better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SortSdk] Refactor single topic manager
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944129933
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InLongSingleTopicManager.java:
##########
@@ -0,0 +1,490 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InLongTopicManager;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.kafka.InLongKafkaFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.pulsar.InLongPulsarFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.InLongTubeFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.TubeConsumerCreater;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InLongSingleTopicManager extends InLongTopicManager {
Review Comment:
1. Please add some Java docs for new classes.
2. Suggested changing the `InLongXxx` to `InlongXxx`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #5512: [INLONG-5496][SortSdk] Refactor single topic manager
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944131040
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InLongSingleTopicManager.java:
##########
@@ -0,0 +1,490 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InLongTopicManager;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.kafka.InLongKafkaFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.pulsar.InLongPulsarFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.InLongTubeFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.TubeConsumerCreater;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InLongSingleTopicManager extends InLongTopicManager {
+
+ private final Logger logger = LoggerFactory.getLogger(InLongSingleTopicManager.class);
Review Comment:
This field should be static, and the name of `static final` should use uppercase letters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r944168597
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/manager/InLongSingleTopicManager.java:
##########
@@ -0,0 +1,490 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InLongTopicManager;
+import org.apache.inlong.sdk.sort.api.InlongTopicTypeEnum;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.ConsumeConfig;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.kafka.InLongKafkaFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.pulsar.InLongPulsarFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.InLongTubeFetcherImpl;
+import org.apache.inlong.sdk.sort.impl.tube.TubeConsumerCreater;
+import org.apache.inlong.sdk.sort.util.PeriodicTask;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.inlong.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.factory.MessageSessionFactory;
+import org.apache.inlong.tubemq.client.factory.TubeSingleSessionFactory;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InLongSingleTopicManager extends InLongTopicManager {
Review Comment:
fixed, thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] vernedeng commented on a diff in pull request #5512: [INLONG-5496][SDK] Refactor single topic manager in Sort SDK
Posted by GitBox <gi...@apache.org>.
vernedeng commented on code in PR #5512:
URL: https://github.com/apache/inlong/pull/5512#discussion_r945027698
##########
inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/manager/InlongSingleTopicManagerTest.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.inlong.sdk.sort.manager;
+
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.api.InlongTopicManager;
+import org.apache.inlong.sdk.sort.api.QueryConsumeConfig;
+import org.apache.inlong.sdk.sort.api.SortClientConfig;
+import org.apache.inlong.sdk.sort.entity.CacheZoneCluster;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.ClientContextImpl;
+import org.apache.inlong.sdk.sort.impl.InlongTopicManagerImpl;
+import org.apache.inlong.sdk.sort.impl.QueryConsumeConfigImpl;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ClientContext.class})
+public class InlongSingleTopicManagerTest {
+
+ private InLongTopic inLongTopic;
+ private ClientContext clientContext;
+ private QueryConsumeConfig queryConsumeConfig;
+ private InlongTopicManager inLongTopicManager;
+
+ {
+ System.setProperty("log4j2.disable.jmx", Boolean.TRUE.toString());
+
+ inLongTopic = new InLongTopic();
+ inLongTopic.setTopic("testTopic");
+ inLongTopic.setPartitionId(0);
+ inLongTopic.setTopicType("pulsar");
+ inLongTopic.setProperties(new HashMap<>());
+
+ CacheZoneCluster cacheZoneCluster = new CacheZoneCluster("clusterId", "bootstraps", "token");
+ inLongTopic.setInLongCluster(cacheZoneCluster);
+
+ clientContext = PowerMockito.mock(ClientContextImpl.class);
+
+ SortClientConfig sortClientConfig = PowerMockito.mock(SortClientConfig.class);
+ when(clientContext.getConfig()).thenReturn(sortClientConfig);
+ when(sortClientConfig.getSortTaskId()).thenReturn("test");
+ when(sortClientConfig.getUpdateMetaDataIntervalSec()).thenReturn(60);
+ queryConsumeConfig = PowerMockito.mock(QueryConsumeConfigImpl.class);
+ inLongTopicManager = new InlongTopicManagerImpl(clientContext, queryConsumeConfig);
+ }
+
+ @Test
+ public void testAddFetcher() {
+ InlongTopicManager inLongTopicManager = new InlongTopicManagerImpl(clientContext, queryConsumeConfig);
+
+ InLongTopicFetcher inLongTopicFetcher = inLongTopicManager.addFetcher(inLongTopic);
+ Assert.assertNull(inLongTopicFetcher);
+ }
+
+ @Test
+ public void testRemoveFetcher() {
+
+ InLongTopicFetcher inLongTopicFetcher = inLongTopicManager.removeFetcher(inLongTopic, true);
+ Assert.assertNull(inLongTopicFetcher);
+
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+
+ inLongTopicFetcher = inLongTopicManager.removeFetcher(inLongTopic, true);
+ Assert.assertNotNull(inLongTopicFetcher);
+
+ }
+
+ @Test
+ public void testGetFetcher() {
+ InLongTopicFetcher fetcher = inLongTopicManager.getFetcher(inLongTopic.getTopicKey());
+ Assert.assertNull(fetcher);
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+
+ fetcher = inLongTopicManager.getFetcher(inLongTopic.getTopicKey());
+ Assert.assertNotNull(fetcher);
+
+ }
+
+ @Test
+ public void testGetManagedInLongTopics() {
+ Set<String> managedInLongTopics = inLongTopicManager.getManagedInLongTopics();
+ Assert.assertEquals(0, managedInLongTopics.size());
+
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+ managedInLongTopics = inLongTopicManager.getManagedInLongTopics();
+ Assert.assertEquals(1, managedInLongTopics.size());
+
+ }
+
+ @Test
+ public void testGetAllFetchers() {
+ Collection<InLongTopicFetcher> allFetchers = inLongTopicManager.getAllFetchers();
+ Assert.assertEquals(0, allFetchers.size());
+
+ ConcurrentHashMap<String, InLongTopicFetcher> fetchers = new ConcurrentHashMap<>();
+ InLongTopicFetcher inLongTopicFetcherRmMock = PowerMockito.mock(InLongTopicFetcher.class);
+ fetchers.put(inLongTopic.getTopicKey(), inLongTopicFetcherRmMock);
+ Whitebox.setInternalState(inLongTopicManager, "fetchers", fetchers);
+ allFetchers = inLongTopicManager.getAllFetchers();
+ Assert.assertEquals(1, allFetchers.size());
+ }
+
+ @Test
+ public void testOfflineAllTp() {
Review Comment:
it means topic and partition
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org