This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 4bbadc856d [INLONG-8751][Manager] Fix the response that was empty for
data preview (#8777)
4bbadc856d is described below
commit 4bbadc856d518bc8e84d55e52d89552d859d1e6f
Author: castor <[email protected]>
AuthorDate: Wed Aug 23 12:40:03 2023 +0800
[INLONG-8751][Manager] Fix the response that was empty for data preview
(#8777)
Co-authored-by: castorqin <[email protected]>
---
.../resource/queue/pulsar/PulsarOperator.java | 76 +++++++++++++++++-----
.../queue/pulsar/PulsarQueueResourceOperator.java | 4 +-
2 files changed, 61 insertions(+), 19 deletions(-)
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
index 470ec417b4..f3f5f36615 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java
@@ -38,6 +38,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.PersistencePolicies;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -387,29 +388,68 @@ public class PulsarOperator {
* Query topic message for the given pulsar cluster.
*/
public List<BriefMQMessage> queryLatestMessage(PulsarAdmin pulsarAdmin,
String topicFullName, String subName,
- Integer messageCount, InlongStreamInfo streamInfo) {
+ Integer messageCount, InlongStreamInfo streamInfo, boolean serial)
{
LOGGER.info("begin to query message for topic {}, subName={}",
topicFullName, subName);
-
List<BriefMQMessage> messageList = new ArrayList<>();
- for (int i = 0; i < messageCount; i++) {
- try {
- Message<byte[]> pulsarMessage =
pulsarAdmin.topics().examineMessage(topicFullName, "latest", i);
- Map<String, String> headers = pulsarMessage.getProperties();
- int wrapTypeId =
Integer.parseInt(headers.getOrDefault(InlongConstants.MSG_ENCODE_VER,
-
Integer.toString(DataProxyMsgEncType.MSG_ENCODE_TYPE_INLONGMSG.getId())));
- DeserializeOperator deserializeOperator =
deserializeOperatorFactory.getInstance(
- DataProxyMsgEncType.valueOf(wrapTypeId));
- messageList.addAll(
- deserializeOperator.decodeMsg(streamInfo,
pulsarMessage.getData(), headers, i));
- } catch (Exception e) {
- String errMsg = "decode msg error: ";
- LOGGER.error(errMsg, e);
- throw new BusinessException(errMsg + e.getMessage());
- }
+ int partitionCount = getPartitionCount(pulsarAdmin, topicFullName);
+ for (int messageIndex = 0; messageIndex < messageCount;
messageIndex++) {
+ int currentPartitionNum = messageIndex % partitionCount;
+ int messagePosition = messageIndex / partitionCount;
+ String topicNameOfPartition =
buildTopicNameOfPartition(topicFullName, currentPartitionNum, serial);
+ messageList.addAll(queryMessageFromPulsar(topicNameOfPartition,
pulsarAdmin, messageIndex,
+ streamInfo, messagePosition));
}
-
LOGGER.info("success query message by subs={} for topic={}", subName,
topicFullName);
return messageList;
}
+ /**
+ * Use pulsar admin to get topic partition count
+ */
+ private int getPartitionCount(PulsarAdmin pulsarAdmin, String
topicFullName) {
+ PartitionedTopicMetadata partitionedTopicMetadata;
+ try {
+ partitionedTopicMetadata = pulsarAdmin.topics()
+ .getPartitionedTopicMetadata(topicFullName);
+ } catch (Exception e) {
+ String errMsg = "get pulsar partition error ";
+ LOGGER.error(errMsg, e);
+ throw new BusinessException(errMsg + e.getMessage());
+ }
+ return partitionedTopicMetadata.partitions > 0 ?
partitionedTopicMetadata.partitions : 1;
+ }
+
+ /**
+ * Use pulsar admin to query message
+ */
+ private List<BriefMQMessage> queryMessageFromPulsar(String topicPartition,
PulsarAdmin pulsarAdmin, int index,
+ InlongStreamInfo streamInfo, int messagePosition) {
+ List<BriefMQMessage> briefMQMessages = new ArrayList<>();
+ try {
+ Message<byte[]> pulsarMessage =
+ pulsarAdmin.topics().examineMessage(topicPartition,
"latest", messagePosition);
+ Map<String, String> headers = pulsarMessage.getProperties();
+ int wrapTypeId =
Integer.parseInt(headers.getOrDefault(InlongConstants.MSG_ENCODE_VER,
+
Integer.toString(DataProxyMsgEncType.MSG_ENCODE_TYPE_INLONGMSG.getId())));
+ DeserializeOperator deserializeOperator =
deserializeOperatorFactory.getInstance(
+ DataProxyMsgEncType.valueOf(wrapTypeId));
+ briefMQMessages.addAll(deserializeOperator.decodeMsg(streamInfo,
pulsarMessage.getData(),
+ headers, index));
+ } catch (Exception e) {
+ LOGGER.warn("query message from pulsar error for groupId = {},
streamId = {}",
+ streamInfo.getInlongGroupId(),
+ streamInfo.getInlongStreamId(), e);
+ }
+ return briefMQMessages;
+ }
+
+ /**
+ * Build topicName Of Partition
+ */
+ private String buildTopicNameOfPartition(String topicName, int partition,
boolean serial) {
+ if (serial) {
+ return topicName;
+ }
+ return topicName + "-partition-" + partition;
+ }
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
index 0746fac320..4d4e7051e2 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java
@@ -326,8 +326,10 @@ public class PulsarQueueResourceOperator implements
QueueResourceOperator {
String fullTopicName = tenant + "/" + namespace + "/" + topicName;
String clusterTag = inlongPulsarInfo.getInlongClusterTag();
String subs = String.format(PULSAR_SUBSCRIPTION_REALTIME_REVIEW,
clusterTag, topicName);
+ boolean serial =
InlongConstants.PULSAR_QUEUE_TYPE_SERIAL.equals(inlongPulsarInfo.getQueueModule());
briefMQMessages =
- pulsarOperator.queryLatestMessage(pulsarAdmin,
fullTopicName, subs, messageCount, streamInfo);
+ pulsarOperator.queryLatestMessage(pulsarAdmin,
fullTopicName, subs, messageCount, streamInfo,
+ serial);
// insert the consumer group info into the inlong_consume table
Integer id = consumeService.saveBySystem(groupInfo, topicName,
subs);
