forked from raystack/firehose
-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* feat: httpv2 sink using depot * feat: httpv2 sink using depot * feat: httpv2 sink using depot * feat: httpv2 sink using depot * docs: add http sink docs * docs: add httpv2 sink docs * docs: add httpv2 sink docs * docs: add httpv2 sink docs * docs: add httpv2 sink docs * docs: add httpv2 sink docs * fix: deprecate HTTP SinkType * docs: deprecate HTTP sink type in docs * docs: deprecate HTTP sink type in docs * docs: deprecate HTTP sink type in docs * docs: deprecate HTTP sink type in docs * chore: depot version bump * chore: depot version bump * fix: add SINK_RETRYABLE_ERROR in ERROR_TYPES_FOR_RETRY * docs: add docs for httpv2 sink (#33) * docs: add docs for httpv2 sink * docs: add docs for httpv2 sink * docs: add docs for httpv2 sink * chore: version bump * refactor: import httpv2 classes * fix: resolve conflicts in import * Update build.gradle * feat: make httpv2 connection equal to num threads * feat: refactor to new class * feat: fix build * chore: version bump of depot to 0.9.0 * Update build.gradle * chore: version bump depot for httpv2 * fix: handle null values for kafka consumer mode and sink pool num threads --------- Co-authored-by: gagan.dhand <[email protected]>
- Loading branch information
1 parent
91794d2
commit e021eeb
Showing
10 changed files
with
103 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| # HttpV2 Sink | ||
|
|
||
| HttpV2 Sink is implemented in Firehose using the Http sink connector implementation in Depot library. For details on all the features supported by HttpV2 Sink, please refer the Depot documentation [here](https://github.com/goto/depot/blob/main/docs/sinks/http-sink.md). | ||
|
|
||
| ### Configuration | ||
|
|
||
| For HttpV2 sink in Firehose we need to set first (`SINK_TYPE`=`httpv2`). There are some generic configs which are common across different sink types which need to be set which are mentioned in [generic.md](../advance/generic.md). Http sink specific configs are mentioned in Depot repository. You can check out the Http Sink configs [here](https://github.com/goto/depot/blob/main/docs/reference/configuration/http-sink.md) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
src/main/java/com/gotocompany/firehose/sink/httpv2/HttpV2SinkUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package com.gotocompany.firehose.sink.httpv2; | ||
|
|
||
| import com.gotocompany.firehose.config.enums.KafkaConsumerMode; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public class HttpV2SinkUtils { | ||
|
|
||
| public static void addAdditionalConfigsForHttpV2Sink(Map<String, String> env) { | ||
|
|
||
| switch (KafkaConsumerMode.valueOf(env.getOrDefault("SOURCE_KAFKA_CONSUMER_MODE", "SYNC"))) { | ||
| case SYNC: | ||
| env.put("SINK_HTTPV2_MAX_CONNECTIONS", "1"); | ||
| break; | ||
| case ASYNC: | ||
| env.put("SINK_HTTPV2_MAX_CONNECTIONS", env.getOrDefault("SINK_POOL_NUM_THREADS", "1")); | ||
| } | ||
| } | ||
| } |
30 changes: 30 additions & 0 deletions
30
src/test/java/com/gotocompany/firehose/sink/httpv2/HttpV2SinkUtilsTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package com.gotocompany.firehose.sink.httpv2; | ||
|
|
||
| import org.junit.Assert; | ||
| import org.junit.Test; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| public class HttpV2SinkUtilsTest { | ||
| @Test | ||
| public void shouldAddAdditionalConfigsForSyncConsumer() { | ||
| Map<String, String> config = new HashMap<String, String>() {{ | ||
| put("SOURCE_KAFKA_CONSUMER_MODE", "sync"); | ||
| put("SINK_HTTPV2_MAX_CONNECTIONS", "5"); | ||
| }}; | ||
| HttpV2SinkUtils.addAdditionalConfigsForHttpV2Sink(config); | ||
| Assert.assertEquals(config.get("SINK_HTTPV2_MAX_CONNECTIONS"), "1"); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldAddAdditionalConfigsForASyncConsumer() { | ||
| Map<String, String> config = new HashMap<String, String>() {{ | ||
| put("SOURCE_KAFKA_CONSUMER_MODE", "async"); | ||
| put("SINK_POOL_NUM_THREADS", "10"); | ||
| put("SINK_HTTPV2_MAX_CONNECTIONS", "5"); | ||
| }}; | ||
| HttpV2SinkUtils.addAdditionalConfigsForHttpV2Sink(config); | ||
| Assert.assertEquals(config.get("SINK_HTTPV2_MAX_CONNECTIONS"), "10"); | ||
| } | ||
| } |