From b5aa59e244dbb482feff0645c39de610d16ea0a2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 20 Oct 2017 14:32:18 -0400 Subject: [PATCH] Prep work for Kafka Issue #236 See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/236 - add `none` as a synonym for `HeaderMode.raw` - add `headers` meaning "use native headers" - change the default header mode - `null` - remove `supportsHeadersNatively` - allow binders to chose header mode for each binding binders decide their own default when the property is null - Rabbit will choose `headers` as default (but can be overridden to `none` for a binding) `embeddedHeaders` will not be allowed - Kafka will choose `embeddedHeaders` as default for compatibility with 1.x peer apps - remove deprecations on header embedding code Polishing; support fallback - if the consumer side is configured to support embedded headers, binders should look for native headers first and signal their presence using the `BinderHeaders.NATIVE_HEADERS_PRESENT` header; fall back to embedded headers if it appears the payload might contain them - remove support for XD embedded headers (first byte < 0xff) - add a quick check for embedded headers by examining the first byte for 0xff - change the log from error to debug when decoding embedded headers fails - see the kafka binder PR for a test case with a consumer that can handle all message flavors (native, embedded, and no headers). --- .../spring-cloud-stream-overview.adoc | 20 +++++---- .../binder/AbstractMessageChannelBinder.java | 31 +++++++------ .../cloud/stream/binder/BinderHeaders.java | 7 +++ .../stream/binder/ConsumerProperties.java | 5 ++- .../stream/binder/EmbeddedHeaderUtils.java | 43 ++++++------------- .../cloud/stream/binder/HeaderMode.java | 22 +++++++++- .../stream/binder/ProducerProperties.java | 5 ++- ...SourceBindingWithGlobalPropertiesTest.java | 7 +-- 8 files changed, 79 insertions(+), 61 deletions(-) diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 4e52c2cb6..4e04b1e7d 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -1165,7 +1165,7 @@ Default: null (the default binder will be used, if one exists). The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings..consumer.`, e.g. `spring.cloud.stream.bindings.input.consumer.concurrency=3`. -Default values can be set by using the prefix `spring.cloud.stream.default.consumer`, e.g. `spring.cloud.stream.default.consumer.headerMode=raw`. +Default values can be set by using the prefix `spring.cloud.stream.default.consumer`, e.g. `spring.cloud.stream.default.consumer.headerMode=none`. concurrency:: The concurrency of the inbound consumer. @@ -1176,11 +1176,13 @@ partitioned:: + Default: `false`. headerMode:: - When set to `raw`, disables header parsing on input. + When set to `none`, disables header parsing on input. Effective only for messaging middleware that does not support message headers natively and requires header embedding. -Useful when inbound data is coming from outside Spring Cloud Stream applications. +This option is useful when consuming data from non-Spring Cloud Stream applications when native headers are not supported. +When set to `headers`, uses the middleware's native header mechanism. +When set to `embeddedHeaders`, embeds headers into the message payload. + -Default: `embeddedHeaders`. +Default: depends on binder implementation. maxAttempts:: If processing fails, the number of attempts to process the message (including the first). Set to 1 to disable retry. @@ -1252,16 +1254,18 @@ Default: `1`. requiredGroups:: A comma-separated list of groups to which the producer must ensure message delivery even if they start after it has been created (e.g., by pre-creating durable queues in RabbitMQ). headerMode:: - When set to `raw`, disables header embedding on output. + When set to `none`, disables header embedding on output. Effective only for messaging middleware that does not support message headers natively and requires header embedding. -Useful when producing data for non-Spring Cloud Stream applications. +This option is useful when producing data for non-Spring Cloud Stream applications when native headers are not supported. +When set to `headers`, uses the middleware's native header mechanism. +When set to `embeddedHeaders`, embeds headers into the message payload. + -Default: `embeddedHeaders`. +Default: Depends on binder implementation. useNativeEncoding:: When set to `true`, the outbound message is serialized directly by client library, which must be configured correspondingly (e.g. setting an appropriate Kafka producer value serializer). When this configuration is being used, the outbound message marshalling is not based on the `contentType` of the binding. When native encoding is used, it is the responsibility of the consumer to use appropriate decoder (ex: Kafka consumer value de-serializer) to deserialize the inbound message. -Also, when native encoding/decoding is used the `headerMode` property is ignored and headers will not be embedded into the message. +Also, when native encoding/decoding is used the `headerMode=embeddedHeaders` property is ignored and headers will not be embedded into the message. + Default: `false`. errorChannelEnabled:: diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 053c74d70..d40e18d60 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -74,24 +74,23 @@ public abstract class AbstractMessageChannelBinder) requestMessage, true); } catch (Exception e) { - AbstractMessageChannelBinder.this.logger.error( + AbstractMessageChannelBinder.this.logger.debug( EmbeddedHeaderUtils.decodeExceptionMessage( requestMessage), e); @@ -575,7 +575,6 @@ public abstract class AbstractMessageChannelBinder headers = new HashMap(); for (int i = 0; i < headerCount; i++) { @@ -131,6 +126,9 @@ public abstract class EmbeddedHeaderUtils { byteBuffer.get(newPayload); return buildMessageValues(newPayload, headers, copyRequestHeaders, requestHeaders); } + else { + return buildMessageValues(payload, new HashMap<>(), copyRequestHeaders, requestHeaders); + } } /** @@ -146,29 +144,6 @@ public abstract class EmbeddedHeaderUtils { return extractHeaders(payload, false, null); } - private static MessageValues oldExtractHeaders(ByteBuffer byteBuffer, byte[] bytes, int headerCount, - boolean copyRequestHeaders, MessageHeaders requestHeaders) throws UnsupportedEncodingException { - Map headers = new HashMap(); - for (int i = 0; i < headerCount; i++) { - int len = byteBuffer.get(); - String headerName = new String(bytes, byteBuffer.position(), len, "UTF-8"); - byteBuffer.position(byteBuffer.position() + len); - len = byteBuffer.get() & 0xff; - String headerValue = new String(bytes, byteBuffer.position(), len, "UTF-8"); - byteBuffer.position(byteBuffer.position() + len); - if (IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(headerName) - || IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName)) { - headers.put(headerName, Integer.parseInt(headerValue)); - } - else { - headers.put(headerName, headerValue); - } - } - byte[] newPayload = new byte[byteBuffer.remaining()]; - byteBuffer.get(newPayload); - return buildMessageValues(newPayload, headers, copyRequestHeaders, requestHeaders); - } - private static MessageValues buildMessageValues(byte[] payload, Map headers, boolean copyRequestHeaders, MessageHeaders requestHeaders) { MessageValues messageValues = new MessageValues(payload, headers); @@ -193,4 +168,14 @@ public abstract class EmbeddedHeaderUtils { return headersToMap; } + /** + * Return true if the bytes might have embedded headers. + * (First byte is 0xff and long enough for at least one header). + * @param bytes the array. + * @return true if it may have embedded headers. + */ + public static boolean mayHaveEmbeddedHeaders(byte[] bytes) { + return bytes.length > 8 && (bytes[0] & 0xff) == 0xff; + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java index 6356ef014..46757dd15 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -18,8 +18,28 @@ package org.springframework.cloud.stream.binder; /** * @author Marius Bogoevici + * @author Gary Russell */ public enum HeaderMode { + + /** + * @deprecated - use {@link #none}. + */ raw, + + /** + * No headers. + */ + none, + + /** + * Native headers. + */ + headers, + + /** + * Headers embedded in payload - e.g. kafka < 0.11 + */ embeddedHeaders + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java index 1ce7eb616..7c8ef7cb2 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -30,6 +30,7 @@ import org.springframework.expression.Expression; * * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Gary Russell */ @JsonInclude(Include.NON_DEFAULT) public class ProducerProperties { @@ -48,7 +49,7 @@ public class ProducerProperties { private String[] requiredGroups = new String[] {}; - private HeaderMode headerMode = HeaderMode.embeddedHeaders; + private HeaderMode headerMode; private boolean useNativeEncoding = false; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java index b87b41578..2015e447b 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 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. @@ -34,13 +34,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = SourceBindingWithGlobalPropertiesTest.TestSource.class, properties = { "spring.cloud.stream.default.contentType=application/json", "spring.cloud.stream.bindings.output.destination=ticktock", "spring.cloud.stream.default.producer.requiredGroups=someGroup", - "spring.cloud.stream.bindings.output.producer.headerMode=raw" }) + "spring.cloud.stream.bindings.output.producer.headerMode=none" }) public class SourceBindingWithGlobalPropertiesTest { @Autowired @@ -53,7 +54,7 @@ public class SourceBindingWithGlobalPropertiesTest { Assertions.assertThat(bindingProperties.getContentType()).isEqualTo("application/json"); Assertions.assertThat(bindingProperties.getDestination()).isEqualTo("ticktock"); Assertions.assertThat(bindingProperties.getProducer().getRequiredGroups()).containsExactly("someGroup"); - Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()).isEqualTo(HeaderMode.raw); + Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()).isEqualTo(HeaderMode.none); } @EnableBinding(Source.class)