You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "yfsn666 (via GitHub)" <gi...@apache.org> on 2024/08/21 04:37:57 UTC
[PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
yfsn666 opened a new pull request, #10837:
URL: https://github.com/apache/inlong/pull/10837
<!-- Prepare a Pull Request
Change the title of pull request refer to the following example:
[INLONG-XYZ][Component] Title of the pull request
-->
<!-- Specify the issue this pull request going to fix.
The following *XYZ* should be replaced by the actual [GitHub Issue](https://github.com/apache/inlong/issues) number)-->
Fixes #10831
### Motivation
<!--Explain here the context, and why you're making that change. What is the problem you're trying to solve.-->
Add a new sink type: `HTTP` for SortStandalone, which can report data to the specific HTTP interface. The flow diagram is shown below:
### Modifications
<!--Describe the modifications you've done.-->
Add a new sink type package `http` in `sort-standalone-source`.
### Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726165930
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSink.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpSink extends AbstractSink implements Configurable {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
+
+ private Context parentContext;
+ private BufferQueue<HttpRequest> dispatchQueue;
+ private HttpSinkContext context;
+ // workers
+ private List<HttpChannelWorker> workers = new ArrayList<>();
+ // output
+ private HttpOutputChannel outputChannel;
+
+ @Override
+ public void start() {
+ super.start();
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new HttpSinkContext(getName(), parentContext, getChannel(), dispatchQueue);
+ this.context.start();
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ HttpChannelWorker worker = new HttpChannelWorker(context, i);
+ this.workers.add(worker);
+ worker.start();
+ }
+ this.outputChannel = HttpSinkFactory.createHttpOutputChannel(context);
+ this.outputChannel.init();
+ this.outputChannel.start();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
Review Comment:
Please print out the specific error log operation for easy troubleshooting, instead of just printing exceptions.
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1732071295
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpIdConfig.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.field.FieldConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.sink.HttpSinkConfig;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public class HttpIdConfig extends IdConfig {
+
+ private String path;
+ private String method;
+ private Map<String, String> headers;
+ private Integer maxRetryTimes;
+ private String separator = "|";
+ private List<String> fieldList;
+
Review Comment:
fixed.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ public static final String HTTP_PREFIX = "http://";
+ public static final String HTTPS_PREFIX = "https://";
+ public static final String INLONG_GROUP_ID_HEADER = "inlongGroupId";
+ public static final String INLONG_STREAM_ID_HEADER = "inlongStreamId";
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724812825
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSinkContext.java:
##########
@@ -0,0 +1,405 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.ClusterTagConfig;
+import org.apache.inlong.common.pojo.sort.TaskConfig;
+import org.apache.inlong.common.pojo.sort.node.HttpNodeConfig;
+import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
+import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
+import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
+import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
+import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+import org.apache.inlong.sort.standalone.utils.Constants;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class HttpSinkContext extends SinkContext {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpSinkContext.class);
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String KEY_BASE_URL = "baseUrl";
+ public static final String KEY_ENABLE_CREDENTIAL = "enableCredential";
+ public static final String KEY_USERNAME = "username";
+ public static final String KEY_PASSWORD = "password";
+ public static final String KEY_MAX_CONNECT_TOTAL = "maxConnect";
+ public static final String KEY_MAX_CONNECT_PER_ROUTE = "maxConnectPerRoute";
+ public static final String KEY_CONNECTION_REQUEST_TIMEOUT = "connectionRequestTimeout";
+ public static final String KEY_SOCKET_TIMEOUT = "socketTimeout";
+ public static final String KEY_MAX_REDIRECTS = "maxRedirects";
+ public static final String KEY_LOG_MAX_LENGTH = "logMaxLength";
+ public static final String KEY_EVENT_HTTPREQUEST_HANDLER = "httpRequestHandler";
Review Comment:
```suggestion
public static final String KEY_EVENT_HTTP_REQUEST_HANDLER = "httpRequestHandler";
```
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726170330
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpOutputChannel.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+
+public class HttpOutputChannel extends Thread {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpOutputChannel.class);
+
+ private LifecycleState status;
+ private HttpSinkContext context;
+ private CloseableHttpAsyncClient httpClient;
+
+ public HttpOutputChannel(HttpSinkContext context) {
+ super(context.getTaskName());
+ this.context = context;
+ this.status = LifecycleState.IDLE;
+ }
+
+ public void init() {
+ initHttpClient();
+ }
+
+ private boolean initHttpClient() {
+ try {
+ if (httpClient == null) {
+ String userName = context.getUsername();
+ String password = context.getPassword();
+ LOG.info("initHttpAsyncClient:url:{}", context.getBaseUrl());
+
+ HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
+ final CredentialsProvider provider = new BasicCredentialsProvider();
+ if (context.getEnableCredential()) {
+ provider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(userName, password));
+ builder.setDefaultCredentialsProvider(provider);
+ }
+
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectionRequestTimeout(context.getConnectionRequestTimeout())
+ .setSocketTimeout(context.getSocketTimeout())
+ .setMaxRedirects(context.getMaxRedirects())
+ .setConnectTimeout(120 * 1000)
+ .build();
+
+ builder.setDefaultRequestConfig(requestConfig)
+ .setMaxConnTotal(context.getMaxConnect())
+ .setMaxConnPerRoute(context.getMaxConnectPerRoute());
+
+ httpClient = HttpSinkFactory.createHttpAsyncClient(builder);
+ httpClient.start();
+ }
+ } catch (Exception e) {
+ LOG.error("init httpclient failed.", e);
+ httpClient = null;
+ return false;
+ }
+ return true;
+ }
+
+ public void close() {
+ status = LifecycleState.STOP;
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ LOG.error(String.format("close HttpClient:%s", e.getMessage()), e);
+ }
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpOutputChannel:{},status:{}", context.getTaskName(), status);
+ while (status == LifecycleState.START) {
+ try {
+ send();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
+ }
+ }
+ }
+
+ public void send() throws InterruptedException {
+ HttpRequest httpRequest = null;
+ try {
+ // get httpRequest
+ httpRequest = context.takeDispatchQueue();
+ if (httpRequest == null) {
+ Thread.sleep(context.getProcessInterval());
+ return;
+ }
+ // get id config
+ String uid = httpRequest.getEvent().getUid();
+ if (context.getIdConfig(uid) == null) {
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ return;
+ }
+ // send
+ httpClient.execute(httpRequest.getRequest(), new HttpCallback(context, httpRequest));
+ context.addSendMetric(httpRequest.getEvent(), context.getTaskName());
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ if (httpRequest != null) {
+ context.backDispatchQueue(httpRequest);
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ }
+ try {
+ Thread.sleep(context.getProcessInterval());
+ } catch (InterruptedException e1) {
Review Comment:
What does e1 mean?
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpOutputChannel.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+
+public class HttpOutputChannel extends Thread {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpOutputChannel.class);
+
+ private LifecycleState status;
+ private HttpSinkContext context;
+ private CloseableHttpAsyncClient httpClient;
+
+ public HttpOutputChannel(HttpSinkContext context) {
+ super(context.getTaskName());
+ this.context = context;
+ this.status = LifecycleState.IDLE;
+ }
+
+ public void init() {
+ initHttpClient();
+ }
+
+ private boolean initHttpClient() {
+ try {
+ if (httpClient == null) {
+ String userName = context.getUsername();
+ String password = context.getPassword();
+ LOG.info("initHttpAsyncClient:url:{}", context.getBaseUrl());
+
+ HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
+ final CredentialsProvider provider = new BasicCredentialsProvider();
+ if (context.getEnableCredential()) {
+ provider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(userName, password));
+ builder.setDefaultCredentialsProvider(provider);
+ }
+
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectionRequestTimeout(context.getConnectionRequestTimeout())
+ .setSocketTimeout(context.getSocketTimeout())
+ .setMaxRedirects(context.getMaxRedirects())
+ .setConnectTimeout(120 * 1000)
+ .build();
+
+ builder.setDefaultRequestConfig(requestConfig)
+ .setMaxConnTotal(context.getMaxConnect())
+ .setMaxConnPerRoute(context.getMaxConnectPerRoute());
+
+ httpClient = HttpSinkFactory.createHttpAsyncClient(builder);
+ httpClient.start();
+ }
+ } catch (Exception e) {
+ LOG.error("init httpclient failed.", e);
+ httpClient = null;
+ return false;
+ }
+ return true;
+ }
+
+ public void close() {
+ status = LifecycleState.STOP;
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ LOG.error(String.format("close HttpClient:%s", e.getMessage()), e);
+ }
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpOutputChannel:{},status:{}", context.getTaskName(), status);
+ while (status == LifecycleState.START) {
+ try {
+ send();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
+ }
+ }
+ }
+
+ public void send() throws InterruptedException {
+ HttpRequest httpRequest = null;
+ try {
+ // get httpRequest
+ httpRequest = context.takeDispatchQueue();
+ if (httpRequest == null) {
+ Thread.sleep(context.getProcessInterval());
+ return;
+ }
+ // get id config
+ String uid = httpRequest.getEvent().getUid();
+ if (context.getIdConfig(uid) == null) {
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ return;
+ }
+ // send
+ httpClient.execute(httpRequest.getRequest(), new HttpCallback(context, httpRequest));
+ context.addSendMetric(httpRequest.getEvent(), context.getTaskName());
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
Review Comment:
Ditto.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSink.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpSink extends AbstractSink implements Configurable {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
+
+ private Context parentContext;
+ private BufferQueue<HttpRequest> dispatchQueue;
+ private HttpSinkContext context;
+ // workers
+ private List<HttpChannelWorker> workers = new ArrayList<>();
+ // output
+ private HttpOutputChannel outputChannel;
+
+ @Override
+ public void start() {
+ super.start();
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new HttpSinkContext(getName(), parentContext, getChannel(), dispatchQueue);
+ this.context.start();
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ HttpChannelWorker worker = new HttpChannelWorker(context, i);
+ this.workers.add(worker);
+ worker.start();
+ }
+ this.outputChannel = HttpSinkFactory.createHttpOutputChannel(context);
+ this.outputChannel.init();
+ this.outputChannel.start();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void stop() {
+ super.stop();
+ try {
+ this.context.close();
+ for (HttpChannelWorker worker : this.workers) {
+ worker.close();
+ }
+ this.workers.clear();
+ this.outputChannel.close();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
Review Comment:
Ditto.
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726170330
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpOutputChannel.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+
+public class HttpOutputChannel extends Thread {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpOutputChannel.class);
+
+ private LifecycleState status;
+ private HttpSinkContext context;
+ private CloseableHttpAsyncClient httpClient;
+
+ public HttpOutputChannel(HttpSinkContext context) {
+ super(context.getTaskName());
+ this.context = context;
+ this.status = LifecycleState.IDLE;
+ }
+
+ public void init() {
+ initHttpClient();
+ }
+
+ private boolean initHttpClient() {
+ try {
+ if (httpClient == null) {
+ String userName = context.getUsername();
+ String password = context.getPassword();
+ LOG.info("initHttpAsyncClient:url:{}", context.getBaseUrl());
+
+ HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
+ final CredentialsProvider provider = new BasicCredentialsProvider();
+ if (context.getEnableCredential()) {
+ provider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(userName, password));
+ builder.setDefaultCredentialsProvider(provider);
+ }
+
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectionRequestTimeout(context.getConnectionRequestTimeout())
+ .setSocketTimeout(context.getSocketTimeout())
+ .setMaxRedirects(context.getMaxRedirects())
+ .setConnectTimeout(120 * 1000)
+ .build();
+
+ builder.setDefaultRequestConfig(requestConfig)
+ .setMaxConnTotal(context.getMaxConnect())
+ .setMaxConnPerRoute(context.getMaxConnectPerRoute());
+
+ httpClient = HttpSinkFactory.createHttpAsyncClient(builder);
+ httpClient.start();
+ }
+ } catch (Exception e) {
+ LOG.error("init httpclient failed.", e);
+ httpClient = null;
+ return false;
+ }
+ return true;
+ }
+
+ public void close() {
+ status = LifecycleState.STOP;
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ LOG.error(String.format("close HttpClient:%s", e.getMessage()), e);
+ }
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpOutputChannel:{},status:{}", context.getTaskName(), status);
+ while (status == LifecycleState.START) {
+ try {
+ send();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
+ }
+ }
+ }
+
+ public void send() throws InterruptedException {
+ HttpRequest httpRequest = null;
+ try {
+ // get httpRequest
+ httpRequest = context.takeDispatchQueue();
+ if (httpRequest == null) {
+ Thread.sleep(context.getProcessInterval());
+ return;
+ }
+ // get id config
+ String uid = httpRequest.getEvent().getUid();
+ if (context.getIdConfig(uid) == null) {
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ return;
+ }
+ // send
+ httpClient.execute(httpRequest.getRequest(), new HttpCallback(context, httpRequest));
+ context.addSendMetric(httpRequest.getEvent(), context.getTaskName());
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ if (httpRequest != null) {
+ context.backDispatchQueue(httpRequest);
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ }
+ try {
+ Thread.sleep(context.getProcessInterval());
+ } catch (InterruptedException e1) {
Review Comment:
What does `e1` mean?
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726167323
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
Review Comment:
Please set the string used multiple times as a constant, like `public static final String HTTP_PREFIX = "http://";
`
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726529262
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
Review Comment:
The `separator` in `idConfig` is a `String` type, but the separator required in the function `UnescapeHelper.toFiledList` is a `char` type, so here `cDelimiter` is the converted separator form the `separator` in `idConfig`.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ switch (idConfig.getMethod().toUpperCase()) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey(), StandardCharsets.UTF_8.name()) + "="
+ + URLEncoder.encode(entry.getValue(), StandardCharsets.UTF_8.name());
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "luchunliang (via GitHub)" <gi...@apache.org>.
luchunliang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1732046068
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ public static final String HTTP_PREFIX = "http://";
+ public static final String HTTPS_PREFIX = "https://";
+ public static final String INLONG_GROUP_ID_HEADER = "inlongGroupId";
+ public static final String INLONG_STREAM_ID_HEADER = "inlongStreamId";
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith(HTTP_PREFIX) && !uriString.startsWith(HTTPS_PREFIX)) {
+ uriString = HTTP_PREFIX + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ String requestMethod = idConfig.getMethod().toUpperCase();
+ switch (requestMethod) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey() + "=" + entry.getValue(),
+ StandardCharsets.UTF_8.name());
Review Comment:
Please use charset of SinkConfig.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ public static final String HTTP_PREFIX = "http://";
+ public static final String HTTPS_PREFIX = "https://";
+ public static final String INLONG_GROUP_ID_HEADER = "inlongGroupId";
+ public static final String INLONG_STREAM_ID_HEADER = "inlongStreamId";
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
Review Comment:
Please use charset of InLong data stream in DataFlowConfig.SourceConfig(encodingType).
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ public static final String HTTP_PREFIX = "http://";
+ public static final String HTTPS_PREFIX = "https://";
+ public static final String INLONG_GROUP_ID_HEADER = "inlongGroupId";
+ public static final String INLONG_STREAM_ID_HEADER = "inlongStreamId";
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith(HTTP_PREFIX) && !uriString.startsWith(HTTPS_PREFIX)) {
+ uriString = HTTP_PREFIX + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ String requestMethod = idConfig.getMethod().toUpperCase();
+ switch (requestMethod) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey() + "=" + entry.getValue(),
Review Comment:
Just need to encode the value, no need for key.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpIdConfig.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.field.FieldConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.sink.HttpSinkConfig;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public class HttpIdConfig extends IdConfig {
+
+ private String path;
+ private String method;
+ private Map<String, String> headers;
+ private Integer maxRetryTimes;
+ private String separator = "|";
+ private List<String> fieldList;
+
Review Comment:
Miss charset field. Please check EsIdConfig java file.
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726167552
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ switch (idConfig.getMethod().toUpperCase()) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey(), StandardCharsets.UTF_8.name()) + "="
+ + URLEncoder.encode(entry.getValue(), StandardCharsets.UTF_8.name());
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(Collectors.joining("&"));
+ request = new HttpGet(uri + "?" + params);
+ for (Map.Entry<String, String> entry : idConfig.getHeaders().entrySet()) {
+ request.setHeader(entry.getKey(), entry.getValue());
+ }
+ request.setHeader("InlongGroupID", idConfig.getInlongGroupId());
Review Comment:
Ditto.
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724811935
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSink.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpSink extends AbstractSink implements Configurable {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
+
+ private Context parentContext;
+ private BufferQueue<HttpRequest> dispatchQueue;
+ private HttpSinkContext context;
+ // workers
+ private List<HttpChannelWorker> workers = new ArrayList<>();
+ // output
+ private HttpOutputChannel outputChannel;
+
+ @Override
+ public void start() {
+ super.start();
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new HttpSinkContext(getName(), parentContext, getChannel(), dispatchQueue);
+ this.context.start();
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ HttpChannelWorker worker = new HttpChannelWorker(context, i);
+ this.workers.add(worker);
+ worker.start();
+ }
+ this.outputChannel = HttpSinkFactory.createHttpOutputChannel(context);
+ this.outputChannel.init();
+ this.outputChannel.start();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void stop() {
+ super.stop();
+ try {
+ this.context.close();
+ for (HttpChannelWorker worker : this.workers) {
+ worker.close();
+ }
+ this.workers.clear();
+ this.outputChannel.close();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void configure(Context context) {
+ LOG.info("start to configure:{}, context:{}.", this.getName(), context.toString());
Review Comment:
```suggestion
LOG.info("Start to configure:{}, context:{}.", this.getName(), context.toString());
```
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726516797
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSink.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpSink extends AbstractSink implements Configurable {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
+
+ private Context parentContext;
+ private BufferQueue<HttpRequest> dispatchQueue;
+ private HttpSinkContext context;
+ // workers
+ private List<HttpChannelWorker> workers = new ArrayList<>();
+ // output
+ private HttpOutputChannel outputChannel;
+
+ @Override
+ public void start() {
+ super.start();
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new HttpSinkContext(getName(), parentContext, getChannel(), dispatchQueue);
+ this.context.start();
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ HttpChannelWorker worker = new HttpChannelWorker(context, i);
+ this.workers.add(worker);
+ worker.start();
+ }
+ this.outputChannel = HttpSinkFactory.createHttpOutputChannel(context);
+ this.outputChannel.init();
+ this.outputChannel.start();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void stop() {
+ super.stop();
+ try {
+ this.context.close();
+ for (HttpChannelWorker worker : this.workers) {
+ worker.close();
+ }
+ this.workers.clear();
+ this.outputChannel.close();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724805038
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpCallback.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.concurrent.FutureCallback;
+import org.slf4j.Logger;
+
+public class HttpCallback implements FutureCallback<HttpResponse> {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpCallback.class);
+
+ private HttpSinkContext context;
+ private HttpRequest requestItem;
+
+ public HttpCallback(HttpSinkContext context, HttpRequest requestItem) {
+ this.context = context;
+ this.requestItem = requestItem;
+ }
+
+ @Override
+ public void completed(HttpResponse httpResponse) {
+ int statusCode = httpResponse.getStatusLine().getStatusCode();
+ ProfileEvent event = requestItem.getEvent();
+ long sendTime = requestItem.getSendTime();
+
+ // is fail
+ if (statusCode != 200) {
+ context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
+ // if reach the max retry times, release the request
+ int remainRetryTimes = requestItem.getRemainRetryTimes();
+ if (remainRetryTimes == 1) {
+ context.releaseDispatchQueue(requestItem);
+ return;
+ }
+ if (remainRetryTimes > 1) {
+ requestItem.setRemainRetryTimes(remainRetryTimes - 1);
+ }
+ context.backDispatchQueue(requestItem);
+ } else {
+ context.addSendResultMetric(event, context.getTaskName(), true, sendTime);
+ context.releaseDispatchQueue(requestItem);
+ event.ack();
+ }
+ }
+
+ @Override
+ public void failed(Exception e) {
+ LOG.error("Http request failed,errorMsg:" + e.getMessage(), e);
Review Comment:
```suggestion
LOG.error("Http request failed,errorMsg:{}", e.getMessage(), e);
```
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725052045
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData = new ObjectMapper().writeValueAsString(fieldMap);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724815772
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSinkContext.java:
##########
@@ -0,0 +1,405 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.ClusterTagConfig;
+import org.apache.inlong.common.pojo.sort.TaskConfig;
+import org.apache.inlong.common.pojo.sort.node.HttpNodeConfig;
+import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
+import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
+import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
+import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
+import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+import org.apache.inlong.sort.standalone.utils.Constants;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class HttpSinkContext extends SinkContext {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpSinkContext.class);
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String KEY_BASE_URL = "baseUrl";
+ public static final String KEY_ENABLE_CREDENTIAL = "enableCredential";
+ public static final String KEY_USERNAME = "username";
+ public static final String KEY_PASSWORD = "password";
+ public static final String KEY_MAX_CONNECT_TOTAL = "maxConnect";
+ public static final String KEY_MAX_CONNECT_PER_ROUTE = "maxConnectPerRoute";
+ public static final String KEY_CONNECTION_REQUEST_TIMEOUT = "connectionRequestTimeout";
+ public static final String KEY_SOCKET_TIMEOUT = "socketTimeout";
+ public static final String KEY_MAX_REDIRECTS = "maxRedirects";
+ public static final String KEY_LOG_MAX_LENGTH = "logMaxLength";
+ public static final String KEY_EVENT_HTTPREQUEST_HANDLER = "httpRequestHandler";
+
+ public static final boolean DEFAULT_ENABLE_CREDENTIAL = false;
+ public static final int DEFAULT_MAX_CONNECT_TOTAL = 1000;
+ public static final int DEFAULT_MAX_CONNECT_PER_ROUTE = 1000;
+ public static final int DEFAULT_CONNECTION_REQUEST_TIMEOUT = 0;
+ public static final int DEFAULT_SOCKET_TIMEOUT = 0;
+ public static final int DEFAULT_MAX_REDIRECTS = 0;
+ public static final int DEFAULT_LOG_MAX_LENGTH = 32 * 1024;
+
+ private Context sinkContext;
+ private HttpNodeConfig httpNodeConfig;
+ private String nodeId;
+ private Map<String, HttpIdConfig> idConfigMap = new ConcurrentHashMap<>();
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private final BufferQueue<HttpRequest> dispatchQueue;
+ private AtomicLong offerCounter = new AtomicLong(0);
+ private AtomicLong takeCounter = new AtomicLong(0);
+ private AtomicLong backCounter = new AtomicLong(0);
+ // rest client
+ private String baseUrl;
+ private boolean enableCredential;
+ private String username;
+ private String password;
+ private int maxConnect = DEFAULT_MAX_CONNECT_TOTAL;
+ private int maxConnectPerRoute = DEFAULT_MAX_CONNECT_PER_ROUTE;
+ private int connectionRequestTimeout = DEFAULT_CONNECTION_REQUEST_TIMEOUT;
+ private int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
+ private int maxRedirects = DEFAULT_MAX_REDIRECTS;
+ private int logMaxLength = DEFAULT_LOG_MAX_LENGTH;
+
+ public HttpSinkContext(String sinkName, Context context, Channel channel,
+ BufferQueue<HttpRequest> dispatchQueue) {
+ super(sinkName, context, channel);
+ this.sinkContext = context;
+ this.dispatchQueue = dispatchQueue;
+ this.nodeId = CommonPropertiesHolder.getString(KEY_NODE_ID);
+ }
+
+ public void reload() {
+ try {
+ LOG.info("SortTask:{},dispatchQueue:{},offer:{},take:{},back:{}",
+ taskName, dispatchQueue.size(), offerCounter.getAndSet(0),
+ takeCounter.getAndSet(0), backCounter.getAndSet(0));
+ TaskConfig newTaskConfig = SortConfigHolder.getTaskConfig(taskName);
+ SortTaskConfig newSortTaskConfig = SortClusterConfigHolder.getTaskConfig(taskName);
+ if ((newTaskConfig == null || newTaskConfig.equals(taskConfig))
+ && (newSortTaskConfig == null || newSortTaskConfig.equals(sortTaskConfig))) {
+ return;
+ }
+ LOG.info("get new SortTaskConfig:taskName:{}", taskName);
+
+ if (newTaskConfig != null) {
+ HttpNodeConfig requestNodeConfig = (HttpNodeConfig) newTaskConfig.getNodeConfig();
+ if (httpNodeConfig == null || requestNodeConfig.getVersion() > httpNodeConfig.getVersion()) {
+ this.httpNodeConfig = requestNodeConfig;
+ }
+ }
+
+ this.taskConfig = newTaskConfig;
+ this.sortTaskConfig = newSortTaskConfig;
+
+ // change current config
+ Map<String, HttpIdConfig> fromTaskConfig = reloadIdParamsFromTaskConfig(taskConfig);
+ Map<String, HttpIdConfig> fromSortTaskConfig = reloadIdParamsFromSortTaskConfig(sortTaskConfig);
+ if (unifiedConfiguration) {
+ idConfigMap = fromTaskConfig;
+ reloadClientsFromNodeConfig(httpNodeConfig);
+ } else {
+ idConfigMap = fromSortTaskConfig;
+ reloadClientsFromSortTaskConfig(sortTaskConfig);
+ }
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
+ // log
+ LOG.info("end to get SortTaskConfig:taskName:{}:newIdConfigMap:{}", taskName,
+ objectMapper.writeValueAsString(idConfigMap));
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ private Map<String, HttpIdConfig> reloadIdParamsFromTaskConfig(TaskConfig taskConfig) {
+ if (taskConfig == null) {
+ return new HashMap<>();
+ }
+ return taskConfig.getClusterTagConfigs()
+ .stream()
+ .map(ClusterTagConfig::getDataFlowConfigs)
+ .flatMap(Collection::stream)
+ .map(HttpIdConfig::create)
+ .collect(Collectors.toMap(
+ config -> InlongId.generateUid(config.getInlongGroupId(), config.getInlongStreamId()),
+ v -> v,
+ (flow1, flow2) -> flow1));
+ }
+
+ private Map<String, HttpIdConfig> reloadIdParamsFromSortTaskConfig(SortTaskConfig sortTaskConfig)
+ throws JsonProcessingException {
+ if (sortTaskConfig == null) {
+ return new HashMap<>();
+ }
+ Map<String, HttpIdConfig> newIdConfigMap = new ConcurrentHashMap<>();
+ List<Map<String, String>> idList = this.sortTaskConfig.getIdParams();
+ ObjectMapper objectMapper = new ObjectMapper();
+ objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ for (Map<String, String> idParam : idList) {
+ String inlongGroupId = idParam.get(Constants.INLONG_GROUP_ID);
+ String inlongStreamId = idParam.get(Constants.INLONG_STREAM_ID);
+ String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
+ String jsonIdConfig = objectMapper.writeValueAsString(idParam);
+ HttpIdConfig idConfig = objectMapper.readValue(jsonIdConfig, HttpIdConfig.class);
+ newIdConfigMap.put(uid, idConfig);
+ }
+ return newIdConfigMap;
+ }
+
+ private void reloadClientsFromNodeConfig(HttpNodeConfig httpNodeConfig) {
+ Map<String, String> properties = httpNodeConfig.getProperties();
+ this.sinkContext = new Context(properties != null ? properties : new HashMap<>());
+ this.baseUrl = httpNodeConfig.getBaseUrl();
+ this.enableCredential = httpNodeConfig.getEnableCredential();;
+ this.username = httpNodeConfig.getUsername();
+ this.password = httpNodeConfig.getPassword();
+ this.maxConnect = httpNodeConfig.getMaxConnect();
+
+ this.maxConnectPerRoute = sinkContext.getInteger(KEY_MAX_CONNECT_PER_ROUTE, DEFAULT_MAX_CONNECT_PER_ROUTE);
+ this.connectionRequestTimeout =
+ sinkContext.getInteger(KEY_CONNECTION_REQUEST_TIMEOUT, DEFAULT_CONNECTION_REQUEST_TIMEOUT);
+ this.socketTimeout = sinkContext.getInteger(KEY_SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT);
+ this.maxRedirects = sinkContext.getInteger(KEY_MAX_REDIRECTS, DEFAULT_MAX_REDIRECTS);
+ this.logMaxLength = sinkContext.getInteger(KEY_LOG_MAX_LENGTH, DEFAULT_LOG_MAX_LENGTH);
+ }
+
+ private void reloadClientsFromSortTaskConfig(SortTaskConfig sortTaskConfig) {
+ this.sinkContext = new Context(sortTaskConfig.getSinkParams());
+ this.baseUrl = sinkContext.getString(KEY_BASE_URL);
+ this.enableCredential = sinkContext.getBoolean(KEY_ENABLE_CREDENTIAL, DEFAULT_ENABLE_CREDENTIAL);
+ this.username = sinkContext.getString(KEY_USERNAME);
+ this.password = sinkContext.getString(KEY_PASSWORD);
+ this.maxConnect = sinkContext.getInteger(KEY_MAX_CONNECT_TOTAL, DEFAULT_MAX_CONNECT_TOTAL);
+
+ this.maxConnectPerRoute = sinkContext.getInteger(KEY_MAX_CONNECT_PER_ROUTE, DEFAULT_MAX_CONNECT_PER_ROUTE);
+ this.connectionRequestTimeout =
+ sinkContext.getInteger(KEY_CONNECTION_REQUEST_TIMEOUT, DEFAULT_CONNECTION_REQUEST_TIMEOUT);
+ this.socketTimeout = sinkContext.getInteger(KEY_SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT);
+ this.maxRedirects = sinkContext.getInteger(KEY_MAX_REDIRECTS, DEFAULT_MAX_REDIRECTS);
+ this.logMaxLength = sinkContext.getInteger(KEY_LOG_MAX_LENGTH, DEFAULT_LOG_MAX_LENGTH);
+ }
+
+ public void addSendMetric(ProfileEvent currentRecord, String bid) {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(SortMetricItem.KEY_CLUSTER_ID, this.getClusterId());
+ dimensions.put(SortMetricItem.KEY_TASK_NAME, this.getTaskName());
+ // metric
+ fillInlongId(currentRecord, dimensions);
+ dimensions.put(SortMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(SortMetricItem.KEY_SINK_DATA_ID, bid);
+ long msgTime = currentRecord.getRawLogTime();
+ long auditFormatTime = msgTime - msgTime % CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(SortMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
+ SortMetricItem metricItem = this.getMetricItemSet().findMetricItem(dimensions);
+ long count = 1;
+ long size = currentRecord.getBody().length;
+ metricItem.sendCount.addAndGet(count);
+ metricItem.sendSize.addAndGet(size);
+ }
+
+ public void addSendFailMetric() {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(SortMetricItem.KEY_CLUSTER_ID, this.getClusterId());
+ dimensions.put(SortMetricItem.KEY_SINK_ID, this.getSinkName());
+ long msgTime = System.currentTimeMillis();
+ long auditFormatTime = msgTime - msgTime % CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(SortMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
+ SortMetricItem metricItem = this.getMetricItemSet().findMetricItem(dimensions);
+ metricItem.readFailCount.incrementAndGet();
+ }
+
+ public void addSendResultMetric(ProfileEvent currentRecord, String bid, boolean result, long sendTime) {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(SortMetricItem.KEY_CLUSTER_ID, this.getClusterId());
+ dimensions.put(SortMetricItem.KEY_TASK_NAME, this.getTaskName());
+ // metric
+ fillInlongId(currentRecord, dimensions);
+ dimensions.put(SortMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(SortMetricItem.KEY_SINK_DATA_ID, bid);
+ final long currentTime = System.currentTimeMillis();
+ long msgTime = currentRecord.getRawLogTime();
+ long auditFormatTime = msgTime - msgTime % CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(SortMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
+ SortMetricItem metricItem = this.getMetricItemSet().findMetricItem(dimensions);
+ if (result) {
+ metricItem.sendSuccessCount.incrementAndGet();
+ metricItem.sendSuccessSize.addAndGet(currentRecord.getBody().length);
+ AuditUtils.add(AuditUtils.AUDIT_ID_SEND_SUCCESS, currentRecord);
+ if (sendTime > 0) {
+ long sinkDuration = currentTime - sendTime;
+ long nodeDuration = currentTime - currentRecord.getFetchTime();
+ long wholeDuration = currentTime - currentRecord.getRawLogTime();
+ metricItem.sinkDuration.addAndGet(sinkDuration);
+ metricItem.nodeDuration.addAndGet(nodeDuration);
+ metricItem.wholeDuration.addAndGet(wholeDuration);
+ }
+ } else {
+ metricItem.sendFailCount.incrementAndGet();
+ metricItem.sendFailSize.addAndGet(currentRecord.getBody().length);
+ }
+ }
+
+ public HttpIdConfig getIdConfig(String uid) {
+ return this.idConfigMap.get(uid);
+ }
+
+ public String getNodeId() {
+ return nodeId;
+ }
+
+ public Map<String, HttpIdConfig> getIdConfigMap() {
+ return idConfigMap;
+ }
+
+ public Context getSinkContext() {
+ return sinkContext;
+ }
+
+ public void setSinkContext(Context sinkContext) {
+ this.sinkContext = sinkContext;
+ }
+
+ public void offerDispatchQueue(HttpRequest httpRequest) {
+ this.offerCounter.incrementAndGet();
+ dispatchQueue.acquire(httpRequest.getEvent().getBody().length);
+ dispatchQueue.offer(httpRequest);
+ }
+
+ public HttpRequest takeDispatchQueue() {
+ HttpRequest httpRequest = this.dispatchQueue.pollRecord();
+ if (httpRequest != null) {
+ this.takeCounter.incrementAndGet();
+ }
+ return httpRequest;
+ }
+
+ public void backDispatchQueue(HttpRequest httpRequest) {
+ this.backCounter.incrementAndGet();
+ dispatchQueue.offer(httpRequest);
+ }
+
+ public void releaseDispatchQueue(HttpRequest httpRequest) {
+ dispatchQueue.release(httpRequest.getEvent().getBody().length);
+ }
+
+ public String getBaseUrl() {
+ return baseUrl;
+ }
+
+ public void setBaseUrl(String baseUrl) {
+ this.baseUrl = baseUrl;
+ }
+
+ public boolean getEnableCredential() {
+ return enableCredential;
+ }
+
+ public void setEnableCredential(boolean enableCredential) {
+ this.enableCredential = enableCredential;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public int getMaxConnect() {
+ return maxConnect;
+ }
+
+ public int getMaxConnectPerRoute() {
+ return maxConnectPerRoute;
+ }
+
+ public int getConnectionRequestTimeout() {
+ return connectionRequestTimeout;
+ }
+
+ public int getSocketTimeout() {
+ return socketTimeout;
+ }
+
+ public int getMaxRedirects() {
+ return maxRedirects;
+ }
+
+ public int getLogMaxLength() {
+ return logMaxLength;
+ }
+
+ public void setMaxConnect(int maxConnect) {
+ this.maxConnect = maxConnect;
+ }
+
+ public void setNodeId(String nodeId) {
+ this.nodeId = nodeId;
+ }
+
+ public void setIdConfigMap(Map<String, HttpIdConfig> idConfigMap) {
+ this.idConfigMap = idConfigMap;
+ }
+
+ public IEvent2HttpRequestHandler createHttpRequestHandler() {
+ // IEvent2HttpRequestHandler
+ String httpRequestHandlerClass = CommonPropertiesHolder.getString(KEY_EVENT_HTTPREQUEST_HANDLER,
+ DefaultEvent2HttpRequestHandler.class.getName());
+ try {
+ Class<?> handlerClass = ClassUtils.getClass(httpRequestHandlerClass);
+ Object handlerObject = handlerClass.getDeclaredConstructor().newInstance();
+ if (handlerObject instanceof IEvent2HttpRequestHandler) {
+ IEvent2HttpRequestHandler handler = (IEvent2HttpRequestHandler) handlerObject;
+ return handler;
Review Comment:
```suggestion
return (IEvent2HttpRequestHandler) handlerObject;
```
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724809315
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpChannelWorker.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Event;
+import org.apache.flume.Transaction;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HttpChannelWorker extends Thread {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpChannelWorker.class);
+
+ private final HttpSinkContext context;
+ private final int workerIndex;
+
+ private LifecycleState status;
+ private IEvent2HttpRequestHandler handler;
+
+ public HttpChannelWorker(HttpSinkContext context, int workerIndex) {
+ this.context = context;
+ this.workerIndex = workerIndex;
+ this.status = LifecycleState.IDLE;
+ this.handler = context.createHttpRequestHandler();
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpChannelWorker:{},status:{},index:{}", context.getTaskName(), status, workerIndex);
+ while (status == LifecycleState.START) {
+ try {
+ this.doRun();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
Review Comment:
add some meaningful logs
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "vernedeng (via GitHub)" <gi...@apache.org>.
vernedeng commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724747158
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData = new ObjectMapper().writeValueAsString(fieldMap);
Review Comment:
no need for GET request, please init jsonData inside other case.
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725054250
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSinkContext.java:
##########
@@ -0,0 +1,405 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.ClusterTagConfig;
+import org.apache.inlong.common.pojo.sort.TaskConfig;
+import org.apache.inlong.common.pojo.sort.node.HttpNodeConfig;
+import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
+import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
+import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
+import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
+import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+import org.apache.inlong.sort.standalone.utils.Constants;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class HttpSinkContext extends SinkContext {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpSinkContext.class);
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String KEY_BASE_URL = "baseUrl";
+ public static final String KEY_ENABLE_CREDENTIAL = "enableCredential";
+ public static final String KEY_USERNAME = "username";
+ public static final String KEY_PASSWORD = "password";
+ public static final String KEY_MAX_CONNECT_TOTAL = "maxConnect";
+ public static final String KEY_MAX_CONNECT_PER_ROUTE = "maxConnectPerRoute";
+ public static final String KEY_CONNECTION_REQUEST_TIMEOUT = "connectionRequestTimeout";
+ public static final String KEY_SOCKET_TIMEOUT = "socketTimeout";
+ public static final String KEY_MAX_REDIRECTS = "maxRedirects";
+ public static final String KEY_LOG_MAX_LENGTH = "logMaxLength";
+ public static final String KEY_EVENT_HTTPREQUEST_HANDLER = "httpRequestHandler";
+
+ public static final boolean DEFAULT_ENABLE_CREDENTIAL = false;
+ public static final int DEFAULT_MAX_CONNECT_TOTAL = 1000;
+ public static final int DEFAULT_MAX_CONNECT_PER_ROUTE = 1000;
+ public static final int DEFAULT_CONNECTION_REQUEST_TIMEOUT = 0;
+ public static final int DEFAULT_SOCKET_TIMEOUT = 0;
+ public static final int DEFAULT_MAX_REDIRECTS = 0;
+ public static final int DEFAULT_LOG_MAX_LENGTH = 32 * 1024;
+
+ private Context sinkContext;
+ private HttpNodeConfig httpNodeConfig;
+ private String nodeId;
+ private Map<String, HttpIdConfig> idConfigMap = new ConcurrentHashMap<>();
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private final BufferQueue<HttpRequest> dispatchQueue;
+ private AtomicLong offerCounter = new AtomicLong(0);
+ private AtomicLong takeCounter = new AtomicLong(0);
+ private AtomicLong backCounter = new AtomicLong(0);
+ // rest client
+ private String baseUrl;
+ private boolean enableCredential;
+ private String username;
+ private String password;
+ private int maxConnect = DEFAULT_MAX_CONNECT_TOTAL;
+ private int maxConnectPerRoute = DEFAULT_MAX_CONNECT_PER_ROUTE;
+ private int connectionRequestTimeout = DEFAULT_CONNECTION_REQUEST_TIMEOUT;
+ private int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
+ private int maxRedirects = DEFAULT_MAX_REDIRECTS;
+ private int logMaxLength = DEFAULT_LOG_MAX_LENGTH;
+
+ public HttpSinkContext(String sinkName, Context context, Channel channel,
+ BufferQueue<HttpRequest> dispatchQueue) {
+ super(sinkName, context, channel);
+ this.sinkContext = context;
+ this.dispatchQueue = dispatchQueue;
+ this.nodeId = CommonPropertiesHolder.getString(KEY_NODE_ID);
+ }
+
+ public void reload() {
+ try {
+ LOG.info("SortTask:{},dispatchQueue:{},offer:{},take:{},back:{}",
+ taskName, dispatchQueue.size(), offerCounter.getAndSet(0),
+ takeCounter.getAndSet(0), backCounter.getAndSet(0));
+ TaskConfig newTaskConfig = SortConfigHolder.getTaskConfig(taskName);
+ SortTaskConfig newSortTaskConfig = SortClusterConfigHolder.getTaskConfig(taskName);
+ if ((newTaskConfig == null || newTaskConfig.equals(taskConfig))
+ && (newSortTaskConfig == null || newSortTaskConfig.equals(sortTaskConfig))) {
+ return;
+ }
+ LOG.info("get new SortTaskConfig:taskName:{}", taskName);
+
+ if (newTaskConfig != null) {
+ HttpNodeConfig requestNodeConfig = (HttpNodeConfig) newTaskConfig.getNodeConfig();
+ if (httpNodeConfig == null || requestNodeConfig.getVersion() > httpNodeConfig.getVersion()) {
+ this.httpNodeConfig = requestNodeConfig;
+ }
+ }
+
+ this.taskConfig = newTaskConfig;
+ this.sortTaskConfig = newSortTaskConfig;
+
+ // change current config
+ Map<String, HttpIdConfig> fromTaskConfig = reloadIdParamsFromTaskConfig(taskConfig);
+ Map<String, HttpIdConfig> fromSortTaskConfig = reloadIdParamsFromSortTaskConfig(sortTaskConfig);
+ if (unifiedConfiguration) {
+ idConfigMap = fromTaskConfig;
+ reloadClientsFromNodeConfig(httpNodeConfig);
+ } else {
+ idConfigMap = fromSortTaskConfig;
+ reloadClientsFromSortTaskConfig(sortTaskConfig);
+ }
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
+ // log
+ LOG.info("end to get SortTaskConfig:taskName:{}:newIdConfigMap:{}", taskName,
+ objectMapper.writeValueAsString(idConfigMap));
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ private Map<String, HttpIdConfig> reloadIdParamsFromTaskConfig(TaskConfig taskConfig) {
+ if (taskConfig == null) {
+ return new HashMap<>();
+ }
+ return taskConfig.getClusterTagConfigs()
+ .stream()
+ .map(ClusterTagConfig::getDataFlowConfigs)
+ .flatMap(Collection::stream)
+ .map(HttpIdConfig::create)
+ .collect(Collectors.toMap(
+ config -> InlongId.generateUid(config.getInlongGroupId(), config.getInlongStreamId()),
+ v -> v,
+ (flow1, flow2) -> flow1));
+ }
+
+ private Map<String, HttpIdConfig> reloadIdParamsFromSortTaskConfig(SortTaskConfig sortTaskConfig)
+ throws JsonProcessingException {
+ if (sortTaskConfig == null) {
+ return new HashMap<>();
+ }
+ Map<String, HttpIdConfig> newIdConfigMap = new ConcurrentHashMap<>();
+ List<Map<String, String>> idList = this.sortTaskConfig.getIdParams();
+ ObjectMapper objectMapper = new ObjectMapper();
+ objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ for (Map<String, String> idParam : idList) {
+ String inlongGroupId = idParam.get(Constants.INLONG_GROUP_ID);
+ String inlongStreamId = idParam.get(Constants.INLONG_STREAM_ID);
+ String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
+ String jsonIdConfig = objectMapper.writeValueAsString(idParam);
+ HttpIdConfig idConfig = objectMapper.readValue(jsonIdConfig, HttpIdConfig.class);
+ newIdConfigMap.put(uid, idConfig);
+ }
+ return newIdConfigMap;
+ }
+
+ private void reloadClientsFromNodeConfig(HttpNodeConfig httpNodeConfig) {
+ Map<String, String> properties = httpNodeConfig.getProperties();
+ this.sinkContext = new Context(properties != null ? properties : new HashMap<>());
+ this.baseUrl = httpNodeConfig.getBaseUrl();
+ this.enableCredential = httpNodeConfig.getEnableCredential();;
+ this.username = httpNodeConfig.getUsername();
+ this.password = httpNodeConfig.getPassword();
+ this.maxConnect = httpNodeConfig.getMaxConnect();
+
+ this.maxConnectPerRoute = sinkContext.getInteger(KEY_MAX_CONNECT_PER_ROUTE, DEFAULT_MAX_CONNECT_PER_ROUTE);
+ this.connectionRequestTimeout =
+ sinkContext.getInteger(KEY_CONNECTION_REQUEST_TIMEOUT, DEFAULT_CONNECTION_REQUEST_TIMEOUT);
+ this.socketTimeout = sinkContext.getInteger(KEY_SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT);
+ this.maxRedirects = sinkContext.getInteger(KEY_MAX_REDIRECTS, DEFAULT_MAX_REDIRECTS);
+ this.logMaxLength = sinkContext.getInteger(KEY_LOG_MAX_LENGTH, DEFAULT_LOG_MAX_LENGTH);
+ }
+
+ private void reloadClientsFromSortTaskConfig(SortTaskConfig sortTaskConfig) {
+ this.sinkContext = new Context(sortTaskConfig.getSinkParams());
+ this.baseUrl = sinkContext.getString(KEY_BASE_URL);
+ this.enableCredential = sinkContext.getBoolean(KEY_ENABLE_CREDENTIAL, DEFAULT_ENABLE_CREDENTIAL);
+ this.username = sinkContext.getString(KEY_USERNAME);
+ this.password = sinkContext.getString(KEY_PASSWORD);
+ this.maxConnect = sinkContext.getInteger(KEY_MAX_CONNECT_TOTAL, DEFAULT_MAX_CONNECT_TOTAL);
+
+ this.maxConnectPerRoute = sinkContext.getInteger(KEY_MAX_CONNECT_PER_ROUTE, DEFAULT_MAX_CONNECT_PER_ROUTE);
+ this.connectionRequestTimeout =
+ sinkContext.getInteger(KEY_CONNECTION_REQUEST_TIMEOUT, DEFAULT_CONNECTION_REQUEST_TIMEOUT);
+ this.socketTimeout = sinkContext.getInteger(KEY_SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT);
+ this.maxRedirects = sinkContext.getInteger(KEY_MAX_REDIRECTS, DEFAULT_MAX_REDIRECTS);
+ this.logMaxLength = sinkContext.getInteger(KEY_LOG_MAX_LENGTH, DEFAULT_LOG_MAX_LENGTH);
+ }
+
+ public void addSendMetric(ProfileEvent currentRecord, String bid) {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(SortMetricItem.KEY_CLUSTER_ID, this.getClusterId());
+ dimensions.put(SortMetricItem.KEY_TASK_NAME, this.getTaskName());
+ // metric
+ fillInlongId(currentRecord, dimensions);
+ dimensions.put(SortMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(SortMetricItem.KEY_SINK_DATA_ID, bid);
+ long msgTime = currentRecord.getRawLogTime();
+ long auditFormatTime = msgTime - msgTime % CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(SortMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
+ SortMetricItem metricItem = this.getMetricItemSet().findMetricItem(dimensions);
+ long count = 1;
+ long size = currentRecord.getBody().length;
+ metricItem.sendCount.addAndGet(count);
+ metricItem.sendSize.addAndGet(size);
+ }
+
+ public void addSendFailMetric() {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(SortMetricItem.KEY_CLUSTER_ID, this.getClusterId());
+ dimensions.put(SortMetricItem.KEY_SINK_ID, this.getSinkName());
+ long msgTime = System.currentTimeMillis();
+ long auditFormatTime = msgTime - msgTime % CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(SortMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
+ SortMetricItem metricItem = this.getMetricItemSet().findMetricItem(dimensions);
+ metricItem.readFailCount.incrementAndGet();
+ }
+
+ public void addSendResultMetric(ProfileEvent currentRecord, String bid, boolean result, long sendTime) {
+ Map<String, String> dimensions = new HashMap<>();
+ dimensions.put(SortMetricItem.KEY_CLUSTER_ID, this.getClusterId());
+ dimensions.put(SortMetricItem.KEY_TASK_NAME, this.getTaskName());
+ // metric
+ fillInlongId(currentRecord, dimensions);
+ dimensions.put(SortMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(SortMetricItem.KEY_SINK_DATA_ID, bid);
+ final long currentTime = System.currentTimeMillis();
+ long msgTime = currentRecord.getRawLogTime();
+ long auditFormatTime = msgTime - msgTime % CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(SortMetricItem.KEY_MESSAGE_TIME, String.valueOf(auditFormatTime));
+ SortMetricItem metricItem = this.getMetricItemSet().findMetricItem(dimensions);
+ if (result) {
+ metricItem.sendSuccessCount.incrementAndGet();
+ metricItem.sendSuccessSize.addAndGet(currentRecord.getBody().length);
+ AuditUtils.add(AuditUtils.AUDIT_ID_SEND_SUCCESS, currentRecord);
+ if (sendTime > 0) {
+ long sinkDuration = currentTime - sendTime;
+ long nodeDuration = currentTime - currentRecord.getFetchTime();
+ long wholeDuration = currentTime - currentRecord.getRawLogTime();
+ metricItem.sinkDuration.addAndGet(sinkDuration);
+ metricItem.nodeDuration.addAndGet(nodeDuration);
+ metricItem.wholeDuration.addAndGet(wholeDuration);
+ }
+ } else {
+ metricItem.sendFailCount.incrementAndGet();
+ metricItem.sendFailSize.addAndGet(currentRecord.getBody().length);
+ }
+ }
+
+ public HttpIdConfig getIdConfig(String uid) {
+ return this.idConfigMap.get(uid);
+ }
+
+ public String getNodeId() {
+ return nodeId;
+ }
+
+ public Map<String, HttpIdConfig> getIdConfigMap() {
+ return idConfigMap;
+ }
+
+ public Context getSinkContext() {
+ return sinkContext;
+ }
+
+ public void setSinkContext(Context sinkContext) {
+ this.sinkContext = sinkContext;
+ }
+
+ public void offerDispatchQueue(HttpRequest httpRequest) {
+ this.offerCounter.incrementAndGet();
+ dispatchQueue.acquire(httpRequest.getEvent().getBody().length);
+ dispatchQueue.offer(httpRequest);
+ }
+
+ public HttpRequest takeDispatchQueue() {
+ HttpRequest httpRequest = this.dispatchQueue.pollRecord();
+ if (httpRequest != null) {
+ this.takeCounter.incrementAndGet();
+ }
+ return httpRequest;
+ }
+
+ public void backDispatchQueue(HttpRequest httpRequest) {
+ this.backCounter.incrementAndGet();
+ dispatchQueue.offer(httpRequest);
+ }
+
+ public void releaseDispatchQueue(HttpRequest httpRequest) {
+ dispatchQueue.release(httpRequest.getEvent().getBody().length);
+ }
+
+ public String getBaseUrl() {
+ return baseUrl;
+ }
+
+ public void setBaseUrl(String baseUrl) {
+ this.baseUrl = baseUrl;
+ }
+
+ public boolean getEnableCredential() {
+ return enableCredential;
+ }
+
+ public void setEnableCredential(boolean enableCredential) {
+ this.enableCredential = enableCredential;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public int getMaxConnect() {
+ return maxConnect;
+ }
+
+ public int getMaxConnectPerRoute() {
+ return maxConnectPerRoute;
+ }
+
+ public int getConnectionRequestTimeout() {
+ return connectionRequestTimeout;
+ }
+
+ public int getSocketTimeout() {
+ return socketTimeout;
+ }
+
+ public int getMaxRedirects() {
+ return maxRedirects;
+ }
+
+ public int getLogMaxLength() {
+ return logMaxLength;
+ }
+
+ public void setMaxConnect(int maxConnect) {
+ this.maxConnect = maxConnect;
+ }
+
+ public void setNodeId(String nodeId) {
+ this.nodeId = nodeId;
+ }
+
+ public void setIdConfigMap(Map<String, HttpIdConfig> idConfigMap) {
+ this.idConfigMap = idConfigMap;
+ }
+
+ public IEvent2HttpRequestHandler createHttpRequestHandler() {
+ // IEvent2HttpRequestHandler
+ String httpRequestHandlerClass = CommonPropertiesHolder.getString(KEY_EVENT_HTTPREQUEST_HANDLER,
+ DefaultEvent2HttpRequestHandler.class.getName());
+ try {
+ Class<?> handlerClass = ClassUtils.getClass(httpRequestHandlerClass);
+ Object handlerObject = handlerClass.getDeclaredConstructor().newInstance();
+ if (handlerObject instanceof IEvent2HttpRequestHandler) {
+ IEvent2HttpRequestHandler handler = (IEvent2HttpRequestHandler) handlerObject;
+ return handler;
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726175727
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ switch (idConfig.getMethod().toUpperCase()) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey(), StandardCharsets.UTF_8.name()) + "="
+ + URLEncoder.encode(entry.getValue(), StandardCharsets.UTF_8.name());
Review Comment:
```suggestion
return URLEncoder.encode(entry.getKey() + "=" + entry.getValue(), StandardCharsets.UTF_8.name());
```
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
Review Comment:
What does `cDelimiter` mean?
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726516317
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
Review Comment:
fixed.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ switch (idConfig.getMethod().toUpperCase()) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey(), StandardCharsets.UTF_8.name()) + "="
+ + URLEncoder.encode(entry.getValue(), StandardCharsets.UTF_8.name());
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(Collectors.joining("&"));
+ request = new HttpGet(uri + "?" + params);
+ for (Map.Entry<String, String> entry : idConfig.getHeaders().entrySet()) {
+ request.setHeader(entry.getKey(), entry.getValue());
+ }
+ request.setHeader("InlongGroupID", idConfig.getInlongGroupId());
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725051631
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpCallback.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.concurrent.FutureCallback;
+import org.slf4j.Logger;
+
+public class HttpCallback implements FutureCallback<HttpResponse> {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpCallback.class);
+
+ private HttpSinkContext context;
+ private HttpRequest requestItem;
+
+ public HttpCallback(HttpSinkContext context, HttpRequest requestItem) {
+ this.context = context;
+ this.requestItem = requestItem;
+ }
+
+ @Override
+ public void completed(HttpResponse httpResponse) {
+ int statusCode = httpResponse.getStatusLine().getStatusCode();
+ ProfileEvent event = requestItem.getEvent();
+ long sendTime = requestItem.getSendTime();
+
+ // is fail
+ if (statusCode != 200) {
+ context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
+ // if reach the max retry times, release the request
+ int remainRetryTimes = requestItem.getRemainRetryTimes();
+ if (remainRetryTimes == 1) {
+ context.releaseDispatchQueue(requestItem);
+ return;
+ }
+ if (remainRetryTimes > 1) {
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "vernedeng (via GitHub)" <gi...@apache.org>.
vernedeng commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724747758
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData = new ObjectMapper().writeValueAsString(fieldMap);
Review Comment:
No need to init object mapper for every single data
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724814355
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSinkContext.java:
##########
@@ -0,0 +1,405 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.ClusterTagConfig;
+import org.apache.inlong.common.pojo.sort.TaskConfig;
+import org.apache.inlong.common.pojo.sort.node.HttpNodeConfig;
+import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
+import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
+import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
+import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
+import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+import org.apache.inlong.sort.standalone.utils.Constants;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class HttpSinkContext extends SinkContext {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpSinkContext.class);
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String KEY_BASE_URL = "baseUrl";
+ public static final String KEY_ENABLE_CREDENTIAL = "enableCredential";
+ public static final String KEY_USERNAME = "username";
+ public static final String KEY_PASSWORD = "password";
+ public static final String KEY_MAX_CONNECT_TOTAL = "maxConnect";
+ public static final String KEY_MAX_CONNECT_PER_ROUTE = "maxConnectPerRoute";
+ public static final String KEY_CONNECTION_REQUEST_TIMEOUT = "connectionRequestTimeout";
+ public static final String KEY_SOCKET_TIMEOUT = "socketTimeout";
+ public static final String KEY_MAX_REDIRECTS = "maxRedirects";
+ public static final String KEY_LOG_MAX_LENGTH = "logMaxLength";
+ public static final String KEY_EVENT_HTTPREQUEST_HANDLER = "httpRequestHandler";
+
+ public static final boolean DEFAULT_ENABLE_CREDENTIAL = false;
+ public static final int DEFAULT_MAX_CONNECT_TOTAL = 1000;
+ public static final int DEFAULT_MAX_CONNECT_PER_ROUTE = 1000;
+ public static final int DEFAULT_CONNECTION_REQUEST_TIMEOUT = 0;
+ public static final int DEFAULT_SOCKET_TIMEOUT = 0;
+ public static final int DEFAULT_MAX_REDIRECTS = 0;
+ public static final int DEFAULT_LOG_MAX_LENGTH = 32 * 1024;
+
+ private Context sinkContext;
+ private HttpNodeConfig httpNodeConfig;
+ private String nodeId;
+ private Map<String, HttpIdConfig> idConfigMap = new ConcurrentHashMap<>();
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private final BufferQueue<HttpRequest> dispatchQueue;
+ private AtomicLong offerCounter = new AtomicLong(0);
+ private AtomicLong takeCounter = new AtomicLong(0);
+ private AtomicLong backCounter = new AtomicLong(0);
+ // rest client
+ private String baseUrl;
+ private boolean enableCredential;
+ private String username;
+ private String password;
+ private int maxConnect = DEFAULT_MAX_CONNECT_TOTAL;
+ private int maxConnectPerRoute = DEFAULT_MAX_CONNECT_PER_ROUTE;
+ private int connectionRequestTimeout = DEFAULT_CONNECTION_REQUEST_TIMEOUT;
+ private int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
+ private int maxRedirects = DEFAULT_MAX_REDIRECTS;
+ private int logMaxLength = DEFAULT_LOG_MAX_LENGTH;
+
+ public HttpSinkContext(String sinkName, Context context, Channel channel,
+ BufferQueue<HttpRequest> dispatchQueue) {
+ super(sinkName, context, channel);
+ this.sinkContext = context;
+ this.dispatchQueue = dispatchQueue;
+ this.nodeId = CommonPropertiesHolder.getString(KEY_NODE_ID);
+ }
+
+ public void reload() {
+ try {
+ LOG.info("SortTask:{},dispatchQueue:{},offer:{},take:{},back:{}",
+ taskName, dispatchQueue.size(), offerCounter.getAndSet(0),
+ takeCounter.getAndSet(0), backCounter.getAndSet(0));
+ TaskConfig newTaskConfig = SortConfigHolder.getTaskConfig(taskName);
+ SortTaskConfig newSortTaskConfig = SortClusterConfigHolder.getTaskConfig(taskName);
+ if ((newTaskConfig == null || newTaskConfig.equals(taskConfig))
+ && (newSortTaskConfig == null || newSortTaskConfig.equals(sortTaskConfig))) {
+ return;
+ }
+ LOG.info("get new SortTaskConfig:taskName:{}", taskName);
+
+ if (newTaskConfig != null) {
+ HttpNodeConfig requestNodeConfig = (HttpNodeConfig) newTaskConfig.getNodeConfig();
+ if (httpNodeConfig == null || requestNodeConfig.getVersion() > httpNodeConfig.getVersion()) {
+ this.httpNodeConfig = requestNodeConfig;
+ }
+ }
+
+ this.taskConfig = newTaskConfig;
+ this.sortTaskConfig = newSortTaskConfig;
+
+ // change current config
+ Map<String, HttpIdConfig> fromTaskConfig = reloadIdParamsFromTaskConfig(taskConfig);
+ Map<String, HttpIdConfig> fromSortTaskConfig = reloadIdParamsFromSortTaskConfig(sortTaskConfig);
+ if (unifiedConfiguration) {
+ idConfigMap = fromTaskConfig;
+ reloadClientsFromNodeConfig(httpNodeConfig);
+ } else {
+ idConfigMap = fromSortTaskConfig;
+ reloadClientsFromSortTaskConfig(sortTaskConfig);
+ }
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
+ // log
+ LOG.info("end to get SortTaskConfig:taskName:{}:newIdConfigMap:{}", taskName,
Review Comment:
```suggestion
LOG.info("End to get SortTaskConfig:taskName:{}:newIdConfigMap:{}", taskName,
```
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang merged PR #10837:
URL: https://github.com/apache/inlong/pull/10837
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725053854
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSinkContext.java:
##########
@@ -0,0 +1,405 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.ClusterTagConfig;
+import org.apache.inlong.common.pojo.sort.TaskConfig;
+import org.apache.inlong.common.pojo.sort.node.HttpNodeConfig;
+import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
+import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
+import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
+import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
+import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+import org.apache.inlong.sort.standalone.utils.Constants;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class HttpSinkContext extends SinkContext {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpSinkContext.class);
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String KEY_BASE_URL = "baseUrl";
+ public static final String KEY_ENABLE_CREDENTIAL = "enableCredential";
+ public static final String KEY_USERNAME = "username";
+ public static final String KEY_PASSWORD = "password";
+ public static final String KEY_MAX_CONNECT_TOTAL = "maxConnect";
+ public static final String KEY_MAX_CONNECT_PER_ROUTE = "maxConnectPerRoute";
+ public static final String KEY_CONNECTION_REQUEST_TIMEOUT = "connectionRequestTimeout";
+ public static final String KEY_SOCKET_TIMEOUT = "socketTimeout";
+ public static final String KEY_MAX_REDIRECTS = "maxRedirects";
+ public static final String KEY_LOG_MAX_LENGTH = "logMaxLength";
+ public static final String KEY_EVENT_HTTPREQUEST_HANDLER = "httpRequestHandler";
+
+ public static final boolean DEFAULT_ENABLE_CREDENTIAL = false;
+ public static final int DEFAULT_MAX_CONNECT_TOTAL = 1000;
+ public static final int DEFAULT_MAX_CONNECT_PER_ROUTE = 1000;
+ public static final int DEFAULT_CONNECTION_REQUEST_TIMEOUT = 0;
+ public static final int DEFAULT_SOCKET_TIMEOUT = 0;
+ public static final int DEFAULT_MAX_REDIRECTS = 0;
+ public static final int DEFAULT_LOG_MAX_LENGTH = 32 * 1024;
+
+ private Context sinkContext;
+ private HttpNodeConfig httpNodeConfig;
+ private String nodeId;
+ private Map<String, HttpIdConfig> idConfigMap = new ConcurrentHashMap<>();
+ private ObjectMapper objectMapper = new ObjectMapper();
+ private final BufferQueue<HttpRequest> dispatchQueue;
+ private AtomicLong offerCounter = new AtomicLong(0);
+ private AtomicLong takeCounter = new AtomicLong(0);
+ private AtomicLong backCounter = new AtomicLong(0);
+ // rest client
+ private String baseUrl;
+ private boolean enableCredential;
+ private String username;
+ private String password;
+ private int maxConnect = DEFAULT_MAX_CONNECT_TOTAL;
+ private int maxConnectPerRoute = DEFAULT_MAX_CONNECT_PER_ROUTE;
+ private int connectionRequestTimeout = DEFAULT_CONNECTION_REQUEST_TIMEOUT;
+ private int socketTimeout = DEFAULT_SOCKET_TIMEOUT;
+ private int maxRedirects = DEFAULT_MAX_REDIRECTS;
+ private int logMaxLength = DEFAULT_LOG_MAX_LENGTH;
+
+ public HttpSinkContext(String sinkName, Context context, Channel channel,
+ BufferQueue<HttpRequest> dispatchQueue) {
+ super(sinkName, context, channel);
+ this.sinkContext = context;
+ this.dispatchQueue = dispatchQueue;
+ this.nodeId = CommonPropertiesHolder.getString(KEY_NODE_ID);
+ }
+
+ public void reload() {
+ try {
+ LOG.info("SortTask:{},dispatchQueue:{},offer:{},take:{},back:{}",
+ taskName, dispatchQueue.size(), offerCounter.getAndSet(0),
+ takeCounter.getAndSet(0), backCounter.getAndSet(0));
+ TaskConfig newTaskConfig = SortConfigHolder.getTaskConfig(taskName);
+ SortTaskConfig newSortTaskConfig = SortClusterConfigHolder.getTaskConfig(taskName);
+ if ((newTaskConfig == null || newTaskConfig.equals(taskConfig))
+ && (newSortTaskConfig == null || newSortTaskConfig.equals(sortTaskConfig))) {
+ return;
+ }
+ LOG.info("get new SortTaskConfig:taskName:{}", taskName);
+
+ if (newTaskConfig != null) {
+ HttpNodeConfig requestNodeConfig = (HttpNodeConfig) newTaskConfig.getNodeConfig();
+ if (httpNodeConfig == null || requestNodeConfig.getVersion() > httpNodeConfig.getVersion()) {
+ this.httpNodeConfig = requestNodeConfig;
+ }
+ }
+
+ this.taskConfig = newTaskConfig;
+ this.sortTaskConfig = newSortTaskConfig;
+
+ // change current config
+ Map<String, HttpIdConfig> fromTaskConfig = reloadIdParamsFromTaskConfig(taskConfig);
+ Map<String, HttpIdConfig> fromSortTaskConfig = reloadIdParamsFromSortTaskConfig(sortTaskConfig);
+ if (unifiedConfiguration) {
+ idConfigMap = fromTaskConfig;
+ reloadClientsFromNodeConfig(httpNodeConfig);
+ } else {
+ idConfigMap = fromSortTaskConfig;
+ reloadClientsFromSortTaskConfig(sortTaskConfig);
+ }
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
+ // log
+ LOG.info("end to get SortTaskConfig:taskName:{}:newIdConfigMap:{}", taskName,
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "vernedeng (via GitHub)" <gi...@apache.org>.
vernedeng commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724750607
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpCallback.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.concurrent.FutureCallback;
+import org.slf4j.Logger;
+
+public class HttpCallback implements FutureCallback<HttpResponse> {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpCallback.class);
+
+ private HttpSinkContext context;
+ private HttpRequest requestItem;
+
+ public HttpCallback(HttpSinkContext context, HttpRequest requestItem) {
+ this.context = context;
+ this.requestItem = requestItem;
+ }
+
+ @Override
+ public void completed(HttpResponse httpResponse) {
+ int statusCode = httpResponse.getStatusLine().getStatusCode();
+ ProfileEvent event = requestItem.getEvent();
+ long sendTime = requestItem.getSendTime();
+
+ // is fail
+ if (statusCode != 200) {
+ context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
+ // if reach the max retry times, release the request
+ int remainRetryTimes = requestItem.getRemainRetryTimes();
+ if (remainRetryTimes == 1) {
+ context.releaseDispatchQueue(requestItem);
+ return;
+ }
+ if (remainRetryTimes > 1) {
Review Comment:
no need to check remainRetryTimes
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "aloyszhang (via GitHub)" <gi...@apache.org>.
aloyszhang commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724808773
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpChannelWorker.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Event;
+import org.apache.flume.Transaction;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HttpChannelWorker extends Thread {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpChannelWorker.class);
+
+ private final HttpSinkContext context;
+ private final int workerIndex;
+
+ private LifecycleState status;
+ private IEvent2HttpRequestHandler handler;
+
+ public HttpChannelWorker(HttpSinkContext context, int workerIndex) {
+ this.context = context;
+ this.workerIndex = workerIndex;
+ this.status = LifecycleState.IDLE;
+ this.handler = context.createHttpRequestHandler();
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpChannelWorker:{},status:{},index:{}", context.getTaskName(), status, workerIndex);
Review Comment:
```suggestion
LOG.info("Starting HttpChannelWorker:{},status:{},index:{}", context.getTaskName(), status, workerIndex);
```
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726516008
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSink.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpSink extends AbstractSink implements Configurable {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
+
+ private Context parentContext;
+ private BufferQueue<HttpRequest> dispatchQueue;
+ private HttpSinkContext context;
+ // workers
+ private List<HttpChannelWorker> workers = new ArrayList<>();
+ // output
+ private HttpOutputChannel outputChannel;
+
+ @Override
+ public void start() {
+ super.start();
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new HttpSinkContext(getName(), parentContext, getChannel(), dispatchQueue);
+ this.context.start();
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ HttpChannelWorker worker = new HttpChannelWorker(context, i);
+ this.workers.add(worker);
+ worker.start();
+ }
+ this.outputChannel = HttpSinkFactory.createHttpOutputChannel(context);
+ this.outputChannel.init();
+ this.outputChannel.start();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1726521076
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpOutputChannel.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+
+public class HttpOutputChannel extends Thread {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpOutputChannel.class);
+
+ private LifecycleState status;
+ private HttpSinkContext context;
+ private CloseableHttpAsyncClient httpClient;
+
+ public HttpOutputChannel(HttpSinkContext context) {
+ super(context.getTaskName());
+ this.context = context;
+ this.status = LifecycleState.IDLE;
+ }
+
+ public void init() {
+ initHttpClient();
+ }
+
+ private boolean initHttpClient() {
+ try {
+ if (httpClient == null) {
+ String userName = context.getUsername();
+ String password = context.getPassword();
+ LOG.info("initHttpAsyncClient:url:{}", context.getBaseUrl());
+
+ HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
+ final CredentialsProvider provider = new BasicCredentialsProvider();
+ if (context.getEnableCredential()) {
+ provider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(userName, password));
+ builder.setDefaultCredentialsProvider(provider);
+ }
+
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectionRequestTimeout(context.getConnectionRequestTimeout())
+ .setSocketTimeout(context.getSocketTimeout())
+ .setMaxRedirects(context.getMaxRedirects())
+ .setConnectTimeout(120 * 1000)
+ .build();
+
+ builder.setDefaultRequestConfig(requestConfig)
+ .setMaxConnTotal(context.getMaxConnect())
+ .setMaxConnPerRoute(context.getMaxConnectPerRoute());
+
+ httpClient = HttpSinkFactory.createHttpAsyncClient(builder);
+ httpClient.start();
+ }
+ } catch (Exception e) {
+ LOG.error("init httpclient failed.", e);
+ httpClient = null;
+ return false;
+ }
+ return true;
+ }
+
+ public void close() {
+ status = LifecycleState.STOP;
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ LOG.error(String.format("close HttpClient:%s", e.getMessage()), e);
+ }
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpOutputChannel:{},status:{}", context.getTaskName(), status);
+ while (status == LifecycleState.START) {
+ try {
+ send();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
+ }
+ }
+ }
+
+ public void send() throws InterruptedException {
+ HttpRequest httpRequest = null;
+ try {
+ // get httpRequest
+ httpRequest = context.takeDispatchQueue();
+ if (httpRequest == null) {
+ Thread.sleep(context.getProcessInterval());
+ return;
+ }
+ // get id config
+ String uid = httpRequest.getEvent().getUid();
+ if (context.getIdConfig(uid) == null) {
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ return;
+ }
+ // send
+ httpClient.execute(httpRequest.getRequest(), new HttpCallback(context, httpRequest));
+ context.addSendMetric(httpRequest.getEvent(), context.getTaskName());
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ if (httpRequest != null) {
+ context.backDispatchQueue(httpRequest);
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ }
+ try {
+ Thread.sleep(context.getProcessInterval());
+ } catch (InterruptedException e1) {
Review Comment:
The `e1` here is to distinguish it from the exception `e` thrown by the outer layer
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpOutputChannel.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+
+public class HttpOutputChannel extends Thread {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpOutputChannel.class);
+
+ private LifecycleState status;
+ private HttpSinkContext context;
+ private CloseableHttpAsyncClient httpClient;
+
+ public HttpOutputChannel(HttpSinkContext context) {
+ super(context.getTaskName());
+ this.context = context;
+ this.status = LifecycleState.IDLE;
+ }
+
+ public void init() {
+ initHttpClient();
+ }
+
+ private boolean initHttpClient() {
+ try {
+ if (httpClient == null) {
+ String userName = context.getUsername();
+ String password = context.getPassword();
+ LOG.info("initHttpAsyncClient:url:{}", context.getBaseUrl());
+
+ HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
+ final CredentialsProvider provider = new BasicCredentialsProvider();
+ if (context.getEnableCredential()) {
+ provider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(userName, password));
+ builder.setDefaultCredentialsProvider(provider);
+ }
+
+ RequestConfig requestConfig = RequestConfig.custom()
+ .setConnectionRequestTimeout(context.getConnectionRequestTimeout())
+ .setSocketTimeout(context.getSocketTimeout())
+ .setMaxRedirects(context.getMaxRedirects())
+ .setConnectTimeout(120 * 1000)
+ .build();
+
+ builder.setDefaultRequestConfig(requestConfig)
+ .setMaxConnTotal(context.getMaxConnect())
+ .setMaxConnPerRoute(context.getMaxConnectPerRoute());
+
+ httpClient = HttpSinkFactory.createHttpAsyncClient(builder);
+ httpClient.start();
+ }
+ } catch (Exception e) {
+ LOG.error("init httpclient failed.", e);
+ httpClient = null;
+ return false;
+ }
+ return true;
+ }
+
+ public void close() {
+ status = LifecycleState.STOP;
+ try {
+ httpClient.close();
+ } catch (IOException e) {
+ LOG.error(String.format("close HttpClient:%s", e.getMessage()), e);
+ }
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpOutputChannel:{},status:{}", context.getTaskName(), status);
+ while (status == LifecycleState.START) {
+ try {
+ send();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
+ }
+ }
+ }
+
+ public void send() throws InterruptedException {
+ HttpRequest httpRequest = null;
+ try {
+ // get httpRequest
+ httpRequest = context.takeDispatchQueue();
+ if (httpRequest == null) {
+ Thread.sleep(context.getProcessInterval());
+ return;
+ }
+ // get id config
+ String uid = httpRequest.getEvent().getUid();
+ if (context.getIdConfig(uid) == null) {
+ context.addSendResultMetric(httpRequest.getEvent(), context.getTaskName(), false,
+ httpRequest.getSendTime());
+ return;
+ }
+ // send
+ httpClient.execute(httpRequest.getRequest(), new HttpCallback(context, httpRequest));
+ context.addSendMetric(httpRequest.getEvent(), context.getTaskName());
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725053330
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSink.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+
+import org.apache.flume.Context;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpSink extends AbstractSink implements Configurable {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
+
+ private Context parentContext;
+ private BufferQueue<HttpRequest> dispatchQueue;
+ private HttpSinkContext context;
+ // workers
+ private List<HttpChannelWorker> workers = new ArrayList<>();
+ // output
+ private HttpOutputChannel outputChannel;
+
+ @Override
+ public void start() {
+ super.start();
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new HttpSinkContext(getName(), parentContext, getChannel(), dispatchQueue);
+ this.context.start();
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ HttpChannelWorker worker = new HttpChannelWorker(context, i);
+ this.workers.add(worker);
+ worker.start();
+ }
+ this.outputChannel = HttpSinkFactory.createHttpOutputChannel(context);
+ this.outputChannel.init();
+ this.outputChannel.start();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void stop() {
+ super.stop();
+ try {
+ this.context.close();
+ for (HttpChannelWorker worker : this.workers) {
+ worker.close();
+ }
+ this.workers.clear();
+ this.outputChannel.close();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void configure(Context context) {
+ LOG.info("start to configure:{}, context:{}.", this.getName(), context.toString());
Review Comment:
fixed.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpSinkContext.java:
##########
@@ -0,0 +1,405 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.common.pojo.sort.ClusterTagConfig;
+import org.apache.inlong.common.pojo.sort.TaskConfig;
+import org.apache.inlong.common.pojo.sort.node.HttpNodeConfig;
+import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig;
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
+import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
+import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
+import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
+import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
+import org.apache.inlong.sort.standalone.sink.SinkContext;
+import org.apache.inlong.sort.standalone.utils.BufferQueue;
+import org.apache.inlong.sort.standalone.utils.Constants;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class HttpSinkContext extends SinkContext {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpSinkContext.class);
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String KEY_BASE_URL = "baseUrl";
+ public static final String KEY_ENABLE_CREDENTIAL = "enableCredential";
+ public static final String KEY_USERNAME = "username";
+ public static final String KEY_PASSWORD = "password";
+ public static final String KEY_MAX_CONNECT_TOTAL = "maxConnect";
+ public static final String KEY_MAX_CONNECT_PER_ROUTE = "maxConnectPerRoute";
+ public static final String KEY_CONNECTION_REQUEST_TIMEOUT = "connectionRequestTimeout";
+ public static final String KEY_SOCKET_TIMEOUT = "socketTimeout";
+ public static final String KEY_MAX_REDIRECTS = "maxRedirects";
+ public static final String KEY_LOG_MAX_LENGTH = "logMaxLength";
+ public static final String KEY_EVENT_HTTPREQUEST_HANDLER = "httpRequestHandler";
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725052438
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpCallback.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.concurrent.FutureCallback;
+import org.slf4j.Logger;
+
+public class HttpCallback implements FutureCallback<HttpResponse> {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpCallback.class);
+
+ private HttpSinkContext context;
+ private HttpRequest requestItem;
+
+ public HttpCallback(HttpSinkContext context, HttpRequest requestItem) {
+ this.context = context;
+ this.requestItem = requestItem;
+ }
+
+ @Override
+ public void completed(HttpResponse httpResponse) {
+ int statusCode = httpResponse.getStatusLine().getStatusCode();
+ ProfileEvent event = requestItem.getEvent();
+ long sendTime = requestItem.getSendTime();
+
+ // is fail
+ if (statusCode != 200) {
+ context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
+ // if reach the max retry times, release the request
+ int remainRetryTimes = requestItem.getRemainRetryTimes();
+ if (remainRetryTimes == 1) {
+ context.releaseDispatchQueue(requestItem);
+ return;
+ }
+ if (remainRetryTimes > 1) {
+ requestItem.setRemainRetryTimes(remainRetryTimes - 1);
+ }
+ context.backDispatchQueue(requestItem);
+ } else {
+ context.addSendResultMetric(event, context.getTaskName(), true, sendTime);
+ context.releaseDispatchQueue(requestItem);
+ event.ack();
+ }
+ }
+
+ @Override
+ public void failed(Exception e) {
+ LOG.error("Http request failed,errorMsg:" + e.getMessage(), e);
Review Comment:
fixed.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpChannelWorker.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Event;
+import org.apache.flume.Transaction;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HttpChannelWorker extends Thread {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpChannelWorker.class);
+
+ private final HttpSinkContext context;
+ private final int workerIndex;
+
+ private LifecycleState status;
+ private IEvent2HttpRequestHandler handler;
+
+ public HttpChannelWorker(HttpSinkContext context, int workerIndex) {
+ this.context = context;
+ this.workerIndex = workerIndex;
+ this.status = LifecycleState.IDLE;
+ this.handler = context.createHttpRequestHandler();
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpChannelWorker:{},status:{},index:{}", context.getTaskName(), status, workerIndex);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725051150
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith("http://") && !uriString.startsWith("https://")) {
+ uriString = "http://" + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData = new ObjectMapper().writeValueAsString(fieldMap);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1725052926
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpChannelWorker.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Event;
+import org.apache.flume.Transaction;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HttpChannelWorker extends Thread {
+
+ public static final Logger LOG = LoggerFactory.getLogger(HttpChannelWorker.class);
+
+ private final HttpSinkContext context;
+ private final int workerIndex;
+
+ private LifecycleState status;
+ private IEvent2HttpRequestHandler handler;
+
+ public HttpChannelWorker(HttpSinkContext context, int workerIndex) {
+ this.context = context;
+ this.workerIndex = workerIndex;
+ this.status = LifecycleState.IDLE;
+ this.handler = context.createHttpRequestHandler();
+ }
+
+ @Override
+ public void run() {
+ status = LifecycleState.START;
+ LOG.info("start to HttpChannelWorker:{},status:{},index:{}", context.getTaskName(), status, workerIndex);
+ while (status == LifecycleState.START) {
+ try {
+ this.doRun();
+ } catch (Throwable t) {
+ LOG.error(t.getMessage(), t);
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "vernedeng (via GitHub)" <gi...@apache.org>.
vernedeng commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1724751006
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/HttpCallback.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.concurrent.FutureCallback;
+import org.slf4j.Logger;
+
+public class HttpCallback implements FutureCallback<HttpResponse> {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(HttpCallback.class);
+
+ private HttpSinkContext context;
+ private HttpRequest requestItem;
+
+ public HttpCallback(HttpSinkContext context, HttpRequest requestItem) {
+ this.context = context;
+ this.requestItem = requestItem;
+ }
+
+ @Override
+ public void completed(HttpResponse httpResponse) {
+ int statusCode = httpResponse.getStatusLine().getStatusCode();
+ ProfileEvent event = requestItem.getEvent();
+ long sendTime = requestItem.getSendTime();
+
+ // is fail
+ if (statusCode != 200) {
+ context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
+ // if reach the max retry times, release the request
+ int remainRetryTimes = requestItem.getRemainRetryTimes();
+ if (remainRetryTimes == 1) {
+ context.releaseDispatchQueue(requestItem);
+ return;
+ }
+ if (remainRetryTimes > 1) {
+ requestItem.setRemainRetryTimes(remainRetryTimes - 1);
+ }
+ context.backDispatchQueue(requestItem);
+ } else {
+ context.addSendResultMetric(event, context.getTaskName(), true, sendTime);
+ context.releaseDispatchQueue(requestItem);
+ event.ack();
+ }
+ }
+
+ @Override
+ public void failed(Exception e) {
+ LOG.error("Http request failed,errorMsg:" + e.getMessage(), e);
+ ProfileEvent event = requestItem.getEvent();
+ long sendTime = requestItem.getSendTime();
+ int remainRetryTimes = requestItem.getRemainRetryTimes();
+ context.addSendResultMetric(event, context.getTaskName(), false, sendTime);
+ // if reach the max retry times, release the request
+ if (remainRetryTimes == 1) {
+ context.releaseDispatchQueue(requestItem);
+ return;
+ }
+ if (remainRetryTimes > 1) {
Review Comment:
ditto
--
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
Re: [PR] [INLONG-10831][Sort] SortStandalone support HTTP sink [inlong]
Posted by "yfsn666 (via GitHub)" <gi...@apache.org>.
yfsn666 commented on code in PR #10837:
URL: https://github.com/apache/inlong/pull/10837#discussion_r1732071417
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ public static final String HTTP_PREFIX = "http://";
+ public static final String HTTPS_PREFIX = "https://";
+ public static final String INLONG_GROUP_ID_HEADER = "inlongGroupId";
+ public static final String INLONG_STREAM_ID_HEADER = "inlongStreamId";
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith(HTTP_PREFIX) && !uriString.startsWith(HTTPS_PREFIX)) {
+ uriString = HTTP_PREFIX + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ String requestMethod = idConfig.getMethod().toUpperCase();
+ switch (requestMethod) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey() + "=" + entry.getValue(),
+ StandardCharsets.UTF_8.name());
Review Comment:
fixed.
##########
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/http/DefaultEvent2HttpRequestHandler.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.sort.standalone.sink.http;
+
+import org.apache.inlong.sort.standalone.channel.ProfileEvent;
+import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory;
+import org.apache.inlong.sort.standalone.utils.UnescapeHelper;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.slf4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultEvent2HttpRequestHandler implements IEvent2HttpRequestHandler {
+
+ public static final Logger LOG = InlongLoggerFactory.getLogger(DefaultEvent2HttpRequestHandler.class);
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ public static final String HTTP_PREFIX = "http://";
+ public static final String HTTPS_PREFIX = "https://";
+ public static final String INLONG_GROUP_ID_HEADER = "inlongGroupId";
+ public static final String INLONG_STREAM_ID_HEADER = "inlongStreamId";
+
+ @Override
+ public HttpRequest parse(HttpSinkContext context, ProfileEvent event)
+ throws URISyntaxException, JsonProcessingException {
+ String uid = event.getUid();
+ HttpIdConfig idConfig = context.getIdConfig(uid);
+ if (idConfig == null) {
+ context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
+ return null;
+ }
+ // get the delimiter
+ String delimiter = idConfig.getSeparator();
+ char cDelimiter = delimiter.charAt(0);
+ // for tab separator
+ byte[] bodyBytes = event.getBody();
+ String strContext = new String(bodyBytes, Charset.defaultCharset());
+ // unescape
+ List<String> columnValues = UnescapeHelper.toFiledList(strContext, cDelimiter);
+ int valueLength = columnValues.size();
+ List<String> fieldList = idConfig.getFieldList();
+ int columnLength = fieldList.size();
+ // get field value
+ Map<String, String> fieldMap = new HashMap<>();
+ for (int i = 0; i < columnLength; ++i) {
+ String fieldName = fieldList.get(i);
+ String fieldValue = i < valueLength ? columnValues.get(i) : "";
+ fieldMap.put(fieldName, fieldValue);
+ }
+
+ // build
+ String uriString = context.getBaseUrl() + idConfig.getPath();
+ if (!uriString.startsWith(HTTP_PREFIX) && !uriString.startsWith(HTTPS_PREFIX)) {
+ uriString = HTTP_PREFIX + uriString;
+ }
+ URI uri = new URI(uriString);
+ String jsonData;
+ HttpUriRequest request;
+ String requestMethod = idConfig.getMethod().toUpperCase();
+ switch (requestMethod) {
+ case "GET":
+ String params = fieldMap.entrySet().stream()
+ .map(entry -> {
+ try {
+ return URLEncoder.encode(entry.getKey() + "=" + entry.getValue(),
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