4
\$\begingroup\$

I'm building an application that's going to create different autocomplete suggestions of data that sits in Kafka using Kafka Streams application.

Data is read from a relational database and looks as follows:

+--------------------------------------+--------------+--------------+-----------+---------------------------+---------------+-------------+----------------+---------------+----------------+--------------+-------------------------------------------------------------------------------------+-----------+-----------+--------------------------+---------+-----------+
|                  Id                  | IndustryCode | IndustryName | ProductId |        ProductName        | CountryCodeId | CountryCode |  CountryName   | MeasureTypeId |  MeasureType   | ResultTypeId |                                     Description                                     | SortOrder | CompanyId |       CompanyName        | BrandId | BrandName |
+--------------------------------------+--------------+--------------+-----------+---------------------------+---------------+-------------+----------------+---------------+----------------+--------------+-------------------------------------------------------------------------------------+-----------+-----------+--------------------------+---------+-----------+
| 2794C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183866 | Spectacles                |            32 | 2A          | Eastern Europe |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2894C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183866 | Spectacles                |           122 | 6A          | Australasia    |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2994C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183859 | Eyewear                   |           389 | WO          | World          |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2A94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183869 | Spectacle Lenses          |           389 | WO          | World          |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2B94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183869 | Spectacle Lenses          |           313 | MX          | Mexico         |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2C94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183866 | Spectacles                |           263 | ID          | Indonesia      |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2D94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183859 | Eyewear                   |           105 | 5A          | Asia Pacific   |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2E94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183866 | Spectacles                |           393 | ZA          | South Africa   |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 2F94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183866 | Spectacles                |           191 | CH          | Switzerland    |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 3094C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183869 | Spectacle Lenses          |           265 | IN          | India          |             4 | Company Shares |            1 | Share of sales and actual sales by company in a time series by standard data types. |       400 |       895 | Essilor International SA | NULL    | NULL      |
| 3A94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183859 | Eyewear                   |           313 | MX          | Mexico         |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 3B94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183867 | Readymade Reading Glasses |            88 | 4A          | Latin America  |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 3C94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183866 | Spectacles                |           313 | MX          | Mexico         |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 3D94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183867 | Readymade Reading Glasses |           389 | WO          | World          |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 3E94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183870 | Sunglasses                |            88 | 4A          | Latin America  |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 3F94C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183870 | Sunglasses                |           313 | MX          | Mexico         |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 4094C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183870 | Sunglasses                |            32 | 2A          | Eastern Europe |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 4194C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183870 | Sunglasses                |           393 | ZA          | South Africa   |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 4294C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183867 | Readymade Reading Glasses |           313 | MX          | Mexico         |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
| 4394C3D9-8EE1-E711-80ED-00155D30AF63 | EYE          | Eyewear      |    183870 | Sunglasses                |           182 | BR          | Brazil         |           119 | Brand Shares   |            1 | Share of sales and actual sales by brand in a time series by standard data types.   |       500 |    470225 | Avon Products Inc        | 14566   | Avon      |
+--------------------------------------+--------------+--------------+-----------+---------------------------+---------------+-------------+----------------+---------------+----------------+--------------+-------------------------------------------------------------------------------------+-----------+-----------+--------------------------+---------+-----------+

I'm transforming this into this kind of structure:

{
    "Importance": 2,
    "SortOrder": 500,
    "Suggestion": "Avon Products Inc in Eyewear",
    "CompanyId": 470225,
    "BrandId": null,
    "MeasureTypeId": 119,
    "MeasureType": "Brand Shares",
    "Description": "Share of sales and actual sales by brand in a time series by standard data types.",
    "Markets": [
        {
            "ProductId": 183859,
            "CountryCodeId": 313,
            "CountryCode": "MX"
        },
        {
            "ProductId": 183867,
            "CountryCodeId": 88,
            "CountryCode": "4A"
        },
        {
            "ProductId": 183866,
            "CountryCodeId": 313,
            "CountryCode": "MX"
        },
        {
            "ProductId": 183867,
            "CountryCodeId": 389,
            "CountryCode": "WO"
        },
        {
            "ProductId": 183870,
            "CountryCodeId": 88,
            "CountryCode": "4A"
        },
        {
            "ProductId": 183870,
            "CountryCodeId": 313,
            "CountryCode": "MX"
        },
        {
            "ProductId": 183870,
            "CountryCodeId": 32,
            "CountryCode": "2A"
        },
        {
            "ProductId": 183870,
            "CountryCodeId": 393,
            "CountryCode": "ZA"
        },
        {
            "ProductId": 183867,
            "CountryCodeId": 313,
            "CountryCode": "MX"
        },
        {
            "ProductId": 183870,
            "CountryCodeId": 182,
            "CountryCode": "BR"
        },
        {
            "ProductId": 183870,
            "CountryCodeId": 344,
            "CountryCode": "RU"
        }
    ]
}

