This is an automated email from the ASF dual-hosted git repository.
zirui 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 097f02f [INLONG-3058][DataProxy] Add some configs while creating
Pulsar producer
097f02f is described below
commit 097f02fa104309eb9e0868e5335d7f18cdab5736
Author: baomingyu <[email protected]>
AuthorDate: Fri Mar 11 11:24:07 2022 +0800
[INLONG-3058][DataProxy] Add some configs while creating Pulsar producer
---
.../config/pojo/ThirdPartyClusterConfig.java | 16 ++-
.../apache/inlong/dataproxy/sink/PulsarSink.java | 47 +++++----
.../dataproxy/sink/pulsar/PulsarClientService.java | 111 +++++++++++++++------
3 files changed, 121 insertions(+), 53 deletions(-)
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
index 44f6591..db34271 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
@@ -42,6 +42,9 @@ public class ThirdPartyClusterConfig extends Context {
private static final String ENABLE_BATCH = "enable_batch";
private static final String BLOCK_IF_QUEUE_FULL = "block_if_queue_full";
private static final String MAX_PENDING_MESSAGES = "max_pending_messages";
+ private static final String MAX_PENDING_MESSAGES_ACROSS_PARTITIONS =
+ "max_pending_messages_across_partitions";
+ private static final String COMPRESSION_TYPE = "compression_type";
private static final String MAX_BATCHING_MESSAGES =
"max_batching_messages";
private static final String RETRY_INTERVAL_WHEN_SEND_ERROR_MILL =
"retry_interval_when_send_error_ms";
private static final String SINK_THREAD_NUM = "thread_num";
@@ -93,6 +96,8 @@ public class ThirdPartyClusterConfig extends Context {
private static final boolean DEFAULT_ENABLE_BATCH = true;
private static final boolean DEFAULT_BLOCK_IF_QUEUE_FULL = true;
private static final int DEFAULT_MAX_PENDING_MESSAGES = 10000;
+ private static final int DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS =
500000;
+ private static final String DEFAULT_COMPRESSION_TYPE = "NONE";
private static final int DEFAULT_MAX_BATCHING_MESSAGES = 1000;
private static final int DEFAULT_MAX_BATCHING_BYTES = 128 * 1024;
private static final long DEFAULT_MAX_BATCHING_PUBLISH_DELAY_MILLIS = 1L;
@@ -203,6 +208,15 @@ public class ThirdPartyClusterConfig extends Context {
return getInteger(MAX_PENDING_MESSAGES, DEFAULT_MAX_PENDING_MESSAGES);
}
+ public int getMaxPendingMessagesAcrossPartitions() {
+ return getInteger(MAX_PENDING_MESSAGES_ACROSS_PARTITIONS,
+ DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS);
+ }
+
+ public String getCompressionType() {
+ return getString(COMPRESSION_TYPE, DEFAULT_COMPRESSION_TYPE);
+ }
+
public int getMaxBatchingMessages() {
return getInteger(MAX_BATCHING_MESSAGES,
DEFAULT_MAX_BATCHING_MESSAGES);
}
@@ -211,7 +225,7 @@ public class ThirdPartyClusterConfig extends Context {
return getLong(RETRY_INTERVAL_WHEN_SEND_ERROR_MILL,
DEFAULT_RETRY_INTERVAL_WHEN_SEND_ERROR_MILL);
}
- public int getRetyCnt() {
+ public int getRetryCnt() {
return getInteger(RETRY_CNT, DEFAULT_RETRY_CNT);
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
index 64be21f..645c32a 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
@@ -153,6 +153,7 @@ public class PulsarSink extends AbstractSink implements
Configurable,
* send thread pool
*/
private Thread[] sinkThreadPool;
+ private int sinkThreadPoolSize;
private PulsarClientService pulsarClientService;
private LinkedBlockingQueue<Event> eventQueue;
@@ -227,7 +228,16 @@ public class PulsarSink extends AbstractSink implements
Configurable,
pulsarCluster = configManager.getThirdPartyClusterUrl2Token();
pulsarConfig = configManager.getThirdPartyClusterConfig(); //pulsar
common config
commonProperties = configManager.getCommonProperties();
- pulsarClientService = new PulsarClientService(pulsarConfig);
+ if (keepOrder) {
+ logger.info("This is order pulsar sink!");
+ sinkThreadPoolSize = 1;
+ } else {
+ sinkThreadPoolSize = pulsarConfig.getThreadNum();
+ }
+ if (sinkThreadPoolSize <= 0) {
+ sinkThreadPoolSize = 1;
+ }
+ pulsarClientService = new PulsarClientService(pulsarConfig,
sinkThreadPoolSize);
boolean enableReportConfigLog =
Boolean.parseBoolean(commonProperties
.getOrDefault(StreamConfigLogMetric.CONFIG_LOG_REPORT_ENABLE,"true"));
@@ -268,12 +278,8 @@ public class PulsarSink extends AbstractSink implements
Configurable,
resendQueue = new LinkedBlockingQueue<EventStat>(badEventQueueSize);
Preconditions.checkArgument(pulsarConfig.getThreadNum() > 0,
"threadNum must be > 0");
- if (keepOrder) {
- logger.info("This is order pulsar sink!");
- sinkThreadPool = new Thread[1];
- } else {
- sinkThreadPool = new Thread[pulsarConfig.getThreadNum()];
- }
+
+ sinkThreadPool = new Thread[sinkThreadPoolSize];
eventQueueSize = pulsarConfig.getEventQueueSize();
eventQueue = new LinkedBlockingQueue<Event>(eventQueueSize);
@@ -383,16 +389,16 @@ public class PulsarSink extends AbstractSink implements
Configurable,
this.canSend = true;
this.canTake = true;
+ try {
+ initTopicSet(pulsarClientService,
+ new HashSet<String>(topicProperties.values()));
+ } catch (Exception e) {
+ logger.info("pulsar sink start publish topic fail.", e);
+ }
+
for (int i = 0; i < sinkThreadPool.length; i++) {
- try {
- initTopicSet(pulsarClientService,
- new HashSet<String>(topicProperties.values()));
- } catch (Exception e) {
- logger.info("pulsar sink start publish topic fail.", e);
- }
- sinkThreadPool[i] = new Thread(new SinkTask(pulsarClientService),
getName()
- + "_pulsar_sink_sender-"
- + i);
+ sinkThreadPool[i] = new Thread(new SinkTask(pulsarClientService,
i), getName()
+ + "_pulsar_sink_sender-" + i);
sinkThreadPool[i].start();
}
logger.debug("pulsar sink started");
@@ -736,8 +742,11 @@ public class PulsarSink extends AbstractSink implements
Configurable,
private PulsarClientService pulsarClientService;
- public SinkTask(PulsarClientService pulsarClientService) {
+ private int poolIndex = 0;
+
+ public SinkTask(PulsarClientService pulsarClientService, int
poolIndex) {
this.pulsarClientService = pulsarClientService;
+ this.poolIndex = poolIndex;
}
@Override
@@ -838,8 +847,8 @@ public class PulsarSink extends AbstractSink implements
Configurable,
if (pulsarConfig.getClientIdCache() && clientId !=
null) {
agentIdCache.put(clientId,
System.currentTimeMillis());
}
- boolean sendResult =
pulsarClientService.sendMessage(topic, event,
- PulsarSink.this, es);
+ boolean sendResult =
pulsarClientService.sendMessage(poolIndex, topic,
+ event, PulsarSink.this, es);
/*
* handle producer is current is null
*/
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
index 87f67dd..64da058 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
@@ -37,6 +37,7 @@ import org.apache.inlong.dataproxy.utils.MessageUtils;
import org.apache.inlong.dataproxy.utils.NetworkUtils;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
@@ -74,6 +75,8 @@ public class PulsarClientService {
private boolean enableBatch = true;
private boolean blockIfQueueFull = true;
private int maxPendingMessages = 10000;
+ private int maxPendingMessagesAcrossPartitions = 500000;
+ private CompressionType compressionType;
private int maxBatchingBytes = 128 * 1024;
private int maxBatchingMessages = 1000;
private long maxBatchingPublishDelayMillis = 1;
@@ -87,15 +90,17 @@ public class PulsarClientService {
private String localIp = "127.0.0.1";
private StreamConfigLogMetric streamConfigLogMetric;
+ private int sinkThreadPoolSize;
/**
* PulsarClientService
*
* @param pulsarConfig
*/
- public PulsarClientService(ThirdPartyClusterConfig pulsarConfig) {
+ public PulsarClientService(ThirdPartyClusterConfig pulsarConfig, int
sinkThreadPoolSize) {
+
+ this.sinkThreadPoolSize = sinkThreadPoolSize;
-// String pulsarServerUrlList =
context.getString(PULSAR_SERVER_URL_LIST);
authType = pulsarConfig.getAuthType();
sendTimeout = pulsarConfig.getSendTimeoutMs();
retryIntervalWhenSendMsgError =
pulsarConfig.getRetryIntervalWhenSendErrorMs();
@@ -109,6 +114,13 @@ public class PulsarClientService {
enableBatch = pulsarConfig.getEnableBatch();
blockIfQueueFull = pulsarConfig.getBlockIfQueueFull();
maxPendingMessages = pulsarConfig.getMaxPendingMessages();
+ maxPendingMessagesAcrossPartitions =
pulsarConfig.getMaxPendingMessagesAcrossPartitions();
+ String compressionTypeStr = pulsarConfig.getCompressionType();
+ if (StringUtils.isNotEmpty(compressionTypeStr)) {
+ compressionType = CompressionType.valueOf(compressionTypeStr);
+ } else {
+ compressionType = CompressionType.NONE;
+ }
maxBatchingMessages = pulsarConfig.getMaxBatchingMessages();
maxBatchingBytes = pulsarConfig.getMaxBatchingBytes();
maxBatchingPublishDelayMillis =
pulsarConfig.getMaxBatchingPublishDelayMillis();
@@ -133,19 +145,20 @@ public class PulsarClientService {
/**
* send message
*
+ * @param poolIndex
* @param topic
* @param event
* @param sendMessageCallBack
* @param es
* @return
*/
- public boolean sendMessage(String topic, Event event,
+ public boolean sendMessage(int poolIndex, String topic, Event event,
SendMessageCallBack sendMessageCallBack,
EventStat es) {
- TopicProducerInfo producer = null;
+ TopicProducerInfo producerInfo = null;
final String inlongStreamId = getInlongStreamId(event);
final String inlongGroupId = getInlongGroupId(event);
try {
- producer = getProducer(topic, inlongGroupId, inlongStreamId);
+ producerInfo = getProducerInfo(poolIndex, topic, inlongGroupId,
inlongStreamId);
} catch (Exception e) {
if (logPrinterA.shouldPrint()) {
/*
@@ -165,15 +178,15 @@ public class PulsarClientService {
* If the producer is a null value,\ it means that the topic is not yet
* ready, and it needs to be played back into the file channel
*/
- if (producer == null) {
+ if (producerInfo == null) {
/*
* Data within 30s is placed in the exception channel to
* prevent frequent checks
* After 30s, reopen the topic check, if it is still a null value,
* put it back into the illegal map
*/
- sendMessageCallBack.handleMessageSendException(topic, es, new
Exception("producer is "
- + "null"));
+ sendMessageCallBack.handleMessageSendException(topic, es, new
Exception("producer "
+ + " info is null"));
return false;
}
@@ -181,12 +194,17 @@ public class PulsarClientService {
proMap.put("data_proxy_ip", localIp);
proMap.put(inlongStreamId,
event.getHeaders().get(ConfigConstants.PKG_TIME_KEY));
- TopicProducerInfo forCallBackP = producer;
-
+ TopicProducerInfo forCallBackP = producerInfo;
+ Producer producer = producerInfo.getProducer(poolIndex);
+ if (producer == null) {
+ sendMessageCallBack.handleMessageSendException(topic, es, new
Exception("producer is "
+ + "null"));
+ return false;
+ }
if (MessageUtils.isSyncSendForOrder(event) && (event instanceof
OrderEvent)) {
String partitionKey =
event.getHeaders().get(AttributeConstants.MESSAGE_PARTITION_KEY);
try {
- MessageId msgId =
forCallBackP.getProducer().newMessage().key(partitionKey)
+ MessageId msgId = producer.newMessage().key(partitionKey)
.properties(proMap).value(event.getBody())
.send();
sendResponse((OrderEvent)event);
@@ -204,7 +222,7 @@ public class PulsarClientService {
}
} else {
-
forCallBackP.getProducer().newMessage().properties(proMap).value(event.getBody())
+ producer.newMessage().properties(proMap).value(event.getBody())
.sendAsync().thenAccept((msgId) -> {
AuditUtils.add(AuditUtils.AUDIT_ID_DATAPROXY_SEND_SUCCESS,
event);
forCallBackP.setCanUseSend(true);
@@ -306,7 +324,8 @@ public class PulsarClientService {
if (pulsarClients != null) {
newList = new ArrayList<>();
for (PulsarClient pulsarClient : pulsarClients.values()) {
- TopicProducerInfo info = new
TopicProducerInfo(pulsarClient, topic);
+ TopicProducerInfo info = new
TopicProducerInfo(pulsarClient,
+ sinkThreadPoolSize, topic);
info.initProducer(inlongGroupId, inlongStreamId);
if (info.isCanUseToSendMessage()) {
newList.add(info);
@@ -325,7 +344,7 @@ public class PulsarClientService {
return initTopicProducer(topic, null, null);
}
- private TopicProducerInfo getProducer(String topic, String inlongGroupId,
+ private TopicProducerInfo getProducerInfo(int poolIndex, String topic,
String inlongGroupId,
String inlongStreamId) {
List<TopicProducerInfo> producerList =
initTopicProducer(topic, inlongGroupId, inlongStreamId);
@@ -341,7 +360,8 @@ public class PulsarClientService {
do {
int index = (int) (topicIndex.getAndIncrement() %
maxTryToGetProducer);
p = producerList.get(index);
- if (p.isCanUseToSendMessage() && p.getProducer().isConnected()) {
+ if (p.isCanUseToSendMessage() && p.getProducer(poolIndex) != null
+ && p.getProducer(poolIndex).isConnected()) {
break;
}
retryTime++;
@@ -422,7 +442,8 @@ public class PulsarClientService {
//create related topicProducers
for (String topic : topicSet) {
- TopicProducerInfo info = new TopicProducerInfo(client,
topic);
+ TopicProducerInfo info = new TopicProducerInfo(client,
sinkThreadPoolSize,
+ topic);
info.initProducer();
if (info.isCanUseToSendMessage()) {
producerInfoMap.computeIfAbsent(topic, k -> new
ArrayList<>()).add(info);
@@ -473,39 +494,45 @@ public class PulsarClientService {
class TopicProducerInfo {
private long lastSendMsgErrorTime;
- private Producer producer;
+ private Producer[] producers;
private PulsarClient pulsarClient;
+ private int sinkThreadPoolSize;
+
private String topic;
private volatile Boolean isCanUseSend = true;
private volatile Boolean isFinishInit = false;
- public TopicProducerInfo(PulsarClient pulsarClient,
+ public TopicProducerInfo(PulsarClient pulsarClient, int
sinkThreadPoolSize,
String topic) {
this.pulsarClient = pulsarClient;
+ this.sinkThreadPoolSize = sinkThreadPoolSize;
this.topic = topic;
+ this.producers = new Producer[sinkThreadPoolSize];
+ }
+
+ public void initProducer() {
+ initProducer(null, null);
}
public void initProducer(String inlongGroupId, String inlongStreamId) {
try {
- producer = pulsarClient.newProducer().sendTimeout(sendTimeout,
- TimeUnit.MILLISECONDS)
- .topic(topic)
- .enableBatching(enableBatch)
- .blockIfQueueFull(blockIfQueueFull)
- .maxPendingMessages(maxPendingMessages)
- .batchingMaxMessages(maxBatchingMessages)
- .batchingMaxBytes(maxBatchingBytes)
-
.batchingMaxPublishDelay(maxBatchingPublishDelayMillis, TimeUnit.MILLISECONDS)
- .create();
+ for (int i = 0; i < sinkThreadPoolSize; i++) {
+ producers[i] = createProducer();
+ }
isFinishInit = true;
} catch (PulsarClientException e) {
logger.error("create pulsar client has error e = {}
inlongGroupId = {}, "
+ "inlongStreamId= {}", e, inlongGroupId,
inlongStreamId);
isFinishInit = false;
+ for (int i = 0; i < sinkThreadPoolSize; i++) {
+ if (producers[i] != null) {
+ producers[i].closeAsync();
+ }
+ }
if (streamConfigLogMetric != null
&& StringUtils.isNotEmpty(inlongGroupId)
&& StringUtils.isNotEmpty(inlongStreamId)) {
@@ -516,8 +543,19 @@ public class PulsarClientService {
}
}
- public void initProducer() {
- initProducer(null, null);
+ private Producer createProducer() throws PulsarClientException {
+ return pulsarClient.newProducer().sendTimeout(sendTimeout,
+ TimeUnit.MILLISECONDS)
+ .topic(topic)
+ .enableBatching(enableBatch)
+ .blockIfQueueFull(blockIfQueueFull)
+ .maxPendingMessages(maxPendingMessages)
+
.maxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions)
+ .compressionType(compressionType)
+ .batchingMaxMessages(maxBatchingMessages)
+ .batchingMaxBytes(maxBatchingBytes)
+ .batchingMaxPublishDelay(maxBatchingPublishDelayMillis,
TimeUnit.MILLISECONDS)
+ .create();
}
public void setCanUseSend(Boolean isCanUseSend) {
@@ -540,14 +578,21 @@ public class PulsarClientService {
public void close() {
try {
- producer.close();
+ for (int i = 0; i < sinkThreadPoolSize; i++) {
+ if (producers[i] != null) {
+ producers[i].close();
+ }
+ }
} catch (PulsarClientException e) {
logger.error("close pulsar producer has error e = {}", e);
}
}
- public Producer getProducer() {
- return producer;
+ public Producer getProducer(int poolIndex) {
+ if (poolIndex >= sinkThreadPoolSize || producers[poolIndex] ==
null) {
+ return producers[0];
+ }
+ return producers[poolIndex];
}
public PulsarClient getPulsarClient() {
