Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
GH-2852: Provide a way to customize a transactionIdSuffix
- Loading branch information
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| /* | ||
| * Copyright 2016-2023 the original author or authors. | ||
| * | ||
| * Licensed 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 | ||
| * | ||
| * https://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.springframework.kafka.core; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.concurrent.BlockingQueue; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.LinkedBlockingQueue; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| import org.springframework.lang.Nullable; | ||
| import org.springframework.util.Assert; | ||
|
|
||
| /** | ||
| * The {@link TransactionSuffixManager} implementation for managing transactional producer suffixes. | ||
| * If the {@link #maxCache} is greater than 0, the suffixes will be cached and reused. | ||
| * Otherwise, the suffixes will be generated on the fly. | ||
| * | ||
| * <p> | ||
| * Setting the {@link #setMaxCache(int)} enables caching to restrict the number of `transactional.id`. | ||
| * The cache is per `transactional.id` prefix. The cache size is limited by the {@link #maxCache}. | ||
| * Default is 0, which means no caching and restriction, so the `transactional.id` will be generated on the fly. | ||
| * | ||
| * @author Ilya Starchenko | ||
| * | ||
| * @since 3.1.1 | ||
| */ | ||
| public class DefaultTransactionSuffixManager implements TransactionSuffixManager { | ||
|
|
||
| private final AtomicInteger transactionIdSuffix = new AtomicInteger(); | ||
|
|
||
| private final Map<String, BlockingQueue<String>> suffixCache = new ConcurrentHashMap<>(); | ||
|
|
||
| private int maxCache; | ||
|
|
||
| /** | ||
| * Retrieve the suffix for the transactional producer from the cache or generate a new one | ||
|
Check failure on line 51 in spring-kafka/src/main/java/org/springframework/kafka/core/DefaultTransactionSuffixManager.java
|
||
| * if caching is disabled. | ||
| * @param txIdPrefix the transaction id prefix. | ||
| * @return the suffix. | ||
| * @throws NoProducerAvailableException if caching is enabled and no suffixes are available. | ||
| */ | ||
| @Override | ||
| public String retrieveSuffix(String txIdPrefix) { | ||
| BlockingQueue<String> cache = getSuffixCache(txIdPrefix); | ||
| if (cache == null) { | ||
| return String.valueOf(this.transactionIdSuffix.getAndIncrement()); | ||
| } | ||
|
|
||
| String suffix = cache.poll(); | ||
| if (suffix == null) { | ||
| throw new NoProducerAvailableException("No available transaction producer suffix for " + txIdPrefix); | ||
| } | ||
| return suffix; | ||
| } | ||
|
|
||
| @Override | ||
| public void returnSuffix(@Nullable String txIdPrefix, @Nullable String suffix) { | ||
| if (this.maxCache <= 0 || suffix == null) { | ||
| return; | ||
| } | ||
| BlockingQueue<String> queue = getSuffixCache(txIdPrefix); | ||
| if (queue != null && !queue.contains(suffix)) { | ||
| queue.add(suffix); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void reset() { | ||
| this.suffixCache.clear(); | ||
| } | ||
|
|
||
| /** | ||
| * Set the maximum size for transaction producer cache. | ||
| * @param maxCache the maxCache to set | ||
| */ | ||
| public void setMaxCache(int maxCache) { | ||
| Assert.isTrue(maxCache >= 0, "'maxCache' must greater than or equal 0"); | ||
| this.maxCache = maxCache; | ||
| } | ||
|
|
||
| @Nullable | ||
| private BlockingQueue<String> getSuffixCache(@Nullable String txIdPrefix) { | ||
| if (txIdPrefix == null || this.maxCache <= 0) { | ||
| return null; | ||
| } | ||
|
|
||
| return this.suffixCache.computeIfAbsent(txIdPrefix, txId -> { | ||
| BlockingQueue<String> queue = new LinkedBlockingQueue<>(); | ||
| for (int suffix = 0; suffix < this.maxCache; suffix++) { | ||
| queue.add(String.valueOf(this.transactionIdSuffix.getAndIncrement())); | ||
| } | ||
| return queue; | ||
| }); | ||
| } | ||
| } | ||
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,48 @@ | ||
| /* | ||
| * Copyright 2016-2023 the original author or authors. | ||
| * | ||
| * Licensed 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 | ||
| * | ||
| * https://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.springframework.kafka.core; | ||
|
|
||
| /** | ||
| * The strategy for managing transactional producer suffixes. | ||
| * | ||
| * @author Ilya Starchenko | ||
| * | ||
| * @since 3.1.1 | ||
| */ | ||
| public interface TransactionSuffixManager { | ||
|
|
||
| /** | ||
| * Retrieve the suffix for the transactional producer. | ||
| * | ||
| * @param txIdPrefix the transaction id prefix. | ||
| * @return the suffix. | ||
| */ | ||
| String retrieveSuffix(String txIdPrefix); | ||
|
|
||
| /** | ||
| * Return the suffix for the transactional producer back for reuse. | ||
| * | ||
| * @param txIdPrefix the transaction id prefix. | ||
| * @param suffix the suffix. | ||
| */ | ||
| void returnSuffix(String txIdPrefix, String suffix); | ||
|
|
||
| /** | ||
| * Clear all suffixes for the transactional producer. | ||
| */ | ||
| void reset(); | ||
| } |
Oops, something went wrong.