KafkaV2SinkConnector #38973
Merged
KafkaV2SinkConnector #38973
Conversation
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
xinlian12
requested review from
alzimmermsft,
samvaity,
g2vinay,
JonathanGiles,
chenrujun,
Netyyyy,
saragluna,
moarychan,
kushagraThapar,
FabianMeiswinkel,
kirankumarkolli,
milismsft,
aayush3011,
simorenoh,
jeet1995 and
Pilchie
as code owners
added 2 commits
February 27, 2024 12:52
xinlian12
force-pushed
the
kafkaV2SinkConnector-2
branch
from
20587ec to
ac1a96f
Compare
|
API change check APIView has identified API level changes in this PR and created following API reviews. |
xinlian12
force-pushed
the
kafkaV2SinkConnector-2
branch
from
83ff738 to
223d6d7
Compare
|
/azp run java - cosmos - tests |
|
Azure Pipelines successfully started running 1 pipeline(s). |
sdk/cosmos/azure-cosmos-kafka-connect/doc/configuration-reference.md
Outdated
Show resolved
Hide resolved
...src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosExceptionsHelper.java
Show resolved
Hide resolved
...nnect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkConfig.java
Outdated
Show resolved
Hide resolved
| .getCosmosAsyncContainerAccessor() | ||
| .getLinkWithoutTrailingSlash(container), | ||
| null, | ||
| new DocumentCollection()) |
There was a problem hiding this comment.
why new DocumentCollection and not null? I think his param is meant to be used when asynccache has an instance believed to be stale?
There was a problem hiding this comment.
both works. The value used here just mean if the cached value is not the same as the staled value, then use the cached value directly. So null or new DocumentCollection() would have the same effect
There was a problem hiding this comment.
BUt null is cheaper (no new garbage instantiation)
There was a problem hiding this comment.
will change in next PR
...va/com/azure/cosmos/kafka/connect/implementation/sink/idstrategy/TemplateStrategyConfig.java
Outdated
Show resolved
Hide resolved
...rc/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosChangeFeedModes.java
Show resolved
Hide resolved
|
/azp run java - cosmos - tests |
|
Azure Pipelines successfully started running 1 pipeline(s). |
|
/azp run java - cosmos - tests |
|
Azure Pipelines successfully started running 1 pipeline(s). |
|
/azp run java - cosmos - tests |
|
Azure Pipelines successfully started running 1 pipeline(s). |
xinlian12
pushed a commit
to xinlian12/azure-sdk-for-java
that referenced
this pull request
Mar 25, 2024
This reverts commit 6c983ab.
kushagraThapar
pushed a commit
that referenced
this pull request
Mar 26, 2024
* Revert "KafkaV2SinkConnector (#38973)" This reverts commit 6c983ab. * Revert "UsingTestContainerForKafkaIntegrationTests (#38884)" This reverts commit 12bec49. * Revert "KafkaV2SourceConnector (#38748)" This reverts commit 30835d9. * revert one more change * revert change --------- Co-authored-by: annie-mac <[email protected]>
Merged
drielenr
pushed a commit
that referenced
this pull request
Apr 2, 2024
* add sink connector v2 implementation --------- Co-authored-by: annie-mac <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Feature request #38769
In this PR, we added the kafka CosmosDB Sink Connector V2 version.
Config
kafka.connect.cosmos.accountEndpoint-> No default. Cosmos DB Account Endpoint Urikafka.connect.cosmos.accountKey-> No default. Cosmos DB Account Keykafka.connect.cosmos.useGatewayMode-> Default false. Flag to indicate whether to use gateway mode. By default it is false.kafka.connect.cosmos.preferredRegionsList-> Default empty list. Preferred regions list to be used for a multi region Cosmos DB account. This is a comma separated value (e.g.,[East US, West US]orEast US, West US) provided preferred regions will be used as hint. You should use a collocated kafka cluster with your Cosmos DB account and pass the kafka cluster region as preferred region. See list of azure regions herekafka.connect.cosmos.applicationName-> Default empty string. Will be added as the userAgent suffix.kafka.connect.cosmos.sink.database.name-> No Default. CosmosDb database name.kafka.connect.cosmos.sink.containers.topicMap-> No Default. A comma delimited list of Kafka topics mapped to Cosmos containers. For example: topic1#con1,topic2#con2.kafka.connect.cosmos.sink.errors.tolerance-> DefaultNone. Error tolerance level after exhausting all retries.Nonefor fail on error.Allfor log and continuekafka.connect.cosmos.sink.bulk.enabled-> Defaulttrue. Flag to indicate whether Cosmos DB bulk mode is enabled for Sink connector.kafka.connect.cosmos.sink.bulk.maxConcurrentCosmosPartitions-> Default -1. Usually this is only required to be tuned for large containers. Cosmos DB Item Write Max Concurrent Cosmos Partitions. If not specified it will be determined based on the number of the container's physical partitions which would indicate every batch is expected to have data from all Cosmos physical partitions. If specified it indicates from at most how many Cosmos Physical Partitions each batch contains data. So this config can be used to make bulk processing more efficient when input data in each batch has been repartitioned to balance to how many Cosmos partitions each batch needs to write. This is mainly useful for very large containers (with hundreds of physical partitions.kafka.connect.cosmos.sink.bulk.initialBatchSize-> Default 1. Cosmos DB initial bulk micro batch size - a micro batch will be flushed to the backend when the number of documents enqueued exceeds this size - or the target payload size is met. The micro batch size is getting automatically tuned based on the throttling rate. By default the initial micro batch size is 1. Reduce this when you want to avoid that the first few requests consume too many RUs.kafka.connect.cosmos.sink.write.strategy-> DefaultItemOverwrite. Cosmos DB Item write Strategy:ItemOverwrite(using upsert),ItemAppend(using create, ignore pre-existing items i.e., Conflicts),ItemDelete(deletes based on id/pk of data frame),ItemDeleteIfNotModified(deletes based on id/pk of data frame if etag hasn't changed since collecting id/pk),ItemOverwriteIfNotModified(using create if etag is empty, update/replace with etag pre-condition otherwise, if document was updated the pre-condition failure is ignored)kafka.connect.cosmos.sink.maxRetryCount-> Default 10. Cosmos DB max retry attempts on write failures for Sink connector. By default, the connector will retry on transient write errors for up to 10 times.kafka.connect.cosmos.sink.id.strategy-> DefaultProvidedInValueStrategy. A strategy used to populate the document with anid. Valid strategies are:TemplateStrategy,FullKeyStrategy,KafkaMetadataStrategy,ProvidedInKeyStrategy,ProvidedInValueStrategy. Configuration properties prefixed withid.strategyare passed through to the strategy. For example, when usingid.strategy=TemplateStrategy, the propertyid.strategy.templateis passed through to the template strategy and used to specify the template string to be used in constructing theid.