Skip to content

STORM-2826: Set key/value deserializer fields when using the convenience builder methods in KafkaSpoutConfig #2428

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
asfgit merged 1 commit into from
Dec 16, 2017
Merged

STORM-2826: Set key/value deserializer fields when using the convenience builder methods in KafkaSpoutConfig #2428

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 46 additions & 46 deletions external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,64 +280,64 @@ public Builder(String bootstrapServers, Subscription subscription) {

private Builder(String bootstrapServers, SerializableDeserializer<K> keyDes, Class<? extends Deserializer<K>> keyDesClazz,
SerializableDeserializer<V> valDes, Class<? extends Deserializer<V>> valDesClazz, Subscription subscription) {
kafkaProps = new HashMap<>();

this(keyDes, keyDesClazz, valDes, valDesClazz, subscription,
new DefaultRecordTranslator<K, V>(), new HashMap<String, Object>());

if (bootstrapServers == null || bootstrapServers.isEmpty()) {
throw new IllegalArgumentException("bootstrap servers cannot be null");
}
kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
this.keyDes = keyDes;
this.keyDesClazz = keyDesClazz;
this.valueDes = valDes;
this.valueDesClazz = valDesClazz;
this.subscription = subscription;
this.translator = new DefaultRecordTranslator<>();

if (keyDesClazz != null) {
this.kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDesClazz);
}
if (keyDes != null) {
this.kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDes.getClass());
}
if (valueDesClazz != null) {
this.kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDesClazz);
}
if (valueDes != null) {
this.kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDes.getClass());
}

setNonNullSerDeKafkaProp(keyDes, keyDesClazz, valueDes, valueDesClazz);
}

private Builder(Builder<?, ?> builder, SerializableDeserializer<K> keyDes, Class<? extends Deserializer<K>> keyDesClazz,
SerializableDeserializer<V> valueDes, Class<? extends Deserializer<V>> valueDesClazz) {
this.kafkaProps = new HashMap<>(builder.kafkaProps);
this.subscription = builder.subscription;
/**
* This constructor will always be called by one of the methods {@code setKey} or {@code setVal}, which implies
* that only one of its SerDe parameters will be non null, for which the corresponding Kafka property will be set
*/
@SuppressWarnings("unchecked")
private Builder(final Builder<?, ?> builder, SerializableDeserializer<K> keyDes, Class<? extends Deserializer<K>> keyDesClazz,
SerializableDeserializer<V> valueDes, Class<? extends Deserializer<V>> valueDesClazz) {

this(keyDes, keyDesClazz, valueDes, valueDesClazz, builder.subscription,
(RecordTranslator<K, V>) builder.translator, new HashMap<>(builder.kafkaProps));

this.pollTimeoutMs = builder.pollTimeoutMs;
this.offsetCommitPeriodMs = builder.offsetCommitPeriodMs;
this.firstPollOffsetStrategy = builder.firstPollOffsetStrategy;
this.maxUncommittedOffsets = builder.maxUncommittedOffsets;
//this could result in a lot of class case exceptions at runtime,
// but because some translators will work no matter what the generics
// are I thought it best not to force someone to reset the translator
// when they change the key/value types.
this.translator = (RecordTranslator<K, V>) builder.translator;
this.retryService = builder.retryService;


setNonNullSerDeKafkaProp(keyDes, keyDesClazz, valueDes, valueDesClazz);
}

private Builder(SerializableDeserializer<K> keyDes, Class<? extends Deserializer<K>> keyDesClazz,
SerializableDeserializer<V> valueDes, Class<? extends Deserializer<V>> valueDesClazz,
Subscription subscription, RecordTranslator<K, V> translator, Map<String, Object> kafkaProps) {
this.keyDes = keyDes;
this.keyDesClazz = keyDesClazz;
this.valueDes = valueDes;
this.valueDesClazz = valueDesClazz;
this.subscription = subscription;
this.translator = translator;
this.kafkaProps = kafkaProps;
}

private void setNonNullSerDeKafkaProp(SerializableDeserializer<K> keyDes, Class<? extends Deserializer<K>> keyDesClazz,
SerializableDeserializer<V> valueDes, Class<? extends Deserializer<V>> valueDesClazz) {
if (keyDesClazz != null) {
this.kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDesClazz);
kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDesClazz);
}
if (keyDes != null) {
this.kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDes.getClass());
kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDes.getClass());
}
if (valueDesClazz != null) {
this.kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDesClazz);
kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDesClazz);
}
if (valueDes != null) {
this.kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDes.getClass());
kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDes.getClass());
}

