Skip to content

Commit

Permalink
Don't require uuid header presence
Browse files Browse the repository at this point in the history
  • Loading branch information
yevhenii0 committed Oct 8, 2024
1 parent aedef3e commit 7079624
Show file tree
Hide file tree
Showing 14 changed files with 101 additions and 350 deletions.
17 changes: 3 additions & 14 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,14 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [1.0.0] - 2024-10-07
## [0.31.0] - 2024-10-07

### Changed

Added two new properties to the `TkmsMessage`:

- `uuid` [required] - specifies uniqueness of the message. Will be delivered to consumers in the `x-wise-uuid` header.
- `priority` [optional] - specifies priority of the message. Lower number - higher priority. Will be delivered to consumers in the `x-wise-uuid` header.
- `uuid` - specifies uniqueness of the message. Will be delivered to consumers in the `x-wise-uuid` header.
- `priority` - specifies priority of the message. Lower number - higher priority. Will be delivered to consumers in the `x-wise-uuid` header.

Consumers of messages that have UUID and priority headers can efficiently use provided values for deduplication and other processing purposes with no need to deserialize payloads.

Expand All @@ -21,17 +21,6 @@ Best practices for setting UUID value:
- If payload already has UUID value then assign the same value to the corresponding `TkmsMessage`. It ensures that consumers of such messages can consistently deduplicate them by depending on one of those UUIDs. It simplifies consumers migration to standard header based UUID deduplication.
- If custom message identification mechanism is used (not based on UUID), still generate and assign UUID to the messages. However, be mindful of cases when messages are sent in non-transactional environments. For example, the same message might be sent twice with different UUIDs but the same identity (according to the custom identification mechanism).

<b>UUID presence is required by default.</b> This is the only breaking change in this version. The intention of such behaviour is to force producers to supply UUID value in messages sent, which will greatly benefit consumers of those messages. This requirement can be relaxed by setting the following configuration property.
```yaml
uuid-header-required: false
```
Added a new gauge metric that exposes value of `uuid-header-required` configuration property.
```
# 0 - uuid header isn't required, 1 - uuid header is required
tw_tkms_configuration_uuid_header_required
```
## [0.30.1] - 2024-08-08
### Changed
- MeterFilter's applied by the library are no longer explicitly applied and are instead
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package com.transferwise.kafka.tkms.demoapp;

