Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ISSUE #4824] Add HTTP Sink Connector #4837

Merged
merged 27 commits into from
May 8, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
8275b59
feat: Add HTTP Sink Connector
cnzakii Apr 13, 2024
6776490
refactor: Replace okHttpClient with vertx.WebClient
cnzakii Apr 13, 2024
003f078
fix: Resolving dependency conflicts
cnzakii Apr 14, 2024
cd21ea3
test: Add HttpSinkConnectorTest
cnzakii Apr 14, 2024
dc08ce1
Merge branch 'apache:master' into feat_4824
cnzakii Apr 14, 2024
8486080
fix: Add License
cnzakii Apr 14, 2024
f2e3835
fix: Solving dependency issues
cnzakii Apr 14, 2024
eda1d14
fix: License Check
cnzakii Apr 14, 2024
611c8d6
feat: Add HTTPS/SSL support
cnzakii Apr 14, 2024
64f2822
fix: Optimize logging
cnzakii Apr 15, 2024
77ad32f
feat: Add webhook functionality
cnzakii Apr 16, 2024
3eb3af6
fix: Fix some bugs
cnzakii Apr 16, 2024
3c59c8e
test: add callback test
cnzakii Apr 16, 2024
408aaa6
refactor: Add webhook Support
cnzakii Apr 17, 2024
5cb55a8
fix: Optimization tests and configuration additions
cnzakii Apr 18, 2024
1050f8a
fix: code style
cnzakii Apr 18, 2024
fe2b732
feat: rebuild WebhookHttpSinkHandler and add RetryHttpSinkHandler
cnzakii Apr 27, 2024
7e2d978
fix: fix ci
cnzakii Apr 27, 2024
9c9e4a8
refactor: Use failsafe alternative resilience4j and optimize webhook …
cnzakii Apr 28, 2024
33b14e4
fix: fix License Check
cnzakii Apr 28, 2024
4af7455
fix: update something
cnzakii Apr 29, 2024
5e2bf52
fix: fix ci
cnzakii Apr 29, 2024
3dcc5a9
fix: update something
cnzakii Apr 29, 2024
6f3b361
fix: Optimized naming
cnzakii Apr 29, 2024
71c12e8
fix: fix ci
cnzakii Apr 29, 2024
6732e53
fix: fix style check error
cnzakii Apr 29, 2024
2f2e81e
test: update HttpSinkConnectorTest
cnzakii Apr 30, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions eventmesh-connectors/eventmesh-connector-http/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ dependencies {
implementation 'io.cloudevents:cloudevents-http-vertx:3.0.0'
implementation 'io.vertx:vertx-web:4.4.6'
implementation 'io.vertx:vertx-web-client:4.4.6'
implementation 'io.github.resilience4j:resilience4j-retry:1.7.1'

testImplementation "org.apache.httpcomponents:httpclient"
testImplementation 'org.mock-server:mockserver-netty:5.15.0'
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,11 @@
package org.apache.eventmesh.connector.http.sink;

import org.apache.eventmesh.connector.http.sink.config.HttpSinkConfig;
import org.apache.eventmesh.connector.http.sink.config.SinkConnectorConfig;
import org.apache.eventmesh.connector.http.sink.handle.CommonHttpSinkHandler;
import org.apache.eventmesh.connector.http.sink.handle.HttpSinkHandler;
import org.apache.eventmesh.connector.http.sink.handle.RetryHttpSinkHandler;
import org.apache.eventmesh.connector.http.sink.handle.WebhookHttpSinkHandler;
import org.apache.eventmesh.openconnect.api.config.Config;
import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
Expand Down Expand Up @@ -61,7 +64,23 @@ public void init(ConnectorContext connectorContext) throws Exception {

@SneakyThrows
private void doInit() {
this.sinkHandler = new CommonHttpSinkHandler(this.httpSinkConfig.connectorConfig);
// Fill default values if absent
SinkConnectorConfig.populateFieldsWithDefaults(this.httpSinkConfig.connectorConfig);
// Create different handlers for different configurations
HttpSinkHandler sinkHandler0;
if (this.httpSinkConfig.connectorConfig.getWebhookConfig().isActivate()) {
sinkHandler0 = new WebhookHttpSinkHandler(this.httpSinkConfig.connectorConfig);
} else {
sinkHandler0 = new CommonHttpSinkHandler(this.httpSinkConfig.connectorConfig);
cnzakii marked this conversation as resolved.
Show resolved Hide resolved
}

if (this.httpSinkConfig.connectorConfig.getRetryConfig().getMaxAttempts() > 1) {
// Wrap the sink handler with a retry handler
this.sinkHandler = new RetryHttpSinkHandler(this.httpSinkConfig.connectorConfig, sinkHandler0);
} else {
// Use the original sink handler
this.sinkHandler = sinkHandler0;
}
}

@Override
Expand Down Expand Up @@ -93,7 +112,7 @@ public void put(List<ConnectRecord> sinkRecords) {
continue;
}
// Handle the ConnectRecord
this.sinkHandler.handle(sinkRecord);
this.sinkHandler.multiHandle(sinkRecord);
} catch (Exception e) {
log.error("Failed to sink message via HTTP. ", e);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
/*
* 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.eventmesh.connector.http.sink.config;

import lombok.Data;

@Data
public class HttpRetryConfig {
// maximum number of attempts to retry, default 3, if set to 0 or 1, no retry
private int maxAttempts = 3;
cnzakii marked this conversation as resolved.
Show resolved Hide resolved

// retry interval, default 2000ms
private int interval = 2000;

// Default value is false, indicating that only requests with network-level errors will be retried.
// If set to true, all failed requests will be retried, including network-level errors and non-2xx responses.
private boolean retryAll = false;
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
/*
* 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.eventmesh.connector.http.sink.config;

import lombok.Data;

@Data
public class HttpWebhookConfig {

private boolean activate = false;

// Path to display/export callback data
private String exportPath = "/export";

private int port;

// timeunit: ms
private int idleTimeout = 5000;
}
Original file line number Diff line number Diff line change
Expand Up @@ -26,17 +26,7 @@ public class SinkConnectorConfig {

private String connectorName;

private String host;

private int port;

private String path;

// whether the connector is HTTPS connector
private boolean ssl;

// whether the connector is a webhook connector
private boolean webhook;
private String[] urls;

// keepAlive, default true
private boolean keepAlive = HttpClientOptions.DEFAULT_KEEP_ALIVE;
Expand All @@ -53,6 +43,12 @@ public class SinkConnectorConfig {
// maximum number of HTTP/1 connections a client will pool, default 5
private int maxConnectionPoolSize = HttpClientOptions.DEFAULT_MAX_POOL_SIZE;

// retry config
private HttpRetryConfig retryConfig = new HttpRetryConfig();

// webhook config
private HttpWebhookConfig webhookConfig = new HttpWebhookConfig();


/**
* Fill default values if absent (When there are multiple default values for a field)
Expand All @@ -69,7 +65,7 @@ public static void populateFieldsWithDefaults(SinkConnectorConfig config) {

// Set default values for idleTimeout
if (config.getIdleTimeout() == 0) {
int idleTimeout = config.isWebhook() ? webhookHttpIdleTimeout : commonHttpIdleTimeout;
int idleTimeout = config.webhookConfig.isActivate() ? webhookHttpIdleTimeout : commonHttpIdleTimeout;
config.setIdleTimeout(idleTimeout);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,16 @@

import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;

import java.time.LocalDateTime;

import lombok.Builder;
import lombok.Data;

/**
* a special ConnectRecord for HttpSinkConnector
*/
@Data
@Builder
public class HttpConnectRecord {

private String type;
Expand All @@ -33,4 +37,18 @@ public class HttpConnectRecord {

private ConnectRecord data;

/**
* Convert ConnectRecord to HttpConnectRecord
*
* @param record the ConnectRecord to convert
* @return the converted HttpConnectRecord
*/
public static HttpConnectRecord convertConnectRecord(ConnectRecord record, String type) {
return HttpConnectRecord.builder()
.type(type)
.timestamp(LocalDateTime.now().toString())
.data(record)
.build();
}

}
Loading