this.keyDes = keyDes;
this.keyDesClazz = keyDesClazz;
this.valueDes = valueDes;
this.valueDesClazz = valueDesClazz;
}

/**
Expand All @@ -348,7 +348,7 @@ private Builder(Builder<?, ?> builder, SerializableDeserializer<K> keyDes, Class
*/
@Deprecated
public <NK> Builder<NK, V> setKey(SerializableDeserializer<NK> keyDeserializer) {
return new Builder<>(this, keyDeserializer, null, valueDes, valueDesClazz);
return new Builder<>(this, keyDeserializer, null, null, null);
}

/**
Expand All @@ -359,7 +359,7 @@ public <NK> Builder<NK, V> setKey(SerializableDeserializer<NK> keyDeserializer)
*/
@Deprecated
public <NK> Builder<NK, V> setKey(Class<? extends Deserializer<NK>> clazz) {

@hmcl hmcl Dec 9, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@srdo do you know why this method returns a new builder object? I can't figure a reason for it to so. I suspect that the only reason for that to happen is because the fields of the builder class are final (e.g keyDesClassClazz), and to make the generics work. There is no benefit in having fields inside the builder class to be final. The code snippet bellow also fixes the generics problem. Any reason not to get rid of the builder (with copy constructor) class completely and make this method like this:

public Builder<K,V> setKey(Class<? extends Deserializer<K>> clazz) {
            this.keyDesClazz = clazz;
            if (keyDesClazz != null) {
                this.kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDesClazz);
            }
            return this;
        }

We should do something similar to the other 3 methods. In my opinion has become a bit confusing, and I believe this is one of the last few opportunities we have to make it better. Please let me know your thoughts. Thanks.

@srdo srdo Dec 9, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, the reasons you mention are the reasons these methods return new builders.

I agree that there is no reason for the fields to be final.

The snippet breaks the ability to change the key/value deserializer types. The existing code allows you to do e.g.

@Test
    public void test() {
       
        KafkaSpoutConfig<String, byte[]> conf = 
            //Use default <String, String> Builder
            KafkaSpoutConfig.builder("localhost:1234", "topic") 
            //Change to byte array value deserializer
            .setValue(ByteArrayDeserializer.class)
            .build();
    }

which now fails to compile because the new type bound on setKey/setValue prevents changing from Deserializer<String> to Deserializer<byte[]>.

You're right that the existing API is somewhat confusing, but it's getting removed in 2.0.0 and is deprecated here. I'd rather not add breaking changes to it if we can avoid it, because if we were going to break the API in a minor version I think we should just have removed the deprecated methods entirely.

return new Builder<>(this, null, clazz, valueDes, valueDesClazz);
return new Builder<>(this, null, clazz, null, null);
}

/**
Expand All @@ -370,7 +370,7 @@ public <NK> Builder<NK, V> setKey(Class<? extends Deserializer<NK>> clazz) {
*/
@Deprecated
public <NV> Builder<K, NV> setValue(SerializableDeserializer<NV> valueDeserializer) {
return new Builder<>(this, keyDes, keyDesClazz, valueDeserializer, null);
return new Builder<>(this, null, null, valueDeserializer, null);
}

/**
Expand All @@ -381,7 +381,7 @@ public <NV> Builder<K, NV> setValue(SerializableDeserializer<NV> valueDeserializ
*/
@Deprecated
public <NV> Builder<K, NV> setValue(Class<? extends Deserializer<NV>> clazz) {
return new Builder<>(this, keyDes, keyDesClazz, null, clazz);
return new Builder<>(this, null, null, null, clazz);
}

