KafkaEmitter for request emitting does not emit "queryType" for native queries

[BartMiki]

Affected Version

The Druid version where the problem was encountered: 0.22.x, 0.23.x

Description

When setting up a request emitter to use KafkaEmitter the queryType property is not serialized to a JSON. The problem is located in the method: public void emit(final Event event) of KafkaEmitter.java.

The problem is that the rendering of the JSON string is done on the Map<String, Object>. The Jackson ignores the polymorphic annotation on the Query class with queryType if the query is serialized as a part of the map (see: DefaultRequestLogEvent for explanation). If we serialize the query independently then the polymorphic annotation is respected.

Proposed solution

The simples solution is to serialize an object to JsonNode first. If the query is present, then serialize it independently and replace the non-polymorphic version in the JsonNode.

[rohangarg]

@BartMiki : Thanks a lot for reporting the issue! I tried it locally in unit-tests and was able to reproduce it. The root cause as you mentioned, is related to missing of type information while doing jackson seriallziation for polymorphic types in maps/arrays.
The solution you proposed sounds good to me conceptually. Maybe instead of just serializing query independently, it could also be checked if we can do independent serializing of all KV pair in the map and then recombine all the KV jsons.

It would be great if you are also interested in contributing the fix patch to the main line!

[BartMiki]

@rohangarg Thank you for the response! I'm interested in contributing, actually, I already fix the issue on my fork. I will link the PR to this Issue once I finish the unit tests. I like your idea to parse each KV pair individually, I'll do that.

[FrankChen021]

I'm wondering if there's any public method provided by jackson to allow us to keep the annotation on objects which are stored in Map or Array?

Serializing objects to JsonNode first and then serializing these JsonNodes into string, I'm worry about the performance of these two phase serialization.

[BartMiki]

Regarding Jackson, there is no such method. The problem occurs due to type erasure of generic classes (see this ticket).

Regarding the two-step serialization, I suggested a solution in the PR comment. The problem is that KafkaEmitter adds an additional property to the output event which complicates serialization a bit.

[rohangarg]

Thanks @FrankChen021 and @BartMiki for the thorough discussion. I have a few more points related to discussions :

1. Regarding performance, I think we can check once the time taken for old vs new approach to prepare the event to see we're not adding a lot of time in absolute terms atleast.
2. Regarding the problem for compatibility arising b/w serializing using annotation vs using asMap method and then serializing, it seems like that should be solved. Otherwise, the kind of JSON created for the same event object would differ according to the implementation of the Emitter as it is happening now. We could call out the incompat in release notes if needed.
   What do you think?

[BartMiki]

@rohangarg regarding your points:

1. Do you have any tips how to measure that? Any recommended profilers? Maybe just a unit test per serialization method that runs serialization in a loop 100k times?
2. Current fix is compatible with previous output so there is no need for incompat. Actually it is interesting issue in general, because what is the method for generating output for the event? What is recommended by the Druid? Serialize Event with Jackson or with .toMap? To .toMap is part of Event which seems to be default, but many core Druid emitters just use Jackson on the Event. If somebody implements new emitter what they would default to? Maybe that should be included in the docs? To be honest I don't have a solution for that at this moment. Maybe we should introduce something like EventToXSerializer interfaces where X is format name. Then we can also provide default implementations, like DefaultObjectMapper is provided. What do you think? But that should be part of a new issue.

[rohangarg]

@BartMiki :

1. Generally, for performance measurement we use JMH benchmarks in the druid-benchmarks module. But for this case, we only need to check once that there is no absurd increase in the time. So, the approach of local UT (no need to commit it) you mention sounds good to me.
2. I agree with your point that the recommended way to serialize an event should be a part of a new issue. For the current change though, I think it could be ok to have sql and sqlContext for KafkaEmitter too since the removal of those keys wasn't intended FWIU. All the rest of the emitters also have those keys (because they directly serialize Event), so having them would maintain parity. But I don't have a strong opinion on it - if you and @FrankChen021 think that maintaining the current compat is better then I'm good with it too.

If we choose to allow the new extra keys way, then we could also do the following code approach which maybe simpler :

1. convert the event object to a map using jsonMapper.convertValue(event, new TypeReference<Map<String, Object>>() {} - this map also contains the queryType
2. Put the additional keys in the map
3. Call jsonMapper.writeValueAsString(mergedMap) to get the JSON string

[FrankChen021]

If we want to fix the problem from root, I think we need to make changes to Event interface to make the special serialization process to be unvisible to callers.

My suggestion is as follows

public interface Event {
    ACustomerMap toMap();
    ...
}

class ACustomerMap extends HashMap<String, Object> {

     // a dedicated method to serialize this object into JSON string to avoid missing type info in the final text
     public String toJSONString() {
     }
}

Since ACustomerMap inherits Map<String, Object>, it makes current code where this method is called compatible with this new class.

[BartMiki]

Regarding the performance, I think there are no big differences. I made these test for both:
Old implementation:

  public void testOldCode() throws JsonProcessingException {
    ObjectMapper mapper = new DefaultObjectMapper();
    Event event = getEvent();

    for (int i = 0; i < LOOP_COUNT; i++) {
      ImmutableMap.Builder<String, Object> resultBuilder = ImmutableMap.<String, Object>builder().putAll(event.toMap());
      if (CLUSTER_NAME != null) {
        resultBuilder.put("clusterName", CLUSTER_NAME);
      }
      Map<String, Object> result = resultBuilder.build();
      String resultJson = mapper.writeValueAsString(result);
    }
  }

New implementation:

  public void testMyCode() throws JsonProcessingException
  {
    ObjectMapper mapper = new DefaultObjectMapper();
    Event event = getEvent();

    for (int i = 0; i < LOOP_COUNT; i++) {
      EventToJsonSerializer serializer = EventToJsonSerializer.of(mapper).withProperty("clusterName", CLUSTER_NAME);
      String actual = serializer.serialize(event);
    }
  }

I set up LOOP_COUNT to 100K. The results differ between the runs slightly, but these are the results (from IntelliJ):

Sometimes my code is slower:

I think that 100K is enough for the JIT to warm up. I guess the performance is comparable and it shouldn't be an issue.