Where key now is (MeasureTypeId, CompanyId, IndustryCode) and all (Product, CountryCodeId, CountryCode) combinations are put into Markets[] (there are different keys for different kind of autocomplete suggestions)

My application works and I just need a second opinion on what could be done better. I already knew that I'll have more than one kind of autocomplete, so I did it that way:

  1. Declared an abstract class that has methods needed for aggregator
  2. Aggregators should extend this class and implement
  3. My topology just consumes a set of these aggregators and creates Kafka Stream based on their implementations

My abstract class:

package com.euromonitor.kafka.streams.suggestion;

import com.euromonitor.kafka.avro.elasticsearch.Market;
import com.euromonitor.kafka.avro.elasticsearch.Suggestion;
import com.euromonitor.kafka.avro.mssql.StatisticKey;
import com.euromonitor.kafka.avro.mssql.StatisticValue;
import com.google.common.collect.ImmutableList;
import org.apache.kafka.streams.KeyValue;

import java.util.List;
import java.util.Objects;
import java.util.function.Predicate;

public abstract class Aggregator {

  /**
   * Determines whether a suggestion has to be built from a record
   *
   * @param key UUID primary key of Elasticsearch.Statistic table
   * @param value a set of values that come from Elasticsearch.Statistic table. An accumulator
   * @return true or false
   */
  protected abstract boolean filterValues(StatisticKey key, StatisticValue value);

  /**
   * Builds new key that will be used for aggregations
   *
   * @param key UUID primary key of Elasticsearch.Statistic table
   * @param value a set of values that come from Elasticsearch.Statistic table. An accumulator
   * @return new key and old value
   */
  protected abstract KeyValue<String, StatisticValue> createKey(
      StatisticKey key, StatisticValue value);

  /**
   * Builds suggestion based on business criteria
   *
   * @param value a set of values that come from Elasticsearch.Statistic table. An accumulator
   * @return suggestion string to be displayed for users
   */
  protected abstract String buildSuggestion(StatisticValue value);

  /**
   * Used in KTables aggregate method. Will occur when a new record is being added or updated
   *
   * @param key UUID primary key of Elasticsearch.Statistic table
   * @param value a set of values that come from Elasticsearch.Statistic table. An accumulator
   * @param aggregate current Suggestion value before modifying it with accumulator data
   * @return a new Suggestion which is (aggregate + value)
   */
  protected abstract Suggestion add(String key, StatisticValue value, Suggestion aggregate);

  /**
   * Used in KTables aggregate method Creates initial value for Suggestion if there's a new key in
   * aggregation
   *
   * @return an empty Suggestion object with default values for further processing
   */
  protected abstract Suggestion initialize();

  /**
   * Used in KTables aggregate method. Will occur when a new record is being removed or updated
   *
   * @param key UUID primary key of Elasticsearch.Statistic table
   * @param value a set of values that come from Elasticsearch.Statistic table. An accumulator
   * @param aggregate current Suggestion value before modifying it with accumulator data
   * @return a new Suggestion which is (aggregate - value)
   */
  @SuppressWarnings("unused")
  protected Suggestion subtract(String key, StatisticValue value, Suggestion aggregate) {
    return Suggestion.newBuilder(aggregate)
        .setMarkets(
            aggregate
                .getMarkets()
                .stream()
                .filter(isDeleted(marketOf(value)))
                .collect(ImmutableList.toImmutableList()))
        .build();
  }

  private Predicate<Market> isDeleted(Market market) {
    return p -> !Objects.equals(p, market);
  }

  protected static Market marketOf(StatisticValue value) {
    return new Market(value.getProductId(), value.getCountryCodeId(), value.getCountryCode());
  }

  protected static List<Market> addMarket(Suggestion current, Market market) {
    return ImmutableList.<Market>builder().addAll(current.getMarkets()).add(market).build();
  }
}

Implementation of one of aggregators:

package com.euromonitor.kafka.streams.suggestion.aggregators;

