vmohanan1

public class MasterOrgBucketAssigner implements BucketAssigner<MeteringEventsWrapper, String> {
    @Override public String getBucketId(final MeteringEventsWrapper element, final Context context) {
        final String partition = getPartitionFromProcessingTime(context);
        final String masterOrg = (String) element.getEvents().stream().findFirst()
                .map(event -> event.get(Constants.EVENT_MASTER_ORG_KEY)).get();
        final String schema = element.getMetadata().get(Constants.EVENT_SCHEMA_ID_KEY);
        return schema + "/" + masterOrg + "/" + partition;
    }

    private String getPartitionFromProcessingTime(final Context context) {

vmohanan1

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.serialization.Encoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;

public class JsonEncoder<T> implements Encoder<T> {

vmohanan1

public class DeduplicatingFlatmap extends RichFlatMapFunction<EventsWrapper, EventsWrapper> {

    private LoadingCache<String, Boolean> dedupeCache;
    private final long cacheExpirationTimeMs;
    private final int cacheSize;

    public DeduplicatingFlatmap(final long cacheExpirationTimeMs, final int cacheSize) {
        this.cacheExpirationTimeMs = cacheExpirationTimeMs;
        this.cacheSize = cacheSize;
    }

vmohanan1

### Keybase proof

I hereby claim:

  * I am vmohanan1 on github.
  * I am vmohanan (https://keybase.io/vmohanan) on keybase.
  * I have a public key ASCuGKT-ljVCVyFf0ZvVtL11RUv2l0HER0cjnZ46__iV2wo

To claim this, I am signing this object:

vmohanan1

package com.mulesoft.mdas.ingestion;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.SneakyThrows;

vmohanan1

package events.piepline;

import events.model.Event;
import events.model.EventsWrapper;

import java.util.ArrayList;
import java.util.List;

public class Driver {

vmohanan1

package events.piepline;

import events.model.EventsWrapper;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;

import java.util.List;

public class EventsPipeline {

vmohanan1

package events.piepline;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import events.model.Event;
import events.model.EventsWrapper;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;

vmohanan1

package events.model;

import java.io.Serializable;

// Events class
public class Event implements Serializable {

  private final String eventId;
  private final String name;