@Override
    public Map<String, Object> toElasticSearchObject(ObjectMapper objectMapper, @Nonnull final Meter invalidTimestampMeter) {
        final Map<String, Object> obj = Maps.newHashMapWithExpectedSize(REQUIRED_FIELDS.size() + fields.size());

        for (Map.Entry<String, Object> entry : fields.entrySet()) {
            final String key = entry.getKey();
            if (key.equals(FIELD_ID)) {
                continue;
            }

            final Object value = entry.getValue();
            // Elasticsearch does not allow "." characters in keys since version 2.0.
            // See: https://www.elastic.co/guide/en/elasticsearch/reference/2.0/breaking_20_mapping_changes.html#_field_names_may_not_contain_dots
            if (key.contains(".")) {
                final String newKey = key.replace('.', KEY_REPLACEMENT_CHAR);

                // If the message already contains the transformed key, we skip the field and emit a warning.
                // This is still not optimal but better than implementing expensive logic with multiple replacement
                // character options. Conflicts should be rare...
                if (!obj.containsKey(newKey)) {
                    obj.put(newKey, value);
                } else {
                    LOG.warn("Keys must not contain a \".\" character! Ignoring field \"{}\"=\"{}\" in message [{}] - Unable to replace \".\" with a \"{}\" because of key conflict: \"{}\"=\"{}\"",
                            key, value, getId(), KEY_REPLACEMENT_CHAR, newKey, obj.get(newKey));
                    LOG.debug("Full message with \".\" in message key: {}", this);
                }
            } else {
                if (obj.containsKey(key)) {
                    final String newKey = key.replace(KEY_REPLACEMENT_CHAR, '.');
                    // Deliberate warning duplicates because the key with the "." might be transformed before reaching
                    // the duplicate original key with a "_". Otherwise we would silently overwrite the transformed key.
                    LOG.warn("Keys must not contain a \".\" character! Ignoring field \"{}\"=\"{}\" in message [{}] - Unable to replace \".\" with a \"{}\" because of key conflict: \"{}\"=\"{}\"",
                            newKey, fields.get(newKey), getId(), KEY_REPLACEMENT_CHAR, key, value);
                    LOG.debug("Full message with \".\" in message key: {}", this);
                }
                obj.put(key, value);
            }
        }

        obj.put(FIELD_MESSAGE, getMessage());
        obj.put(FIELD_SOURCE, getSource());
        obj.put(FIELD_STREAMS, getStreamIds());
        obj.put(FIELD_GL2_ACCOUNTED_MESSAGE_SIZE, getSize());

        final Object timestampValue = getField(FIELD_TIMESTAMP);
        DateTime dateTime = timestampValue == null ? fallbackForNullTimestamp() : convertToDateTime(timestampValue);
        obj.put(FIELD_TIMESTAMP, buildElasticSearchTimeFormat(dateTime.withZone(UTC)));

        if (processingErrors != null && !processingErrors.isEmpty()) {
            if (processingErrors.stream().anyMatch(processingError -> processingError.getCause().equals(ProcessingFailureCause.InvalidTimestampException))) {
                invalidTimestampMeter.mark();
            }
            obj.put(FIELD_GL2_PROCESSING_ERROR,
                    processingErrors.stream()
                            .map(pe -> pe.getMessage() + " - " + pe.getDetails())
                            .collect(Collectors.joining(", ")));
        }

        return obj;
    }
@Test
    public void testToElasticSearchObject() throws Exception {
        message.addField("field1", "wat");
        message.addField("field2", "that");
        message.addField(Message.FIELD_STREAMS, Collections.singletonList("test-stream"));

        final Map<String, Object> object = message.toElasticSearchObject(objectMapper, invalidTimestampMeter);

        assertEquals("foo", object.get("message"));
        assertEquals("bar", object.get("source"));
        assertEquals("wat", object.get("field1"));
        assertEquals("that", object.get("field2"));
        assertEquals(Tools.buildElasticSearchTimeFormat((DateTime) message.getField("timestamp")), object.get("timestamp"));

        @SuppressWarnings("unchecked")
        final Collection<String> streams = (Collection<String>) object.get("streams");
        assertThat(streams).containsOnly("test-stream");
    }