import com.euromonitor.kafka.avro.elasticsearch.Suggestion;
import com.euromonitor.kafka.avro.mssql.StatisticKey;
import com.euromonitor.kafka.avro.mssql.StatisticValue;
import com.euromonitor.kafka.streams.suggestion.Aggregator;
import com.google.common.collect.ImmutableList;
import org.apache.kafka.streams.KeyValue;

public class CompanyIndustry extends Aggregator {
  @Override
  protected boolean filterValues(StatisticKey key, StatisticValue value) {
    return value.getCompanyId() != null;
  }

  @Override
  protected KeyValue<String, StatisticValue> createKey(StatisticKey key, StatisticValue value) {
    return KeyValue.pair(
        String.format(
            "company_industry_%d_%d_%s",
            value.getMeasureTypeId(), value.getCompanyId(), value.getIndustryCode()),
        value);
  }

  @Override
  protected String buildSuggestion(StatisticValue value) {
    return String.format("%s in %s", value.getCompanyName(), value.getIndustryName());
  }

  @Override
  protected Suggestion initialize() {
    return Suggestion.newBuilder()
        .setImportance(2)
        .setSortOrder(0)
        .setSuggestion("")
        .setMeasureTypeId(0)
        .setMeasureType("")
        .setMarkets(ImmutableList.of())
        .build();
  }

  @Override
  protected Suggestion add(String key, StatisticValue value, Suggestion aggregate) {
    return Suggestion.newBuilder(aggregate)
        .setSortOrder(value.getSortOrder())
        .setCompanyId(value.getCompanyId())
        .setDescription(value.getDescription())
        .setMeasureTypeId(value.getMeasureTypeId())
        .setMeasureType(value.getMeasureType())
        .setSuggestion(buildSuggestion(value))
        .setMarkets(addMarket(aggregate, marketOf(value)))
        .build();
  }
}

All of them look quite similar, except that they have to aggregate by different values.

This is my class that accepts set of aggregators and configuration and builds stream topology:

package com.euromonitor.kafka.streams.suggestion;

import com.euromonitor.kafka.avro.mssql.StatisticKey;
import com.euromonitor.kafka.avro.mssql.StatisticValue;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.KTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.Set;

final class Topology {
  private static final Logger LOGGER = LoggerFactory.getLogger(Topology.class);

  private Topology() {}

  static KafkaStreams of(
      Set<Aggregator> aggregators, Serialization serialization, Configuration configuration) {
    final KStreamBuilder builder = new KStreamBuilder();

    final Properties streamConfig = configuration.streamConfig;

    LOGGER.trace("Initializing stream");

    KTable<StatisticKey, StatisticValue> statistic =
        builder.table(
            serialization.sourceKeySerde,
            serialization.sourceValueSerde,
            configuration.sourceTopicName);

    aggregators.forEach(
        aggregator ->
            statistic
                .filter(aggregator::filterValues)
                .groupBy(
                    aggregator::createKey,
                    serialization.targetKeySerde,
                    serialization.sourceValueSerde)
                .aggregate(
                    aggregator::initialize,
                    aggregator::add,
                    aggregator::subtract,
                    serialization.targetValueSerde)
                .to(
                    serialization.targetKeySerde,
                    serialization.targetValueSerde,
                    configuration.targetTopicName));

    return new KafkaStreams(builder, streamConfig);
  }
}

And last but not least, this is the main one where I just instantiate everything:

package com.euromonitor.kafka.streams.suggestion;

import com.euromonitor.kafka.streams.suggestion.aggregators.*;
import com.google.common.collect.ImmutableSet;
import org.apache.kafka.streams.KafkaStreams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;

public class Main {
  private static final Logger LOGGER = LoggerFactory.getLogger(Main.class.getSimpleName());

  public static void main(String... args) {
    LOGGER.info("Loaded streamConfiguration. Ready to shape data!");

    start();
  }

  private static void start() {
    final Configuration configuration = new Configuration();
    final Serialization serialization = new Serialization(configuration.schemaRegistryConfig);

    final Set<Aggregator> aggregators =
        ImmutableSet.of(
            new ProductCountry(),
            new CompanyIndustry(),
            new CompanyCountry(),
            new BrandIndustry(),
            new BrandCountry());

    final KafkaStreams topology = Topology.of(aggregators, serialization, configuration);

    topology.cleanUp();
    topology.start();

    Runtime.getRuntime().addShutdownHook(new Thread(topology::close));
  }
}

Therefore I am primarily looking for advice and suggestions on what could've been done better in regards to the code structure and method implementation, ease of code understanding

