diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 646d1312e..0fe626e12 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -1579,7 +1579,6 @@ By default, the `Kafkastreams.cleanup()` method is called when the binding is st See https://docs.spring.io/spring-kafka/reference/html/_reference.html#_configuration[the Spring Kafka documentation]. To modify this behavior simply add a single `CleanupConfig` `@Bean` (configured to clean up on start, stop, or neither) to the application context; the bean will be detected and wired into the factory bean. - === Kafka Streams topology visualization Kafka Streams binder provides the following actuator endpoints for retrieving the topology description using which you can visualize the topology using external tools. @@ -1592,6 +1591,39 @@ You need to include the actuator and web dependencies from Spring Boot to access Further, you also need to add `kafkastreamstopology` to `management.endpoints.web.exposure.include` property. By default, the `kafkastreamstopology` endpoint is disabled. +=== Event type based routing in Kafka Streams applications + +Routing functions available in regular message channel based binders are not supported in Kafka Streams binder. +However, Kafka Streams binder still provides routing capabilities through the event type record header on the inbound records. + +To enable routing based on event types, the application must provide the following property. + +`spring.cloud.stream.kafka.streams.bindings..consumer.eventTypes`. + +This can be a comma separated value. + +For example, lets assume we have this function: + +``` +@Bean +public Function, KStream> process() { + return input -> input; +} +``` + +Let us also assume that we only want the business logic in this function to be executed, if the incoming record has event types as `foo` or `bar`. +That can be expressed as below using the `eventTypes` property on the binding. + +`spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypes=foo,bar` + +Now, when the application runs, the binder checks each incoming records for the header `event_type` and see if it has value set as `foo` or `bar`. +If it does not find either of them, then the function execution will be skipped. + +By default, the binder expects the record header key to be `event_type`, but that can be changed per binding. +For instance, if we want to change the header key on this binding to `my_event` instead of the default, that can be changed as below. + +`spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypeHeaderKey=my_event`. + === Configuration Options This section contains the configuration options used by the Kafka Streams binder. @@ -1603,9 +1635,9 @@ For common configuration options and properties pertaining to binder, refer to t The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.` configuration:: - Map with a key/value pair containing properties pertaining to Apache Kafka Streams API. - This property must be prefixed with `spring.cloud.stream.kafka.streams.binder.`. - Following are some examples of using this property. +Map with a key/value pair containing properties pertaining to Apache Kafka Streams API. +This property must be prefixed with `spring.cloud.stream.kafka.streams.binder.`. +Following are some examples of using this property. [source] ---- @@ -1621,56 +1653,56 @@ If you have more than processors in the application, all of them will acquire th In the case of properties like `application.id`, this will become problematic and therefore you have to carefully examine how the properties from `StreamsConfig` are mapped using this binder level `configuration` property. functions..applicationId:: - Applicable only for functional style processors. - This can be used for setting application ID per function in the application. - In the case of multiple functions, this is a handy way to set the application ID. +Applicable only for functional style processors. +This can be used for setting application ID per function in the application. +In the case of multiple functions, this is a handy way to set the application ID. functions..configuration:: - Applicable only for functional style processors. - Map with a key/value pair containing properties pertaining to Apache Kafka Streams API. - This is similar to the binder level `configuration` property describe above, but this level of `configuration` property is restricted only against the named function. - When you have multiple processors and you want to restrict access to the configuration based on particular functions, you might want to use this. - All `StreamsConfig` properties can be used here. +Applicable only for functional style processors. +Map with a key/value pair containing properties pertaining to Apache Kafka Streams API. +This is similar to the binder level `configuration` property describe above, but this level of `configuration` property is restricted only against the named function. +When you have multiple processors and you want to restrict access to the configuration based on particular functions, you might want to use this. +All `StreamsConfig` properties can be used here. brokers:: - Broker URL +Broker URL + Default: `localhost` zkNodes:: - Zookeeper URL +Zookeeper URL + Default: `localhost` deserializationExceptionHandler:: - Deserialization error handler type. - This handler is applied at the binder level and thus applied against all input binding in the application. - There is a way to control it in a more fine-grained way at the consumer binding level. - Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq` +Deserialization error handler type. +This handler is applied at the binder level and thus applied against all input binding in the application. +There is a way to control it in a more fine-grained way at the consumer binding level. +Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq` + Default: `logAndFail` applicationId:: - Convenient way to set the application.id for the Kafka Streams application globally at the binder level. - If the application contains multiple functions or `StreamListener` methods, then the application id should be set differently. - See above where setting the application id is discussed in detail. +Convenient way to set the application.id for the Kafka Streams application globally at the binder level. +If the application contains multiple functions or `StreamListener` methods, then the application id should be set differently. +See above where setting the application id is discussed in detail. + Default: application will generate a static application ID. See the application ID section for more details. stateStoreRetry.maxAttempts:: - Max attempts for trying to connect to a state store. +Max attempts for trying to connect to a state store. + Default: 1 stateStoreRetry.backoffPeriod:: - Backoff period when trying to connect to a state store on a retry. +Backoff period when trying to connect to a state store on a retry. + Default: 1000 ms consumerProperties:: - Arbitrary consumer properties at the binder level. +Arbitrary consumer properties at the binder level. producerProperties:: - Arbitrary producer properties at the binder level. +Arbitrary producer properties at the binder level. ==== Kafka Streams Producer Properties @@ -1678,23 +1710,23 @@ The following properties are _only_ available for Kafka Streams producers and mu For convenience, if there are multiple output bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.producer.`. keySerde:: - key serde to use +key serde to use + Default: See the above discussion on message de/serialization valueSerde:: - value serde to use +value serde to use + Default: See the above discussion on message de/serialization useNativeEncoding:: - flag to enable/disable native encoding +flag to enable/disable native encoding + Default: `true`. streamPartitionerBeanName: - Custom outbound partitioner bean name to be used at the consumer. - Applications can provide custom `StreamPartitioner` as a Spring bean and the name of this bean can be provided to the producer to use instead of the default one. +Custom outbound partitioner bean name to be used at the consumer. +Applications can provide custom `StreamPartitioner` as a Spring bean and the name of this bean can be provided to the producer to use instead of the default one. + Default: See the discussion above on outbound partition support. @@ -1704,40 +1736,40 @@ The following properties are available for Kafka Streams consumers and must be p For convenience, if there are multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.`. applicationId:: - Setting application.id per input binding. This is only preferred for `StreamListener` based processors, for function based processors see other approaches outlined above. +Setting application.id per input binding. This is only preferred for `StreamListener` based processors, for function based processors see other approaches outlined above. + Default: See above. keySerde:: - key serde to use +key serde to use + Default: See the above discussion on message de/serialization valueSerde:: - value serde to use +value serde to use + Default: See the above discussion on message de/serialization materializedAs:: - state store to materialize when using incoming KTable types +state store to materialize when using incoming KTable types + Default: `none`. useNativeDecoding:: - flag to enable/disable native decoding +flag to enable/disable native decoding + Default: `true`. dlqName:: - DLQ topic name. +DLQ topic name. + Default: See above on the discussion of error handling and DLQ. startOffset:: - Offset to start from if there is no committed offset to consume from. - This is mostly used when the consumer is consuming from a topic for the first time. - Kafka Streams uses `earliest` as the default strategy and the binder uses the same default. - This can be overridden to `latest` using this property. +Offset to start from if there is no committed offset to consume from. +This is mostly used when the consumer is consuming from a topic for the first time. +Kafka Streams uses `earliest` as the default strategy and the binder uses the same default. +This can be overridden to `latest` using this property. + Default: `earliest`. @@ -1745,18 +1777,28 @@ Note: Using `resetOffsets` on the consumer does not have any effect on Kafka Str Unlike the message channel based binder, Kafka Streams binder does not seek to beginning or end on demand. deserializationExceptionHandler:: - Deserialization error handler type. - This handler is applied per consumer binding as opposed to the binder level property described before. - Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq` +Deserialization error handler type. +This handler is applied per consumer binding as opposed to the binder level property described before. +Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq` + Default: `logAndFail` timestampExtractorBeanName:: - Specific time stamp extractor bean name to be used at the consumer. - Applications can provide `TimestampExtractor` as a Spring bean and the name of this bean can be provided to the consumer to use instead of the default one. +Specific time stamp extractor bean name to be used at the consumer. +Applications can provide `TimestampExtractor` as a Spring bean and the name of this bean can be provided to the consumer to use instead of the default one. + Default: See the discussion above on timestamp extractors. +eventTypes:: +Comma separated list of supported event types for this binding. ++ +Default: `none` + +eventTypeHeaderKey:: +Event type header key on each incoming records through this binding. ++ +Default: `event_type` + ==== Special note on concurrency In Kafka Streams, you can control of the number of threads a processor can create using the `num.stream.threads` property. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index b112c91bb..f5a21c57c 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -19,11 +19,14 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Arrays; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.regex.Pattern; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; @@ -37,6 +40,8 @@ import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.processor.Processor; +import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.TimestampExtractor; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; @@ -396,25 +401,13 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } } + @SuppressWarnings({"rawtypes", "unchecked"}) protected KStream getKStream(String inboundName, BindingProperties bindingProperties, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, - StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) { + StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) { if (firstBuild) { addStateStoreBeans(streamsBuilder); } - KStream stream; - if (this.kafkaStreamsExtendedBindingProperties - .getExtendedConsumerProperties(inboundName).isDestinationIsPattern()) { - final Pattern pattern = Pattern.compile(this.bindingServiceProperties.getBindingDestination(inboundName)); - stream = streamsBuilder.stream(pattern); - } - else { - String[] bindingTargets = StringUtils.commaDelimitedListToStringArray( - this.bindingServiceProperties.getBindingDestination(inboundName)); - final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerde, autoOffsetReset); - stream = streamsBuilder.stream(Arrays.asList(bindingTargets), - consumed); - } final boolean nativeDecoding = this.bindingServiceProperties .getConsumerProperties(inboundName).isUseNativeDecoding(); if (nativeDecoding) { @@ -426,6 +419,62 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application + ". Inbound message conversion done by Spring Cloud Stream."); } + KStream stream; + if (this.kafkaStreamsExtendedBindingProperties + .getExtendedConsumerProperties(inboundName).isDestinationIsPattern()) { + final Pattern pattern = Pattern.compile(this.bindingServiceProperties.getBindingDestination(inboundName)); + stream = streamsBuilder.stream(pattern); + } + else { + String[] bindingTargets = StringUtils.commaDelimitedListToStringArray( + this.bindingServiceProperties.getBindingDestination(inboundName)); + final Serde valueSerdeToUse = StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes()) ? + new Serdes.BytesSerde() : valueSerde; + final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerdeToUse, autoOffsetReset); + stream = streamsBuilder.stream(Arrays.asList(bindingTargets), + consumed); + } + //Check to see if event type based routing is enabled. + //See this issue for more context: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003 + if (StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes())) { + AtomicBoolean matched = new AtomicBoolean(); + // Processor to retrieve the header value. + stream.process(() -> new Processor() { + + ProcessorContext context; + + @Override + public void init(ProcessorContext context) { + this.context = context; + } + + @Override + public void process(Object key, Object value) { + final Headers headers = this.context.headers(); + final Iterable
eventTypeHeader = headers.headers(kafkaStreamsConsumerProperties.getEventTypeHeaderKey()); + if (eventTypeHeader != null && eventTypeHeader.iterator().hasNext()) { + String eventTypeFromHeader = new String(eventTypeHeader.iterator().next().value()); + final String[] eventTypesFromBinding = StringUtils.commaDelimitedListToStringArray(kafkaStreamsConsumerProperties.getEventTypes()); + for (String eventTypeFromBinding : eventTypesFromBinding) { + if (eventTypeFromHeader.equals(eventTypeFromBinding)) { + matched.set(true); + break; + } + } + } + } + + @Override + public void close() { + + } + }); + // Branching based on event type match. + final KStream[] branch = stream.branch((key, value) -> matched.getAndSet(false)); + // Deserialize if we have a branch from above. + final KStream deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize(null, ((Bytes) value).get())); + return getkStream(bindingProperties, deserializedKStream, nativeDecoding); + } return getkStream(bindingProperties, stream, nativeDecoding); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java index d0ba77836..a9a39b6cb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java @@ -54,6 +54,18 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { */ private String timestampExtractorBeanName; + /** + * Comma separated list of supported event types for this binding. + */ + private String eventTypes; + + /** + * Record level header key for event type. + * If the default value is overridden, then that is expected on each record header if eventType based + * routing is enabled on this binding (by setting eventTypes). + */ + private String eventTypeHeaderKey = "event_type"; + public String getApplicationId() { return this.applicationId; } @@ -101,4 +113,20 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { public void setDeserializationExceptionHandler(DeserializationExceptionHandler deserializationExceptionHandler) { this.deserializationExceptionHandler = deserializationExceptionHandler; } + + public String getEventTypes() { + return eventTypes; + } + + public void setEventTypes(String eventTypes) { + this.eventTypes = eventTypes; + } + + public String getEventTypeHeaderKey() { + return this.eventTypeHeaderKey; + } + + public void setEventTypeHeaderKey(String eventTypeHeaderKey) { + this.eventTypeHeaderKey = eventTypeHeaderKey; + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java new file mode 100644 index 000000000..4bddd0457 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java @@ -0,0 +1,190 @@ +/* + * Copyright 2019-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.function.Function; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerializer; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +public class KafkaStreamsEventTypeRoutingTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "foo-1", "foo-2"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() { + Map consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put("value.deserializer", JsonDeserializer.class); + consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "foo-2"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + //See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003 for more context on this test. + @Test + public void testRoutingWorksBasedOnEventTypes() { + SpringApplication app = new SpringApplication(EventTypeRoutingTestConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=process", + "--spring.cloud.stream.bindings.process-in-0.destination=foo-1", + "--spring.cloud.stream.bindings.process-out-0.destination=foo-2", + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypes=foo,bar", + "--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-foo-0", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put("value.serializer", JsonSerializer.class); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foo-1"); + Foo foo1 = new Foo(); + foo1.setFoo("foo-1"); + Headers headers = new RecordHeaders(); + headers.add(new RecordHeader("event_type", "foo".getBytes())); + + final ProducerRecord producerRecord1 = new ProducerRecord<>("foo-1", 0, 56, foo1, headers); + template.send(producerRecord1); + + Foo foo2 = new Foo(); + foo2.setFoo("foo-2"); + + final ProducerRecord producerRecord2 = new ProducerRecord<>("foo-1", 0, 57, foo2); + template.send(producerRecord2); + + Foo foo3 = new Foo(); + foo3.setFoo("foo-3"); + + final ProducerRecord producerRecord3 = new ProducerRecord<>("foo-1", 0, 58, foo3, headers); + template.send(producerRecord3); + + Foo foo4 = new Foo(); + foo4.setFoo("foo-4"); + Headers headers1 = new RecordHeaders(); + headers1.add(new RecordHeader("event_type", "bar".getBytes())); + + final ProducerRecord producerRecord4 = new ProducerRecord<>("foo-1", 0, 59, foo4, headers1); + template.send(producerRecord4); + + final ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + + assertThat(records.count()).isEqualTo(3); + + List keys = new ArrayList<>(); + List values = new ArrayList<>(); + + records.forEach(integerFooConsumerRecord -> { + keys.add(integerFooConsumerRecord.key()); + values.add(integerFooConsumerRecord.value()); + }); + + assertThat(keys).containsExactlyInAnyOrder(56, 58, 59); + assertThat(values).containsExactlyInAnyOrder(foo1, foo3, foo4); + } + finally { + pf.destroy(); + } + } + } + + @EnableAutoConfiguration + public static class EventTypeRoutingTestConfig { + + @Bean + public Function, KStream> process() { + return input -> input; + } + + } + + static class Foo { + String foo; + + public String getFoo() { + return foo; + } + + public void setFoo(String foo) { + this.foo = foo; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Foo foo1 = (Foo) o; + return Objects.equals(foo, foo1.foo); + } + + @Override + public int hashCode() { + return Objects.hash(foo); + } + } +}