import com.transferwise.common.baseutils.ExceptionUtils;
import com.transferwise.common.baseutils.UuidUtils;
import com.transferwise.kafka.tkms.api.ITransactionalKafkaMessageSender;
import com.transferwise.kafka.tkms.api.TkmsMessage;
import java.nio.charset.StandardCharsets;
Expand Down Expand Up @@ -51,8 +50,6 @@ public void produce(long threadCount, long batchCount, long batchSize) {
String key = String.valueOf(finalT * batchCount * batchSize + finalI * batchSize + j);

TkmsMessage message = new TkmsMessage()
.setUuid(UuidUtils.generatePrefixCombUuid())
.setPriority(17L)
.setTopic("MyTopic")
.setTimestamp(Instant.now())
.setKey(key).setValue(textMessage.getBytes(StandardCharsets.UTF_8));
Expand Down
2 changes: 1 addition & 1 deletion gradle.properties
Original file line number Diff line number Diff line change
@@ -1 +1 @@
version=1.0.0
version=0.31.0
Original file line number Diff line number Diff line change
Expand Up @@ -395,23 +395,13 @@ protected void validateMessage(TkmsMessage message, int messageIdx) {
message.getShard(), properties.getShardsCount());
}
Preconditions.checkNotNull(message.getValue(), "%s: Value can not be null.", messageIdx);
boolean uuidHeaderPresent = false;
if (message.getHeaders() != null) {
for (int headerIdx = 0; headerIdx < message.getHeaders().size(); headerIdx++) {
Header header = message.getHeaders().get(headerIdx);
Preconditions.checkNotNull(header.getValue(), "%s: Header value @{%s} can not be null.", messageIdx, headerIdx);
Preconditions.checkArgument(!Strings.isNullOrEmpty(header.getKey()), "%s: Header key @{%s} can not be null.", messageIdx, headerIdx);
uuidHeaderPresent |= StandardHeaders.X_WISE_UUID.equals(header.getKey());
}
}
if (properties.isUuidHeaderRequired() && !uuidHeaderPresent) {
throw new IllegalArgumentException(
"%d: Message is required to have @{%s} header.".formatted(
messageIdx,
StandardHeaders.X_WISE_UUID
)
);
}
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import com.transferwise.common.baseutils.validation.LegacyResolvedValue;
import com.transferwise.common.baseutils.validation.ResolvedValue;
import com.transferwise.kafka.tkms.CompressionAlgorithm;
import com.transferwise.kafka.tkms.api.TkmsMessage;
import com.transferwise.kafka.tkms.api.TkmsShardPartition;
import java.time.Duration;
import java.util.ArrayList;
Expand Down Expand Up @@ -239,11 +238,6 @@ public void afterPropertiesSet() {
*/
private boolean deferMessageRegistrationUntilCommit = false;

/**
* Whether every message sent is required to have {@code x-wise-uuid} header. See {@link TkmsMessage#getUuid()} for more details.
*/
private boolean uuidHeaderRequired = true;

@Valid
@jakarta.validation.Valid
private Compression compression = new Compression();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,15 @@
import com.transferwise.kafka.tkms.config.TkmsProperties;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.Meter;
import io.micrometer.core.instrument.Meter.Type;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.config.MeterFilter;
import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
import java.time.Instant;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import lombok.Data;
Expand All @@ -31,7 +37,6 @@ public class TkmsMetricsTemplate implements ITkmsMetricsTemplate, InitializingBe
// Miccrometer 1.13 (which comes with Spring boot 3.3) doesn't properly convert gauge metrics with info suffix when using underscore,
// using dot here as a workaround
public static final String GAUGE_LIBRARY_INFO = "tw.library.info";
public static final String GAUGE_CONFIGURATION_UUID_HEADER_REQUIRED = "tw_tkms_configuration_uuid_header_required";
public static final String TIMER_PROXY_POLL = "tw_tkms_proxy_poll";
public static final String GAUGE_PROXY_POLL_IN_PROGRESS = "tw_tkms_proxy_poll_in_progress";
public static final String TIMER_PROXY_CYCLE = "tw_tkms_proxy_cycle";
Expand Down Expand Up @@ -283,9 +288,6 @@ public void registerLibrary() {
Gauge.builder(GAUGE_LIBRARY_INFO, () -> 1d).tags("version", version, "library", "tw-tkms")
.description("Provides metadata about the library, for example the version.")
.register(meterCache.getMeterRegistry());
Gauge.builder(GAUGE_CONFIGURATION_UUID_HEADER_REQUIRED, tkmsProperties, props -> props.isUuidHeaderRequired() ? 1d : 0d)
.description("0 - uuid header isn't required, 1 - uuid header is required")
.register(meterCache.getMeterRegistry());
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import com.transferwise.common.baseutils.UuidUtils;
import com.transferwise.common.baseutils.clock.TestClock;
import com.transferwise.common.baseutils.transactionsmanagement.ITransactionsHelper;
import com.transferwise.kafka.tkms.api.TkmsMessage;
Expand Down Expand Up @@ -107,12 +106,7 @@ void testIfEarliestMessageTrackerBehavesAsExpected() {

protected void sendMessageAndWaitForArrival() {
transactionsHelper.withTransaction().run(() -> {
var result = tkms.sendMessage(
new TkmsMessage()
.setUuid(UuidUtils.generatePrefixCombUuid())
.setTopic(testTopic)
.setValue("Hello Kristo!".getBytes(StandardCharsets.UTF_8))
);
var result = tkms.sendMessage(new TkmsMessage().setTopic(testTopic).setValue("Hello Kristo!".getBytes(StandardCharsets.UTF_8)));
log.info("Registered a message with storage id " + result.getStorageId());
}
);
Expand Down
Loading

0 comments on commit 7079624

Please sign in to comment.