/**
Expand Down Expand Up @@ -680,7 +680,7 @@ public KafkaSpoutConfig<K, V> build() {
* @return The new builder
*/
public static Builder<String, String> builder(String bootstrapServers, String... topics) {
return setStringDeserializers(new Builder<String, String>(bootstrapServers, topics));
return setStringDeserializers(new Builder<String, String>(bootstrapServers, StringDeserializer.class, StringDeserializer.class, topics));
}

/**
Expand All @@ -691,7 +691,7 @@ public static Builder<String, String> builder(String bootstrapServers, String...
* @return The new builder
*/
public static Builder<String, String> builder(String bootstrapServers, Collection<String> topics) {
return setStringDeserializers(new Builder<String, String>(bootstrapServers, topics));
return setStringDeserializers(new Builder<String, String>(bootstrapServers, StringDeserializer.class, StringDeserializer.class, topics));
}

/**
Expand All @@ -702,7 +702,7 @@ public static Builder<String, String> builder(String bootstrapServers, Collectio
* @return The new builder
*/
public static Builder<String, String> builder(String bootstrapServers, Pattern topics) {
return setStringDeserializers(new Builder<String, String>(bootstrapServers, topics));
return setStringDeserializers(new Builder<String, String>(bootstrapServers, StringDeserializer.class, StringDeserializer.class, topics));
}

private static Builder<String, String> setStringDeserializers(Builder<String, String> builder) {
Expand Down
102 changes: 102 additions & 0 deletions ...l/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
*/
package org.apache.storm.kafka.spout;

import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.junit.Assert.assertEquals;
Expand All @@ -25,9 +26,12 @@
import static org.junit.Assert.assertTrue;

import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy;
import org.hamcrest.CoreMatchers;
import org.junit.Test;

public class KafkaSpoutConfigTest {
Expand Down Expand Up @@ -121,4 +125,102 @@ public void testCanConfigureWithExplicitFalseStringAutoCommitMode() {
assertThat("When setting enable auto commit explicitly to false the spout should use the 'at-least-once' processing guarantee",
conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE));
}

@Test
public void testCanGetKeyDeserializerWhenUsingDefaultBuilder() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.build();

assertThat("When using the default builder methods, the key deserializer should default to StringDeserializer",
conf.getKeyDeserializer(), instanceOf(StringDeserializer.class));
}

@Test
public void testCanGetValueDeserializerWhenUsingDefaultBuilder() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.build();

assertThat("When using the default builder methods, the value deserializer should default to StringDeserializer",
conf.getValueDeserializer(), instanceOf(StringDeserializer.class));
}

@Test
public void testCanOverrideDeprecatedDeserializerClassWithKafkaProps() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.setKey(StringDeserializer.class)
.setValue(StringDeserializer.class)
.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.build();

assertThat("The last set key deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
assertThat("The last set value deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
}

private static class SerializableStringDeserializer implements SerializableDeserializer {

private final StringDeserializer delegate = new StringDeserializer();

@Override
public void configure(Map configs, boolean isKey) {
delegate.configure(configs, isKey);
}

@Override
public Object deserialize(String topic, byte[] data) {
return delegate.deserialize(topic, data);
}

@Override
public void close() {
delegate.close();
}
}

@Test
public void testCanOverrideDeprecatedDeserializerInstanceWithKafkaProps() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.setKey(new SerializableStringDeserializer())
.setValue(new SerializableStringDeserializer())
.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.build();

assertThat("The last set key deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
assertThat("The last set value deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
}

@Test
public void testCanOverrideKafkaPropsWithDeprecatedDeserializerSetter() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.setKey(new SerializableStringDeserializer())
.setValue(new SerializableStringDeserializer())
.build();

assertThat("The last set key deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class));
assertThat("The last set value deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class));
}

@Test
public void testCanMixOldAndNewDeserializerSetter() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.setKey(new SerializableStringDeserializer())
.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
.setValue(new SerializableStringDeserializer())
.build();

assertThat("The last set key deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
assertThat("The last set value deserializer should be used, regardless of how it is set",
conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class));
}
}