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/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 47f19b3 [INLONG-2077] sort-sdk change pulsar consume mode from
listener to fetch (#2142)
47f19b3 is described below
commit 47f19b32cf85f8e360c0e870ece6fc4a8b47a47f
Author: wardli <[email protected]>
AuthorDate: Mon Jan 17 19:43:16 2022 +0800
[INLONG-2077] sort-sdk change pulsar consume mode from listener to fetch
(#2142)
---
.../apache/inlong/sdk/sort/api/ClientContext.java | 7 +
.../inlong/sdk/sort/api/InLongTopicFetcher.java | 11 +-
.../apache/inlong/sdk/sort/api/ReadCallback.java | 3 +
.../inlong/sdk/sort/api/SortClientConfig.java | 37 +++
.../InLongMessage.java} | 26 +-
.../inlong/sdk/sort/entity/MessageRecord.java | 21 +-
.../sdk/sort/impl/InLongPulsarFetcherImpl.java | 224 --------------
.../sdk/sort/impl/InLongTopicManagerImpl.java | 1 +
.../sdk/sort/impl/PulsarMessageListener.java | 97 ------
.../sort/impl/pulsar/InLongPulsarFetcherImpl.java | 336 +++++++++++++++++++++
.../sdk/sort/stat/SortClientStateCounter.java | 103 +++++--
.../apache/inlong/sdk/sort/stat/StatManager.java | 9 +-
.../sdk/sort/impl/InLongPulsarFetcherImplTest.java | 19 +-
.../standalone/source/sortsdk/FetchCallback.java | 44 ++-
.../standalone/source/sortsdk/SortSdkSource.java | 25 +-
.../source/sortsdk/SubscribeFetchResult.java | 49 ++-
pom.xml | 8 +-
17 files changed, 580 insertions(+), 440 deletions(-)
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ClientContext.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ClientContext.java
index 73a9d5f..060d8fe 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ClientContext.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ClientContext.java
@@ -44,4 +44,11 @@ public abstract class ClientContext implements Cleanable {
return statManager;
}
+ public void acquireRequestPermit() throws InterruptedException {
+ config.getGlobalInProgressRequest().acquireUninterruptibly();
+ }
+
+ public void releaseRequestPermit() {
+ config.getGlobalInProgressRequest().release();
+ }
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
index dfcb725..e2ef03e 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
@@ -18,9 +18,6 @@
package org.apache.inlong.sdk.sort.api;
import org.apache.inlong.sdk.sort.entity.InLongTopic;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
public abstract class InLongTopicFetcher {
@@ -32,7 +29,7 @@ public abstract class InLongTopicFetcher {
this.context = context;
}
- public abstract boolean init(PulsarClient pulsarClient);
+ public abstract boolean init(Object client);
public abstract void ack(String msgOffset) throws Exception;
@@ -53,10 +50,4 @@ public abstract class InLongTopicFetcher {
public abstract long getConsumedDataSize();
public abstract long getAckedOffset();
-
- public abstract void seek(long offset) throws Exception;
-
- public abstract void seek(MessageId messageId) throws
PulsarClientException;
-
- public abstract String getFetcherType();
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
index 5f7f810..9306e5c 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
@@ -17,9 +17,12 @@
package org.apache.inlong.sdk.sort.api;
+import java.util.List;
import org.apache.inlong.sdk.sort.entity.MessageRecord;
public interface ReadCallback {
void onFinished(MessageRecord messageRecord);
+
+ void onFinishedBatch(List<MessageRecord> messageRecord);
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
index b50eace..116e7e5 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/SortClientConfig.java
@@ -18,6 +18,7 @@
package org.apache.inlong.sdk.sort.api;
import java.io.Serializable;
+import java.util.concurrent.Semaphore;
public class SortClientConfig implements Serializable {
@@ -31,6 +32,7 @@ public class SortClientConfig implements Serializable {
private ReadCallback callback;
private int callbackQueueSize = 100;
private int pulsarReceiveQueueSize = 2000;
+ private Semaphore globalInProgressRequest = new Semaphore(Short.MAX_VALUE,
true);
private String localIp;
private String appName;
private String serverName;
@@ -45,6 +47,9 @@ public class SortClientConfig implements Serializable {
private int ackTimeoutSec = 10;
private volatile boolean stopConsume = false;
private boolean isPrometheusEnabled = true;
+ private int emptyPollSleepStepMs = 50;
+ private int maxEmptyPollSleepMs = 500;
+ private int emptyPollTimes = 10;
public SortClientConfig(String sortTaskId, String sortClusterName,
InLongTopicChangeListener assignmentsListener,
ConsumeStrategy consumeStrategy, String localIp) {
@@ -155,6 +160,14 @@ public class SortClientConfig implements Serializable {
this.pulsarReceiveQueueSize = pulsarReceiveQueueSize;
}
+ public Semaphore getGlobalInProgressRequest() {
+ return globalInProgressRequest;
+ }
+
+ public void setGlobalInProgressRequest(Semaphore globalInProgressRequest) {
+ this.globalInProgressRequest = globalInProgressRequest;
+ }
+
/**
* get localIp
*
@@ -237,6 +250,30 @@ public class SortClientConfig implements Serializable {
isPrometheusEnabled = prometheusEnabled;
}
+ public int getEmptyPollSleepStepMs() {
+ return emptyPollSleepStepMs;
+ }
+
+ public void setEmptyPollSleepStepMs(int emptyPollSleepStepMs) {
+ this.emptyPollSleepStepMs = emptyPollSleepStepMs;
+ }
+
+ public int getMaxEmptyPollSleepMs() {
+ return maxEmptyPollSleepMs;
+ }
+
+ public void setMaxEmptyPollSleepMs(int maxEmptyPollSleepMs) {
+ this.maxEmptyPollSleepMs = maxEmptyPollSleepMs;
+ }
+
+ public int getEmptyPollTimes() {
+ return emptyPollTimes;
+ }
+
+ public void setEmptyPollTimes(int emptyPollTimes) {
+ this.emptyPollTimes = emptyPollTimes;
+ }
+
/**
* ConsumeStrategy
*/
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java
similarity index 60%
copy from
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
copy to
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java
index 5f7f810..f332812 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/ReadCallback.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java
@@ -15,11 +15,29 @@
* limitations under the License.
*/
-package org.apache.inlong.sdk.sort.api;
+package org.apache.inlong.sdk.sort.entity;
-import org.apache.inlong.sdk.sort.entity.MessageRecord;
+import java.util.Map;
-public interface ReadCallback {
+public class InLongMessage {
- void onFinished(MessageRecord messageRecord);
+ private final Map<String, String> msgHeader;
+ private byte[] data;
+
+ public InLongMessage(byte[] data, Map<String, String> msgHeader) {
+ this.data = data;
+ this.msgHeader = msgHeader;
+ }
+
+ public byte[] getData() {
+ return data;
+ }
+
+ public void setData(byte[] data) {
+ this.data = data;
+ }
+
+ public Map<String, String> getMsgHeader() {
+ return msgHeader;
+ }
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
index 0abf1f0..6917426 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
@@ -17,22 +17,20 @@
package org.apache.inlong.sdk.sort.entity;
-import java.util.Map;
+import java.util.List;
public class MessageRecord {
private final String msgKey;
- private final byte[] message;
- private final Map<String, String> msgHeader;
+ private final List<InLongMessage> msgs;
private final String offset;
private final long recTime;
- public MessageRecord(String msgKey, byte[] message, Map<String, String>
msgHeader, String offset, long recTime) {
+ public MessageRecord(String msgKey, List<InLongMessage> msgs, String
offset, long recTime) {
this.msgKey = msgKey;
- this.message = message;
- this.msgHeader = msgHeader;
+ this.msgs = msgs;
this.offset = offset;
this.recTime = recTime;
}
@@ -41,12 +39,8 @@ public class MessageRecord {
return msgKey;
}
- public byte[] getMessage() {
- return message;
- }
-
- public Map<String, String> getMsgHeader() {
- return msgHeader;
+ public List<InLongMessage> getMsgs() {
+ return msgs;
}
public String getOffset() {
@@ -61,8 +55,7 @@ public class MessageRecord {
public String toString() {
return "MessageRecord{"
+ "msgKey='" + msgKey
- + ", message=" + new String(message)
- + ", msgHeader=" + msgHeader
+ + ", message=" + String.valueOf(msgs)
+ ", offset='" + offset
+ ", recTime=" + recTime
+ '}';
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImpl.java
deleted file mode 100644
index 85c9047..0000000
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImpl.java
+++ /dev/null
@@ -1,224 +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.inlong.sdk.sort.impl;
-
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.inlong.sdk.sort.api.ClientContext;
-import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
-import org.apache.inlong.sdk.sort.entity.InLongTopic;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.shade.org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class InLongPulsarFetcherImpl extends InLongTopicFetcher {
-
- private final Logger logger =
LoggerFactory.getLogger(InLongPulsarFetcherImpl.class);
- private final ReentrantReadWriteLock mainLock = new
ReentrantReadWriteLock(true);
- private final ConcurrentHashMap<String, MessageId> offsetCache = new
ConcurrentHashMap<>();
- private volatile boolean closed = false;
- private Consumer<byte[]> consumer;
- private volatile boolean stopConsume = false;
-
- public InLongPulsarFetcherImpl(InLongTopic inLongTopic,
- ClientContext context) {
- super(inLongTopic, context);
- }
-
- @Override
- public void stopConsume(boolean stopConsume) {
- this.stopConsume = stopConsume;
- }
-
- @Override
- public boolean isConsumeStop() {
- return stopConsume;
- }
-
- @Override
- public InLongTopic getInLongTopic() {
- return inLongTopic;
- }
-
- @Override
- public long getConsumedDataSize() {
- return 0;
- }
-
- @Override
- public long getAckedOffset() {
- return 0;
- }
-
- @Override
- public void seek(long offset) throws Exception {
-
- }
-
- @Override
- public void seek(MessageId messageId) throws PulsarClientException {
- if (consumer != null) {
- consumer.seek(messageId);
- }
- }
-
- @Override
- public String getFetcherType() {
- return inLongTopic.getTopicType();
- }
-
- private void ackSucc(String offset) {
- logger.info("ack succ:{}", offset);
- offsetCache.remove(offset);
-
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic()).addAckSuccTimes(1);
- }
-
- /**
- * ack Offset
- *
- * @param msgOffset String
- */
- @Override
- public void ack(String msgOffset) throws Exception {
- if (!StringUtils.isEmpty(msgOffset)) {
- logger.debug("## ack {}", msgOffset);
- try {
- if (consumer == null) {
-
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
- .addAckFailTimes(1);
- logger.error("consumer == null {}", msgOffset);
- return;
- }
- MessageId messageId = offsetCache.get(msgOffset);
- if (messageId == null) {
-
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
- .addAckFailTimes(1);
- logger.error("messageId == null {}", msgOffset);
- return;
- }
- consumer.acknowledgeAsync(messageId)
- .thenAccept(consumer -> ackSucc(msgOffset))
- .exceptionally(exception -> {
- logger.error("ack fail:{}", msgOffset);
-
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
-
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
- .addAckFailTimes(1);
- return null;
- });
- } catch (Exception e) {
-
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic()).addAckFailTimes(1);
- logger.error(e.getMessage(), e);
- throw e;
- }
- }
- }
-
- /**
- * create and init Consumer
- *
- * @return boolean
- */
- @Override
- public boolean init(PulsarClient pulsarClient) {
- return createConsumer(pulsarClient);
- }
-
- private boolean createConsumer(PulsarClient client) {
- try {
- consumer = client.newConsumer(Schema.BYTES)
- .topic(inLongTopic.getTopic())
- .subscriptionName(context.getConfig().getSortTaskId())
- .subscriptionType(SubscriptionType.Shared)
- .startMessageIdInclusive()
- .ackTimeout(context.getConfig().getAckTimeoutSec(),
TimeUnit.SECONDS)
-
.receiverQueueSize(context.getConfig().getPulsarReceiveQueueSize())
- .messageListener(new PulsarMessageListener(this, context,
inLongTopic, offsetCache))
- .subscribe();
- } catch (Exception e) {
- e.printStackTrace();
- return false;
- }
- return true;
- }
-
- /**
- * current fetcher is closed
- */
- public void isValidState() {
- if (closed) {
- throw new IllegalStateException(inLongTopic + " closed.");
- }
- }
-
- /**
- * pause
- */
- @Override
- public void pause() {
- if (consumer != null) {
- consumer.pause();
- }
- }
-
- /**
- * resume
- */
- @Override
- public void resume() {
- if (consumer != null) {
- consumer.resume();
- }
- }
-
- @Override
- public boolean close() {
- mainLock.writeLock().lock();
- try {
- this.closed = true;
- try {
- if (consumer != null) {
- consumer.close();
- }
- } catch (PulsarClientException e) {
- e.printStackTrace();
- }
-
- logger.info("{} closed.", inLongTopic);
- return true;
- } finally {
- mainLock.writeLock().unlock();
- }
- }
-
- @Override
- public boolean isClosed() {
- return closed;
- }
-
-}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
index 012d6b1..cfbd56c 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
@@ -34,6 +34,7 @@ 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.pulsar.InLongPulsarFetcherImpl;
import org.apache.inlong.sdk.sort.util.PeriodicTask;
import org.apache.inlong.sdk.sort.util.StringUtil;
import org.apache.pulsar.client.api.AuthenticationFactory;
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/PulsarMessageListener.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/PulsarMessageListener.java
deleted file mode 100644
index f4afe82..0000000
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/PulsarMessageListener.java
+++ /dev/null
@@ -1,97 +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.inlong.sdk.sort.impl;
-
-import java.util.Base64;
-import java.util.concurrent.ConcurrentHashMap;
-import org.apache.inlong.sdk.sort.api.ClientContext;
-import org.apache.inlong.sdk.sort.entity.InLongTopic;
-import org.apache.inlong.sdk.sort.entity.MessageRecord;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.MessageListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class PulsarMessageListener implements MessageListener<byte[]> {
-
- private final Logger logger =
LoggerFactory.getLogger(PulsarMessageListener.class);
-
- private final ClientContext clientContext;
- private final InLongTopic inLongTopic;
- private final InLongPulsarFetcherImpl inLongTopicInFetcher;
- private final ConcurrentHashMap<String, MessageId> offsetCache;
-
- public PulsarMessageListener(InLongPulsarFetcherImpl inLongTopicInFetcher,
ClientContext clientContext,
- InLongTopic inLongTopic, ConcurrentHashMap<String, MessageId>
offsetCache) {
- this.inLongTopicInFetcher = inLongTopicInFetcher;
- this.clientContext = clientContext;
- this.inLongTopic = inLongTopic;
- this.offsetCache = offsetCache;
- }
-
- /**
- * put the received msg to onFinished method
- *
- * @param messageRecord MessageRecord
- */
- public void handleMsg(MessageRecord messageRecord) {
- long start = System.currentTimeMillis();
- try {
- clientContext.getConfig().getCallback().onFinished(messageRecord);
- clientContext.getStatManager()
- .getStatistics(clientContext.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
- .addCallbackTimeCost(System.currentTimeMillis() -
start).addCallbackDoneTimes(1);
- } catch (Exception e) {
- clientContext.getStatManager()
- .getStatistics(clientContext.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
- .addCallbackErrorTimes(1);
- throw e;
- }
- }
-
- @Override
- public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
- String offsetKey;
- try {
- inLongTopicInFetcher.isValidState();
- clientContext.getStatManager()
- .getStatistics(clientContext.getConfig().getSortTaskId(),
- inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
-
.addConsumeSize(msg.getData().length).addCallbackTimes(1).addMsgCount(1);
-
- offsetKey = getOffset(msg.getMessageId());
- offsetCache.put(offsetKey, msg.getMessageId());
- callbackMessageRecord(msg, offsetKey);
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- private void callbackMessageRecord(Message<byte[]> msg, String offsetKey) {
- handleMsg(new MessageRecord(inLongTopic.getTopicKey(), msg.getData(),
msg.getProperties(),
- offsetKey, System.currentTimeMillis()));
- }
-
- private String getOffset(MessageId msgId) {
- return Base64.getEncoder().encodeToString(msgId.toByteArray());
- }
-}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java
new file mode 100644
index 0000000..ad463e5
--- /dev/null
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java
@@ -0,0 +1,336 @@
+/*
+ * 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.impl.pulsar;
+
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
+import org.apache.inlong.sdk.sort.entity.InLongMessage;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.entity.MessageRecord;
+import org.apache.inlong.sdk.sort.util.StringUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Messages;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.shade.org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InLongPulsarFetcherImpl extends InLongTopicFetcher {
+
+ private final Logger logger =
LoggerFactory.getLogger(InLongPulsarFetcherImpl.class);
+ private final ReentrantReadWriteLock mainLock = new
ReentrantReadWriteLock(true);
+ private final ConcurrentHashMap<String, MessageId> offsetCache = new
ConcurrentHashMap<>();
+ private volatile boolean closed = false;
+ private Consumer<byte[]> consumer;
+ private volatile boolean stopConsume = false;
+ private volatile Thread fetchThread;
+ private long sleepTime = 0L;
+ private int emptyPollTimes = 0;
+
+ public InLongPulsarFetcherImpl(InLongTopic inLongTopic,
+ ClientContext context) {
+ super(inLongTopic, context);
+ }
+
+ @Override
+ public void stopConsume(boolean stopConsume) {
+ this.stopConsume = stopConsume;
+ }
+
+ @Override
+ public boolean isConsumeStop() {
+ return stopConsume;
+ }
+
+ @Override
+ public InLongTopic getInLongTopic() {
+ return inLongTopic;
+ }
+
+ @Override
+ public long getConsumedDataSize() {
+ return 0;
+ }
+
+ @Override
+ public long getAckedOffset() {
+ return 0;
+ }
+
+ private void ackSucc(String offset) {
+ offsetCache.remove(offset);
+
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic()).addAckSuccTimes(1);
+ }
+
+ /**
+ * ack Offset
+ *
+ * @param msgOffset String
+ */
+ @Override
+ public void ack(String msgOffset) throws Exception {
+ if (!StringUtils.isEmpty(msgOffset)) {
+ try {
+ if (consumer == null) {
+
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
+ .addAckFailTimes(1);
+ logger.error("consumer == null");
+ return;
+ }
+ MessageId messageId = offsetCache.get(msgOffset);
+ if (messageId == null) {
+
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
+ .addAckFailTimes(1);
+ logger.error("messageId == null");
+ return;
+ }
+ consumer.acknowledgeAsync(messageId)
+ .thenAccept(consumer -> ackSucc(msgOffset))
+ .exceptionally(exception -> {
+ logger.error("ack fail:{}", msgOffset);
+
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addAckFailTimes(1);
+ return null;
+ });
+ } catch (Exception e) {
+
context.getStatManager().getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic()).addAckFailTimes(1);
+ logger.error(e.getMessage(), e);
+ throw e;
+ }
+ }
+ }
+
+ /**
+ * create Consumer and fetch thread
+ *
+ * @return boolean
+ */
+ @Override
+ public boolean init(Object object) {
+ PulsarClient pulsarClient = (PulsarClient) object;
+ return createConsumer(pulsarClient);
+ }
+
+ private boolean createConsumer(PulsarClient client) {
+ try {
+ consumer = client.newConsumer(Schema.BYTES)
+ .topic(inLongTopic.getTopic())
+ .subscriptionName(context.getConfig().getSortTaskId())
+ .subscriptionType(SubscriptionType.Shared)
+ .startMessageIdInclusive()
+ .ackTimeout(10, TimeUnit.SECONDS)
+
.receiverQueueSize(context.getConfig().getPulsarReceiveQueueSize())
+ .subscribe();
+
+ String threadName = "sort_sdk_fetch_thread_" +
StringUtil.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
+ this.fetchThread = new Thread(new Fetcher(), threadName);
+ this.fetchThread.start();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * isValidState
+ */
+ public void isValidState() {
+ if (closed) {
+ throw new IllegalStateException(inLongTopic + " closed.");
+ }
+ }
+
+ /**
+ * pause
+ */
+ @Override
+ public void pause() {
+ if (consumer != null) {
+ consumer.pause();
+ }
+ }
+
+ /**
+ * resume
+ */
+ @Override
+ public void resume() {
+ if (consumer != null) {
+ consumer.resume();
+ }
+ }
+
+ /**
+ * close
+ *
+ * @return true/false
+ */
+ @Override
+ public boolean close() {
+ mainLock.writeLock().lock();
+ try {
+ this.closed = true;
+ try {
+ if (consumer != null) {
+ consumer.close();
+ }
+ if (fetchThread != null) {
+ fetchThread.interrupt();
+ }
+ } catch (PulsarClientException e) {
+ e.printStackTrace();
+ }
+
+ logger.info("closed {}", inLongTopic);
+ return true;
+ } finally {
+ mainLock.writeLock().unlock();
+ }
+ }
+
+ @Override
+ public boolean isClosed() {
+ return closed;
+ }
+
+ public class Fetcher implements Runnable {
+
+ /**
+ * put the received msg to onFinished method
+ *
+ * @param messageRecords {@link List<MessageRecord>}
+ */
+ private void handleAndCallbackMsg(List<MessageRecord> messageRecords) {
+ long start = System.currentTimeMillis();
+ try {
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
+ .addCallbackTimes(1);
+
context.getConfig().getCallback().onFinishedBatch(messageRecords);
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
+ .addCallbackTimeCost(System.currentTimeMillis() -
start).addCallbackDoneTimes(1);
+ } catch (Exception e) {
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getTopic())
+ .addCallbackErrorTimes(1);
+ e.printStackTrace();
+ }
+ }
+
+ private String getOffset(MessageId msgId) {
+ return Base64.getEncoder().encodeToString(msgId.toByteArray());
+ }
+
+ @Override
+ public void run() {
+ boolean hasPermit;
+ while (true) {
+ hasPermit = false;
+ try {
+ if (context.getConfig().isStopConsume() || stopConsume) {
+ TimeUnit.MILLISECONDS.sleep(50);
+ continue;
+ }
+
+ if (sleepTime > 0) {
+ TimeUnit.MILLISECONDS.sleep(sleepTime);
+ }
+
+ context.acquireRequestPermit();
+ hasPermit = true;
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addMsgCount(1).addFetchTimes(1);
+
+ long startFetchTime = System.currentTimeMillis();
+ Messages<byte[]> messages = consumer.batchReceive();
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addFetchTimeCost(System.currentTimeMillis() -
startFetchTime);
+ if (null != messages && messages.size() != 0) {
+ List<MessageRecord> msgs = new ArrayList<>();
+ for (Message<byte[]> msg : messages) {
+ String offsetKey = getOffset(msg.getMessageId());
+ offsetCache.put(offsetKey, msg.getMessageId());
+
+ msgs.add(new
MessageRecord(inLongTopic.getTopicKey(),
+ Collections.singletonList(new
InLongMessage(msg.getData(), msg.getProperties())),
+ offsetKey, System.currentTimeMillis()));
+ context.getStatManager()
+
.getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addConsumeSize(msg.getData().length);
+ }
+ context.getStatManager()
+
.getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addMsgCount(msgs.size());
+ handleAndCallbackMsg(msgs);
+ sleepTime = 0L;
+ } else {
+ context.getStatManager()
+
.getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addEmptyFetchTimes(1);
+ emptyPollTimes++;
+ if (emptyPollTimes >=
context.getConfig().getEmptyPollTimes()) {
+ sleepTime = Math.min((sleepTime +=
context.getConfig().getEmptyPollSleepStepMs()),
+
context.getConfig().getMaxEmptyPollSleepMs());
+ emptyPollTimes = 0;
+ }
+ }
+ } catch (Exception e) {
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+
inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addFetchErrorTimes(1);
+ logger.error(e.getMessage(), e);
+ } finally {
+ if (hasPermit) {
+ context.releaseRequestPermit();
+ }
+ }
+ }
+ }
+ }
+}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
index 5fe1c75..e9352b6 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
@@ -45,7 +45,7 @@ public class SortClientStateCounter {
/**
* reset Counter
*
- * @return SortClientStateCounter
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter reset() {
SortClientStateCounter counter = new
SortClientStateCounter(sortTaskId, cacheClusterId, topic, partitionId);
@@ -72,7 +72,7 @@ public class SortClientStateCounter {
* consume byte size
*
* @param num long byte
- * @return SortClientStateCounter
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addConsumeSize(long num) {
count.getAndAdd(0, num);
@@ -82,8 +82,8 @@ public class SortClientStateCounter {
/**
* count receive event num
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addMsgCount(long num) {
count.getAndAdd(1, num);
@@ -93,8 +93,8 @@ public class SortClientStateCounter {
/**
* count callbak times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addCallbackTimes(long num) {
count.getAndAdd(2, num);
@@ -104,8 +104,8 @@ public class SortClientStateCounter {
/**
* count callbak done times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addCallbackDoneTimes(long num) {
count.getAndAdd(3, num);
@@ -116,7 +116,7 @@ public class SortClientStateCounter {
* count callbak time cost
*
* @param num long
- * @return SortClientStateCounter
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addCallbackTimeCost(long num) {
count.getAndAdd(4, num);
@@ -126,8 +126,8 @@ public class SortClientStateCounter {
/**
* count callbak error times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addCallbackErrorTimes(long num) {
count.getAndAdd(5, num);
@@ -137,8 +137,8 @@ public class SortClientStateCounter {
/**
* count topic online times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addTopicOnlineTimes(long num) {
count.getAndAdd(6, num);
@@ -148,8 +148,8 @@ public class SortClientStateCounter {
/**
* count topic offline times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addTopicOfflineTimes(long num) {
count.getAndAdd(7, num);
@@ -159,8 +159,8 @@ public class SortClientStateCounter {
/**
* count ack fail times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addAckFailTimes(long num) {
count.getAndAdd(8, num);
@@ -170,8 +170,8 @@ public class SortClientStateCounter {
/**
* count ack succ times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addAckSuccTimes(long num) {
count.getAndAdd(9, num);
@@ -181,8 +181,8 @@ public class SortClientStateCounter {
/**
* count request manager times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addRequestManagerTimes(long num) {
count.getAndAdd(10, num);
@@ -193,7 +193,7 @@ public class SortClientStateCounter {
* count request manager time cost
*
* @param num long
- * @return SortClientStateCounter
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addRequestManagerTimeCost(long num) {
count.getAndAdd(11, num);
@@ -203,8 +203,8 @@ public class SortClientStateCounter {
/**
* count request manager fail times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addRequestManagerFailTimes(long num) {
count.getAndAdd(12, num);
@@ -214,8 +214,8 @@ public class SortClientStateCounter {
/**
* count manager conf changed times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addManagerConfChangedTimes(long num) {
count.getAndAdd(13, num);
@@ -225,8 +225,8 @@ public class SortClientStateCounter {
/**
* count manager result code common error times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addRequestManagerCommonErrorTimes(long num) {
count.getAndAdd(14, num);
@@ -236,12 +236,55 @@ public class SortClientStateCounter {
/**
* count manager result param error times
*
- * @param num int
- * @return SortClientStateCounter
+ * @param num long
+ * @return {@link SortClientStateCounter}
*/
public SortClientStateCounter addRequestManagerParamErrorTimes(long num) {
count.getAndAdd(15, num);
return this;
}
+ /**
+ * count thread fetch times
+ *
+ * @param num long
+ * @return {@link SortClientStateCounter}
+ */
+ public SortClientStateCounter addFetchTimes(long num) {
+ count.getAndAdd(16, num);
+ return this;
+ }
+
+ /**
+ * count fetch error times
+ *
+ * @param num long
+ * @return {@link SortClientStateCounter}
+ */
+ public SortClientStateCounter addFetchErrorTimes(long num) {
+ count.getAndAdd(17, num);
+ return this;
+ }
+
+ /**
+ * count empty fetch times
+ *
+ * @param num long
+ * @return {@link SortClientStateCounter}
+ */
+ public SortClientStateCounter addEmptyFetchTimes(long num) {
+ count.getAndAdd(18, num);
+ return this;
+ }
+
+ /**
+ * count fetch time cost
+ *
+ * @param num long
+ * @return {@link SortClientStateCounter}
+ */
+ public SortClientStateCounter addFetchTimeCost(long num) {
+ count.getAndAdd(19, num);
+ return this;
+ }
}
diff --git
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
index b4ca542..cfa9ad6 100644
---
a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
+++
b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/StatManager.java
@@ -41,8 +41,8 @@ public class StatManager implements Cleanable {
/**
* StatManager Constructor
*
- * @param context ClientContext
- * @param reporter MetricReporter
+ * @param context {@link ClientContext}
+ * @param reporter {@link MetricReporter}
*/
public StatManager(ClientContext context, MetricReporter reporter) {
this.config = context.getConfig();
@@ -105,7 +105,7 @@ public class StatManager implements Cleanable {
}
/**
- * use for
+ * use for common SortClientStateCounter
*
* @param sortTaskId String
* @param clusterId String
@@ -142,9 +142,8 @@ public class StatManager implements Cleanable {
}
}
} catch (Exception e) {
- logger.error("StatManager doWork error" + e.getMessage(), e);
+ logger.error("StatManager doWork error " + e.getMessage(), e);
}
}
}
-
}
diff --git
a/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImplTest.java
b/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImplTest.java
index 19078cd..ca17f58 100644
---
a/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImplTest.java
+++
b/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/InLongPulsarFetcherImplTest.java
@@ -29,6 +29,7 @@ import org.apache.inlong.sdk.sort.api.InLongTopicFetcher;
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.pulsar.InLongPulsarFetcherImpl;
import org.apache.inlong.sdk.sort.stat.SortClientStateCounter;
import org.apache.inlong.sdk.sort.stat.StatManager;
import org.apache.pulsar.client.api.Consumer;
@@ -109,17 +110,6 @@ public class InLongPulsarFetcherImplTest {
}
@Test
- public void seek() {
- InLongTopicFetcher inLongTopicFetcher = new
InLongPulsarFetcherImpl(inLongTopic, clientContext);
- MessageId messageId = PowerMockito.mock(MessageId.class);
- try {
- inLongTopicFetcher.seek(messageId);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- @Test
public void ack() {
InLongTopicFetcher inLongTopicFetcher = new
InLongPulsarFetcherImpl(inLongTopic, clientContext);
MessageId messageId = PowerMockito.mock(MessageId.class);
@@ -188,11 +178,4 @@ public class InLongPulsarFetcherImplTest {
boolean closed = inLongTopicFetcher.isClosed();
Assert.assertTrue(closed);
}
-
- @Test
- public void getFetcherType() {
- InLongTopicFetcher inLongTopicFetcher = new
InLongPulsarFetcherImpl(inLongTopic, clientContext);
- String fetcherType = inLongTopicFetcher.getFetcherType();
- Assert.assertEquals(inLongTopic.getTopicType(), fetcherType);
- }
}
\ No newline at end of file
diff --git
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
index 87ac897..53e96dd 100644
---
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
+++
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
@@ -18,28 +18,29 @@
package org.apache.inlong.sort.standalone.source.sortsdk;
import com.google.common.base.Preconditions;
+import java.util.List;
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.inlong.sdk.sort.api.ReadCallback;
import org.apache.inlong.sdk.sort.api.SortClient;
+import org.apache.inlong.sdk.sort.entity.InLongMessage;
import org.apache.inlong.sdk.sort.entity.MessageRecord;
import org.apache.inlong.sort.standalone.channel.ProfileEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.validation.constraints.NotBlank;
-import javax.validation.constraints.NotNull;
-
/**
* Implementation of {@link ReadCallback}.
*
* TODO: Sort sdk should deliver one object which is held by {@link
ProfileEvent} and used to ack upstream data store
* The code should be like :
*
- * public void onFinished(final MessageRecord messageRecord, ACKer
acker) {
- * doSomething();
- * final ProfileEvent profileEvent = new
ProfileEvent(result.getBody(), result.getHeaders(), acker);
- * channelProcessor.processEvent(profileEvent);
- * }
+ * public void onFinished(final MessageRecord messageRecord, ACKer acker) {
+ * doSomething();
+ * final ProfileEvent profileEvent = new ProfileEvent(result.getBody(),
result.getHeaders(), acker);
+ * channelProcessor.processEvent(profileEvent);
+ * }
*
* The ACKer will be used to <b>ACK</b> upstream after that the downstream
<b>ACKed</b> sort-standalone.
* This process seems like <b>transaction</b> of the whole sort-standalone,
and which
@@ -81,6 +82,7 @@ public class FetchCallback implements ReadCallback {
/**
* Set client for ack.
+ *
* @param client client for ack.
*/
public void setClient(@NotNull SortClient client) {
@@ -100,10 +102,17 @@ public class FetchCallback implements ReadCallback {
public void onFinished(final MessageRecord messageRecord) {
try {
Preconditions.checkState(messageRecord != null, "Fetched msg is
null.");
- final SubscribeFetchResult result =
SubscribeFetchResult.Factory.create(sortId, messageRecord);
- final ProfileEvent profileEvent = new
ProfileEvent(result.getBody(), result.getHeaders());
- channelProcessor.processEvent(profileEvent);
- context.reportToMetric(profileEvent, sortId, "-",
SortSdkSourceContext.FetchResult.SUCCESS);
+ for (InLongMessage inLongMessage : messageRecord.getMsgs()) {
+ //TODO fix here
+ final SubscribeFetchResult result =
SubscribeFetchResult.Factory
+ .create(sortId, messageRecord.getMsgKey(),
messageRecord.getOffset(),
+ inLongMessage.getMsgHeader(),
messageRecord.getRecTime(),
+ inLongMessage.getData());
+ final ProfileEvent profileEvent = new
ProfileEvent(result.getBody(), result.getHeaders());
+ channelProcessor.processEvent(profileEvent);
+ context.reportToMetric(profileEvent, sortId, "-",
SortSdkSourceContext.FetchResult.SUCCESS);
+ }
+
client.ack(messageRecord.getMsgKey(), messageRecord.getMsgKey());
} catch (NullPointerException npe) {
LOG.error("Got a null pointer exception for sortId " + sortId,
npe);
@@ -114,6 +123,16 @@ public class FetchCallback implements ReadCallback {
}
/**
+ * The callback function that SortSDK invoke when fetch messages batch
+ *
+ * @param messageRecord {@link List<MessageRecord>}
+ */
+ @Override
+ public void onFinishedBatch(List<MessageRecord> messageRecord) {
+ //TODO
+ }
+
+ /**
* Factory of {@link FetchCallback}
*/
public static class Factory {
@@ -125,7 +144,6 @@ public class FetchCallback implements ReadCallback {
* @param sortId The sortId of fetched message.
* @param channelProcessor The channelProcessor that put message in
specific channel.
* @param context The context to report fetch results.
- *
* @return One FetchCallback.
*/
public static FetchCallback create(
diff --git
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
index abad948..e9a7837 100644
---
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
+++
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java
@@ -18,6 +18,14 @@
package org.apache.inlong.sort.standalone.source.sortsdk;
import com.google.common.base.Preconditions;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import org.apache.flume.Context;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
@@ -25,21 +33,11 @@ import org.apache.flume.source.AbstractSource;
import org.apache.inlong.sdk.sort.api.SortClient;
import org.apache.inlong.sdk.sort.api.SortClientConfig;
import org.apache.inlong.sdk.sort.api.SortClientFactory;
-import org.apache.inlong.sdk.sort.entity.MessageRecord;
import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
import org.apache.inlong.sort.standalone.config.pojo.SortTaskConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
/**
* Default Source implementation of InLong.
*
@@ -62,7 +60,7 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
private static final int CORE_POOL_SIZE = 1;
// Default consume strategy of {@link SortClient}.
- private static final SortClientConfig.ConsumeStrategy defaultStrategy =
SortClientConfig.ConsumeStrategy.lastest;
+ private static final SortClientConfig.ConsumeStrategy defaultStrategy =
SortClientConfig.ConsumeStrategy.lastest;
// Map of {@link SortClient}.
private Map<String, SortClient> clients;
@@ -133,8 +131,8 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
* <p> Create new clients with new sort task id, and remove the finished
or scheduled ones. </p>
*
* <p> Current version of SortSdk <b>DO NOT</b> support to get the
corresponding sort id of {@link SortClient}.
- * Hence, the maintenance of mapping of <SortId, SortClient> should be
done by Source itself. Which is not elegant,
- * the <b>REMOVE</b> of expire clients will <b>NOT</b> be supported right
now. </p>
+ * Hence, the maintenance of mapping of <SortId, SortClient> should be
done by Source itself. Which is not elegant,
+ * the <b>REMOVE</b> of expire clients will <b>NOT</b> be supported right
now. </p>
*/
private void reload() {
@@ -166,7 +164,6 @@ public final class SortSdkSource extends AbstractSource
implements Configurable,
* For more details see {@link FetchCallback#onFinished(MessageRecord)}</p>
*
* @param sortId Sort in of new client.
- *
* @return New sort client.
*/
private SortClient newClient(final String sortId) {
diff --git
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
index 638fb07..0c6bc21 100644
---
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
+++
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
@@ -17,13 +17,12 @@
package org.apache.inlong.sort.standalone.source.sortsdk;
-import org.apache.inlong.sdk.sort.entity.MessageRecord;
-import org.apache.inlong.sort.standalone.utils.Constants;
-
-import javax.validation.constraints.NotBlank;
-import javax.validation.constraints.NotNull;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
+import org.apache.inlong.sdk.sort.entity.MessageRecord;
+import org.apache.inlong.sort.standalone.utils.Constants;
/**
* SubscribeFetchResult is the wrapper of {@link MessageRecord}.
@@ -47,6 +46,7 @@ public class SubscribeFetchResult {
* @param sortId The sortId of fetched message.
* @param message Message that fetched from upstream data storage.
*/
+
private SubscribeFetchResult(
final String sortId,
final MessageRecord message) {
@@ -54,12 +54,37 @@ public class SubscribeFetchResult {
this.headers.put(Constants.HEADER_KEY_MESSAGE_KEY,
message.getMsgKey());
this.headers.put(Constants.HEADER_KEY_MSG_OFFSET, message.getOffset());
this.headers.put(Constants.HEADER_KEY_MSG_TIME,
String.valueOf(message.getRecTime()));
- this.headers.putAll(message.getMsgHeader());
- this.body = message.getMessage();
+ //TODO to fix here
+ this.headers.putAll(null);
+ this.body = null;
+ }
+
+ /**
+ * Private constructor of SubscribeFetchResult.
+ * <p> The construction of SubscribeFetchResult should be initiated by
{@link SubscribeFetchResult.Factory}.</p>
+ *
+ * @param sortId String
+ * @param msgKey String
+ * @param offset String
+ * @param headers {@link Map<String,String>}
+ * @param recTime long
+ * @param body byte[]
+ */
+ private SubscribeFetchResult(
+ final String sortId,
+ final String msgKey, final String offset, final Map<String,
String> headers, final long recTime,
+ final byte[] body) {
+ this.sortId = sortId;
+ this.headers.put(Constants.HEADER_KEY_MESSAGE_KEY, msgKey);
+ this.headers.put(Constants.HEADER_KEY_MSG_OFFSET, offset);
+ this.headers.put(Constants.HEADER_KEY_MSG_TIME,
String.valueOf(recTime));
+ this.headers.putAll(headers);
+ this.body = body;
}
/**
* Get row data that in binary format.
+ *
* @return Row data.
*/
public byte[] getBody() {
@@ -68,6 +93,7 @@ public class SubscribeFetchResult {
/**
* Get important metrics in Map format called headers.
+ *
* @return headers.
*/
public Map<String, String> getHeaders() {
@@ -76,6 +102,7 @@ public class SubscribeFetchResult {
/**
* Get sortId of fetched message.
+ *
* @return SortId of message.
*/
public String getSortId() {
@@ -93,7 +120,6 @@ public class SubscribeFetchResult {
*
* @param sortId The sortId of fetched message.
* @param messageRecord Message that fetched from upstream data
storage.
- *
* @return One SubscribeFetchResult.
*/
public static SubscribeFetchResult create(
@@ -101,5 +127,12 @@ public class SubscribeFetchResult {
@NotNull(message = "MessageRecord should not be null.") final
MessageRecord messageRecord) {
return new SubscribeFetchResult(sortId, messageRecord);
}
+
+ public static SubscribeFetchResult create(
+ final String sortId,
+ final String msgKey, final String offset, final Map<String,
String> headers, final long recTime,
+ final byte[] body) {
+ return new SubscribeFetchResult(sortId, msgKey, offset, headers,
recTime, body);
+ }
}
}
diff --git a/pom.xml b/pom.xml
index e8e5112..bf552bb 100644
--- a/pom.xml
+++ b/pom.xml
@@ -33,8 +33,10 @@
<version>0.13.0-incubating-SNAPSHOT</version>
<name>Apache InLong</name>
- <description>InLong is a one-stop data integration framework donated by
Tencent to the Apache community.
- It provides automatic, safe, reliable, and high-performance data
transmission capabilities to
+ <description>InLong is a one-stop data integration framework donated by
Tencent to the Apache
+ community.
+ It provides automatic, safe, reliable, and high-performance data
transmission capabilities
+ to
facilitate the construction of streaming-based data analysis,
modeling, and applications.
</description>
<url>https://github.com/apache/incubator-inlong</url>
@@ -95,10 +97,10 @@
<module>inlong-agent</module>
<module>inlong-manager</module>
<module>inlong-sort</module>
+ <module>inlong-sdk</module>
<module>inlong-sort-standalone</module>
<module>inlong-website</module>
<module>inlong-distribution</module>
- <module>inlong-sdk</module>
</modules>
