Github user nickwallen commented on a diff in the pull request:
    --- Diff: 
    @@ -89,29 +91,25 @@ public void prepare(Map map, TopologyContext 
topologyContext, OutputCollector ou
         if (this.maxTimeRetain == null) {
           throw new IllegalStateException("maxTimeRetain must be specified");
    -    loader = new CacheLoader<String, Map<String, Tuple>>() {
    -      @Override
    -      public Map<String, Tuple> load(String key) throws Exception {
    -        return new HashMap<>();
    -      }
    -    };
    -    cache = CacheBuilder.newBuilder().maximumSize(maxCacheSize)
    -            .expireAfterWrite(maxTimeRetain, 
TimeUnit.MINUTES).removalListener(new JoinRemoveListener())
    -            .build(loader);
    +    loader = s -> new HashMap<>();
    +    cache = Caffeine.newBuilder().maximumSize(maxCacheSize)
    +                         .expireAfterWrite(maxTimeRetain, TimeUnit.MINUTES)
    +                         .removalListener(new JoinRemoveListener())
    --- End diff --
    Yes, it is pre-existing.  We can address at a later time.
    I remember now, maxing out this cache causes the Split/Join to fail, which 
is a major problem for the Split/Join topology.  And this cache here is only 
for the Split/Join, not the Unified topology.
    We should probably look at adding similar logging (only when ERROR enabled) 
for the other places where we use the cache.  Or just some mechanism to 
periodically log cache stats.  Anywho, down the road.


Reply via email to