SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

polishing

Browse Source 
Resolves #472

polishing
This commit is contained in:
Oleg Zhurakousky
2018-10-22 11:35:29 +02:00
parent 4c15cbc1d6
commit 8ab289a715
3 changed files with 32 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java
View File

@@ -83,7 +83,6 @@ class KStreamBinder extends
}
@Override
@SuppressWarnings("unchecked")
protected Binding<KStream<Object, Object>> doBindConsumer(String name, String group,
KStream<Object, Object> inputTarget,
ExtendedConsumerProperties<KafkaStreamsConsumerProperties> properties) {

57
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
View File

@@ -34,6 +34,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
@@ -75,6 +76,7 @@ public class KafkaStreamsMessageConversionDelegate {
* @param outboundBindTarget outbound KStream target
* @return serialized KStream
*/
@SuppressWarnings("rawtypes")
public KStream serializeOnOutbound(KStream<?,?> outboundBindTarget) {
String contentType = this.kstreamBindingInformationCatalogue.getContentType(outboundBindTarget);
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
@@ -87,7 +89,7 @@ public class KafkaStreamsMessageConversionDelegate {
headers.put(MessageHeaders.CONTENT_TYPE, contentType);
}
MessageHeaders messageHeaders = new MessageHeaders(headers);
return
return
messageConverter.toMessage(message.getPayload(),
messageHeaders).getPayload();
});
@@ -100,7 +102,7 @@ public class KafkaStreamsMessageConversionDelegate {
* @param bindingTarget inbound KStream target
* @return deserialized KStream
*/
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
public KStream deserializeOnInbound(Class<?> valueClass, KStream<?, ?> bindingTarget) {
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
final PerRecordContentTypeHolder perRecordContentTypeHolder = new PerRecordContentTypeHolder();
@@ -116,26 +118,17 @@ public class KafkaStreamsMessageConversionDelegate {
try {
//if the record is a tombstone, ignore and exit from processing further.
if (o2 != null) {
if (valueClass.isAssignableFrom(o2.getClass())) {
keyValueThreadLocal.set(new KeyValue<>(o, o2));
}
else if (o2 instanceof Message) {
Message<?> m1 = (Message) o2;
if (perRecordContentTypeHolder.contentType != null) {
m1 = MessageBuilder.fromMessage(m1).setHeader("contentType", perRecordContentTypeHolder.contentType).build();
}
if (valueClass.isAssignableFrom(m1.getPayload().getClass())) {
keyValueThreadLocal.set(new KeyValue<>(o, m1.getPayload()));
if (o2 instanceof Message || o2 instanceof String || o2 instanceof byte[]) {
Message<?> m1 = null;
if (o2 instanceof Message) {
m1 = perRecordContentTypeHolder.contentType != null
? MessageBuilder.fromMessage((Message<?>) o2).setHeader(MessageHeaders.CONTENT_TYPE, perRecordContentTypeHolder.contentType).build() : (Message<?>)o2;
}
else {
convertAndSetMessage(o, valueClass, messageConverter, m1);
m1 = perRecordContentTypeHolder.contentType != null ? MessageBuilder.withPayload(o2)
.setHeader(MessageHeaders.CONTENT_TYPE, perRecordContentTypeHolder.contentType).build() : MessageBuilder.withPayload(o2).build();
}
}
else if (o2 instanceof String || o2 instanceof byte[]) {
Message<?> message = perRecordContentTypeHolder.contentType != null ? MessageBuilder.withPayload(o2)
.setHeader("contentType", perRecordContentTypeHolder.contentType).build() : MessageBuilder.withPayload(o2).build();
convertAndSetMessage(o, valueClass, messageConverter, message);
convertAndSetMessage(o, valueClass, messageConverter, m1);
}
else {
keyValueThreadLocal.set(new KeyValue<>(o, o2));
@@ -174,7 +167,7 @@ public class KafkaStreamsMessageConversionDelegate {
}
}
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
private void resolvePerRecordContentType(KStream<?, ?> outboundBindTarget, PerRecordContentTypeHolder perRecordContentTypeHolder) {
outboundBindTarget.process(() -> new Processor() {
@@ -188,7 +181,7 @@ public class KafkaStreamsMessageConversionDelegate {
@Override
public void process(Object key, Object value) {
final Headers headers = context.headers();
final Iterable<Header> contentTypes = headers.headers("contentType");
final Iterable<Header> contentTypes = headers.headers(MessageHeaders.CONTENT_TYPE);
if (contentTypes != null && contentTypes.iterator().hasNext()) {
final String contentType = new String(contentTypes.iterator().next().value());
//remove leading and trailing quotes
@@ -205,14 +198,15 @@ public class KafkaStreamsMessageConversionDelegate {
}
private void convertAndSetMessage(Object o, Class<?> valueClass, MessageConverter messageConverter, Message<?> msg) {
Object messageConverted = messageConverter.fromMessage(msg, valueClass);
if (messageConverted == null) {
throw new IllegalStateException("Inbound data conversion failed.");
}
keyValueThreadLocal.set(new KeyValue<>(o, messageConverted));
Object result = valueClass.isAssignableFrom(msg.getPayload().getClass())
? msg.getPayload() : messageConverter.fromMessage(msg, valueClass);
Assert.notNull(result, "Failed to convert message " + msg);
keyValueThreadLocal.set(new KeyValue<>(o, result));
}
@SuppressWarnings("unchecked")
@SuppressWarnings({ "unchecked", "rawtypes" })
private void processErrorFromDeserialization(KStream<?, ?> bindingTarget, KStream<?, ?> branch) {
branch.process(() -> new Processor() {
ProcessorContext context;
@@ -231,12 +225,15 @@ public class KafkaStreamsMessageConversionDelegate {
if (o2 instanceof Message) {
Message message = (Message) o2;
sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) message.getPayload(), context.partition());
} else {
}
else {
sendToDlqAndContinue.sendToDlq(destination, (byte[]) o, (byte[]) o2, context.partition());
}
} else if (kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
}
else if (kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) {
throw new IllegalStateException("Inbound deserialization failed.");
} else if (kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) {
}
else if (kstreamBinderConfigurationProperties.getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) {
//quietly pass through. No action needed, this is similar to log and continue.
}
}

13
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerde.java
View File

@@ -69,11 +69,10 @@ import org.springframework.util.MimeTypeUtils;
* can autowire that bean. This is the expected usage pattern of this class.
*
* @author Soby Chacko
* @since 2.1
*/
public class CompositeNonNativeSerde<T> implements Serde<T> {
private static final String CONTENT_TYPE_HEADER = "contentType";
private static final String VALUE_CLASS_HEADER = "valueClass";
private static final String AVRO_FORMAT = "avro";
@@ -111,8 +110,8 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
}
private static MimeType resolveMimeType(Map<String, ?> configs) {
if (configs.containsKey(CONTENT_TYPE_HEADER)){
String contentType = (String)configs.get(CONTENT_TYPE_HEADER);
if (configs.containsKey(MessageHeaders.CONTENT_TYPE)){
String contentType = (String)configs.get(MessageHeaders.CONTENT_TYPE);
if (DEFAULT_AVRO_MIME_TYPE.equals(MimeTypeUtils.parseMimeType(contentType))) {
return DEFAULT_AVRO_MIME_TYPE;
}
@@ -158,11 +157,9 @@ public class CompositeNonNativeSerde<T> implements Serde<T> {
@Override
public U deserialize(String topic, byte[] data) {
Message<?> message = MessageBuilder.withPayload(data)
.setHeader(CONTENT_TYPE_HEADER, this.mimeType.toString()).build();
.setHeader(MessageHeaders.CONTENT_TYPE, this.mimeType.toString()).build();
U messageConverted = (U)messageConverter.fromMessage(message, this.valueClass);
if (messageConverted == null) {
throw new IllegalStateException("Deserialization failed.");
}
Assert.notNull(messageConverted, "Deserialization failed.");
return messageConverted;
}