\$\endgroup\$
2
  • 1
    \$\begingroup\$ @Vogel612 that doesn't seem like a very useful edit to me, lol (?) \$\endgroup\$
    – Daniel
    Dec 29 '17 at 12:31
  • \$\begingroup\$ @Coal_ doesn't seem one either, but he has put a "punny title" into his edit comment, so I'd just like to see what he had in mind. This article suggests that Gregor Samsa has turned into a monstrous vermin. One of Franz Kafka characters in his book Metamorphosis. \$\endgroup\$ Dec 29 '17 at 12:40
5
\$\begingroup\$

Looks quite decent.

What I don't like is the abstract class. I actually don't see a reason to use abstraction. There's static and non-static methods. For me it looks like, the non-static methods can be static too, since they do not use any state. And therefore all implemented methods do not need an actual instance. An interface should be fine. Considering testing: It's much better.

The parameter String key is used often. And described in the javadoc. I'd either rename it ElasticsearchStatisticKeyUUID or something - which is actually bad, because the caller shouldn't know that ElasticSearch is behind it, imo. Implementation shouldn't be only hidden by design, but also in JavaDoc. Or I would introduce a separate type for the Key. So, if one day you have to use e.g. the UUID type as parameter, your signatures stay the same. And also, there's a StatisticKey, which doesn't really help - which one is which?

The JavaDoc: "Determines whether a suggestion has to be built from a record". But the method is called filterValues. Well, if it "determines" something, it just figures something out. But if it "filters" something, well, it does filter something.

The JavaDoc for the subtract method: "Used in KTables aggregate method.". Uhm. Well, if you have to write in your JavaDoc, that a method is called from somewhere else, something's wrong. Is it public for other users? Why is it so important to know, that it's called from somewhere else? It also says: "@return a new Suggestion which is (aggregate + value)". Wouldn't a method name addValueToAggregate be better?

Then, JavaDoc "Builds new key that will be used for aggregations". Methodname: createKey. Next method: "Builds suggestion based on business criteria". Methodname: buildSuggestion. Also: "@return new key and old value". That's a bit confusing. I want to create a key, receive a the new key and the old value? Huh?

Also, JavaDoc for initialize: "Used in KTables aggregate method Creates initial value for Suggestion if there's a new key in aggregation". What if not? "return an empty Suggestion object with default values for further processing" - why should an initialize method return something? It then is at least two things, the method doesn. Maybe createInitialSuggestion?

JavaDoc for add(): "aggregate current Suggestion value before modifying it with accumulator data" - well, the method name says add, but it does also modify?

Why is Market marketOf(StatisticValue value) in the Aggregator type? Is it really the Aggregator's job to instantiate types?

LOGGER.info("Loaded streamConfiguration. Ready to shape data!"); -> That's a lie. The program didn't do anything, really, in the first line of the main method :P

In general: I'd get rid of all JavaDocs. If filterValues is not clear, the method name sucks - especially if it doesn't filter. If it's not clear, what a parameter must do, the parameter name sucks. Or you have too many parameters. If I want to call a method or I'm looking for one, I usually list the methods of an object in my IDE. If I don't get what a method does, and have to read the JavaDoc, which leads to adding up my confusion, I will have to read the implementation. And both - having to read the JavaDoc and to read the implementation - usually tells me, that the API is not well written.

\$\endgroup\$
4
  • \$\begingroup\$ These are handful insights! Thanks a lot! \$\endgroup\$ Dec 31 '17 at 9:11
  • \$\begingroup\$ So for instance renaming filterValues to shouldBuildSuggestion would make more sense? \$\endgroup\$ Dec 31 '17 at 10:00
  • 2
    \$\begingroup\$ I'd argue against removing Javadoc generally, since it can be a useful extension to clarify a method name. It also should help establish exact semantics (e.g. difference between pop and remove or add, enqueue and put). \$\endgroup\$
    – Vogel612
    Jan 1 '18 at 14:40
  • \$\begingroup\$ @Vogel612 I agree to a certain extent. But from experience, the quality of the javadoc is only ensured organisational, e.g. code reviews and/or checkstyle or the like, and if you have a highly motivated team to do so. It's a lot of effort a team has to agree to make. If that's not the case, it's more of a tool to tangle people. JavaDoc can also lead to a lot of in code clutter. Or even bugs, if people forget to update the javadoc. But anyway, to actually write javadoc is opinion based and in the end, it is a team decision. \$\endgroup\$
    – slowy
    Jan 2 '18 at 10:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.