Aiven-Open / Aiven-Open/http-connector-for-apache-kafka

Bug: Static cache pollution affects decimal.format configuration across concurrent connectors

Aberta
#308 0 comentários 0 reações 0 responsáveis Ver no GitHub
Linguagem predominante
Java
Estrelas
119
Forks
63
Métricas de merge de PRs
Nenhum PR com merge em 30d

Descrição

## Problem Description

`decimal.format` does not appear to be reliably isolated between multiple HTTP sink connector instances running in the same Kafka Connect worker.

When two connector instances use different `decimal.format` values, Avro decimal fields can be serialized using the wrong format. For example, a connector configured with `decimal.format=NUMERIC` may still serialize decimals as base64 when another connector instance in the same worker uses the default `BASE64` behavior.

## Expected Behavior

`decimal.format` should be applied per connector instance/task.

A connector configured with:

```properties
decimal.format=NUMERIC
```

should always serialize Avro decimal logical types as JSON numbers in the HTTP request body, regardless of other HTTP sink connectors running in the same Kafka Connect worker.

Example expected output:

```json
{
"amount": 25.5
}
```

## Actual Behavior

A connector configured with `decimal.format=NUMERIC` may serialize Avro decimal logical types as base64 strings.

Example actual output:

```json
{
"amount": "Y5w="
}
```

## Suspected Root Cause

In [`RecordValueConverter.java`](https://github.com/Aiven-Open/http-connector-for-apache-kafka/blob/a054599ac7ea7505c4db2acdd755e94fea9b264a/src/main/java/io/aiven/kafka/connect/http/converter/RecordValueConverter.java#L34), converter instances are cached in a static map:

```java
private static final ConcurrentHashMap, Converter> RUNTIME_CLASS_TO_CONVERTER_CACHE =
new ConcurrentHashMap<>();
```

`RecordValueConverter#getConverter()` stores the configured converter in this static cache based only on the runtime record value class:

```java
RUNTIME_CLASS_TO_CONVERTER_CACHE.computeIfAbsent(record.value().getClass(), ...)
```

For Avro records, the runtime value class is commonly a structured record type. If one connector initializes and caches a converter for that value class using one `decimal.format`, another connector processing the same value class can reuse that cached converter even if it was configured with a different `decimal.format`.

The cache key does not appear to include connector-specific configuration such as `decimal.format`, so configured converter behavior can leak across connector instances/tasks.

## Impact

- HTTP payloads can contain decimal values in a format that does not match the connector's `decimal.format` configuration.
- Avro decimal logical types may be emitted as base64 strings instead of JSON numbers in the HTTP request body.
- Downstream HTTP consumers that expect JSON numbers may reject the payload or process the decimal field incorrectly.
- This affects deployments with multiple HTTP sink connector instances in the same Kafka Connect worker using different `decimal.format` configurations.

## Example Scenario

1. Connector 1 runs with the default `decimal.format=BASE64`.
2. Connector 2 runs in the same Kafka Connect worker with `decimal.format=NUMERIC`.
3. Both connectors process Avro records containing decimal logical types.
4. Connector 2 should serialize decimals as JSON numbers.
5. Connector 2 may instead reuse converter behavior initialized by Connector 1 and serialize decimals as base64.

## Related Issues & References

- **PR #281**: [feat(avro-decimal-format): add configurable serialization for Avro decimals](https://github.com/Aiven-Open/http-connector-for-apache-kafka/pull/281)
- **Issue #280**: [Avro decimal format configuration](https://github.com/Aiven-Open/http-connector-for-apache-kafka/issues/280)

## Affected File

- [`src/main/java/io/aiven/kafka/connect/http/converter/RecordValueConverter.java`](https://github.com/Aiven-Open/http-connector-for-apache-kafka/blob/a054599ac7ea7505c4db2acdd755e94fea9b264a/src/main/java/io/aiven/kafka/connect/http/converter/RecordValueConverter.java#L34)

## Testing Recommendations

- Add a test with two HTTP sink connector instances running in the same Kafka Connect worker.
- Configure one connector with `decimal.format=BASE64`.
- Configure the other connector with `decimal.format=NUMERIC`.
- Send Avro records containing decimal logical types to both connectors.
- Verify each connector consistently uses its own configured decimal format.
- Verify behavior is stable regardless of connector startup order or message processing order.

Guia de contribuição

Nenhum guia de contribuição indexado para este repositório

Avaliação

Esta issue ainda não foi avaliada.

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.