You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/12/12 11:49:19 UTC
[GitHub] [inlong] woofyzhao opened a new pull request, #6849: [INLONG-6848][DataProxy] Adapt original InlongMsg protocol and headers
woofyzhao opened a new pull request, #6849:
URL: https://github.com/apache/inlong/pull/6849
- Fixes #6848
### Modifications
- support original default protocol and headers under the new sink layer
- fix script config file name error
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6849: [INLONG-6848][DataProxy] Adapt original InlongMsg protocol and headers
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #6849:
URL: https://github.com/apache/inlong/pull/6849#discussion_r1046590929
##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java:
##########
@@ -239,12 +240,16 @@ public void onException(Throwable ex) {
private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event, IdTopicConfig idConfig,
String topic) throws Exception {
// headers
- Map<String, String> headers = event.getSimpleProfile().getHeaders();
+ Map<String, String> headers = event.getProperties();
+ if (MapUtils.isEmpty(event.getProperties())) {
+ headers = event.getSimpleProfile().getHeaders();
+ }
// compress
byte[] bodyBytes = event.getSimpleProfile().getBody();
// sendAsync
Message message = new Message(topic, bodyBytes);
// add headers
+
Review Comment:
A strange blank line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6849: [INLONG-6848][DataProxy] Adapt original InlongMsg protocol and headers
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #6849:
URL: https://github.com/apache/inlong/pull/6849#discussion_r1046590628
##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java:
##########
@@ -239,12 +240,16 @@ public void onException(Throwable ex) {
private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event, IdTopicConfig idConfig,
String topic) throws Exception {
// headers
- Map<String, String> headers = event.getSimpleProfile().getHeaders();
+ Map<String, String> headers = event.getProperties();
+ if (MapUtils.isEmpty(event.getProperties())) {
Review Comment:
```suggestion
if (MapUtils.isEmpty(headers)) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #6849: [INLONG-6848][DataProxy] Adapt original InlongMsg protocol and headers
Posted by GitBox <gi...@apache.org>.
dockerzhang merged PR #6849:
URL: https://github.com/apache/inlong/pull/6849
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang commented on pull request #6849: [INLONG-6848][DataProxy] Adapt original InlongMsg protocol and headers
Posted by GitBox <gi...@apache.org>.
dockerzhang commented on PR #6849:
URL: https://github.com/apache/inlong/pull/6849#issuecomment-1347630231
@luchunliang PTAL, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] healchow commented on a diff in pull request #6849: [INLONG-6848][DataProxy] Adapt original InlongMsg protocol and headers
Posted by GitBox <gi...@apache.org>.
healchow commented on code in PR #6849:
URL: https://github.com/apache/inlong/pull/6849#discussion_r1046591063
##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java:
##########
@@ -204,14 +205,18 @@ public void onCompletion(RecordMetadata arg0, Exception ex) {
private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event, IdTopicConfig idConfig,
String topic) throws Exception {
// headers
- Map<String, String> headers = event.getSimpleProfile().getHeaders();
- // compress
+ Map<String, String> headers = event.getProperties();
+ if (MapUtils.isEmpty(headers)) {
+ headers = event.getSimpleProfile().getHeaders();
+ }
+ // body
byte[] bodyBytes = event.getSimpleProfile().getBody();
// sendAsync
long sendTime = System.currentTimeMillis();
// prepare ProducerRecord
ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>(topic, bodyBytes);
+
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