APEXMALHAR-2473 Support for global cache meta information in db Cache… #605

oliverwnk · 2017-04-17T18:40:26Z

…Manager

Uses Component interface and newly implemented CacheContext to pass properties to the Stores by calling setup(CacheContext).
APEXMALHAR-2474: FSLoader implements Component to get numInitLinesToCache from CacheManager and use it in initial load.
Add implementation of get() function so data will also be loaded after initial load.
APEXMALHAR-2475: CacheStore implements Component for passing readOnly and numInitLinesToCache.
Added NO_EVICTION expire strategy. This strategy will be set in setup(CacheContext) if readOnly is true.

@PramodSSImmaneni: please review

pramodin · 2017-04-30T17:17:34Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

      while ((line = bin.readLine()) != null) {
+        if ( numInitCachedLines > 0 && linesCount >= numInitCachedLines) {


Can this be added to the while condition

pramodin · 2017-04-30T22:17:46Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

+    FSDataInputStream in = null;
+    BufferedReader bin = null;
+    try {
+      this.connect();


Why open file system for each call?

Also shouldn't fs be already initialized?

Okay, I added a function where it tries to open a stream with the initialized file system. If it happens to fail it will try to initialize file system once again.

pramodin · 2017-04-30T22:22:06Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

+    try {
+      this.connect();
+      in = fs.open(filePath);
+      bin = new BufferedReader(new InputStreamReader(in));


You can also do... try with close

pramodin · 2017-04-30T22:22:53Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

+            return getValue(tuple);
+          }
+        } catch (Exception parseExp) {
+          logger.info("Unable to parse line {}", line);


pramodin · 2017-04-30T22:36:49Z

contrib/src/main/java/com/datatorrent/contrib/enrich/AbstractEnricher.java

  protected transient List<FieldInfo> lookupFieldInfo = new ArrayList<>();
  protected transient List<FieldInfo> includeFieldInfo = new ArrayList<>();

+  private CacheManager cacheManager = new NullValuesCacheManager();


Enricher is initializing cacheManager during activate but doesn't seem to be closing it in deactivate.

pramodin · 2017-04-30T22:38:07Z

library/src/main/java/com/datatorrent/lib/db/cache/CacheManager.java

+  public static class CacheContext implements Context
+  {
+    public static final transient Attribute<Boolean> READ_ONLY_ATTR = new Attribute<>((false));
+    public static final transient Attribute<Integer> NUM_INIT_CACHED_LINES_ATTR = new Attribute<>((-1));


The checks in other places might be straightforward if it is initialized to Integer.MAX_VALUE.

I see, it would slightly change the behavior, e.g. in FSLoader it has no upper bound if num_init_cached_lines_attr was not set. With the change Integer.MAX_VALUE would always be the default limit there

Might be necessary to leave as is to keep functionality of CacheStore (primary store) to not limit maxCacheSize explicitly when numInitCacheLines is not set, but rise maxCacheSize if a greater numInitCacheLines is chosen.

pramodin · 2017-04-30T22:45:06Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

 {
  @NotNull
  private String fileName;

  private transient Path filePath;
  private transient FileSystem fs;
  private transient boolean connected;
+  private transient int numInitCachedLines = -1;
+  private transient CacheManager.CacheContext cacheContext;


Is there a need to save the cache context?

pramodin · 2017-04-30T22:47:18Z

library/src/main/java/com/datatorrent/lib/db/cache/CacheManager.java

+    this.numInitCachedLines = numInitCachedLines;
+  }
+
+  public boolean getReadOnlyAttribute()


How about just readOnlyData instead of Attribute in name

pramodin · 2017-04-30T22:49:46Z

library/src/main/java/com/datatorrent/lib/db/cache/CacheManager.java

+        primary.getClass().getMethod("setup", parameters);
+        attributeMap.put(CacheContext.READ_ONLY_ATTR, readOnlyAttribute);
+        attributeMap.put(CacheContext.NUM_INIT_CACHED_LINES_ATTR, numInitCachedLines);
+        ((Component)primary).setup(new CacheContext(attributeMap));


How about reusing the CacheContext instance between the primary and backup initialization. Looks like same attributeMap is being re-written.

pramodin · 2017-05-14T02:08:29Z

contrib/src/main/java/com/datatorrent/contrib/enrich/AbstractEnricher.java

@@ -185,13 +185,18 @@ public void activate(Context context)
    try {
      cacheManager.initialize();
    } catch (IOException e) {
-      throw new RuntimeException("Unable to initialize primary cache", e);
+      throw new RuntimeException("Unable to initialize cache store", e);


Why would be just cache store initialization failure, the backup could fail as well, isn't it. Shouldn't this be unable to initialize cache manager.

I meant generally a store of the cache, but I see the confusion with 'CacheStore'. I will change it accordingly.

pramodin · 2017-05-14T02:10:52Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

-    try {
+    Map<Object, Object> result;
+    try (
+      FSDataInputStream in = fs.open(filePath);


Where is the input stream closed?

try-with-resources should close it automatically

pramodin · 2017-05-14T02:11:15Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

        try {
          Map<String, Object> tuple = extractFields(line);
          if (tuple != null && !tuple.isEmpty()) {
            result.put(getKey(tuple), getValue(tuple));
          }
        } catch (Exception parseExp) {
-          logger.info("Unable to parse line {}", line);
+          logger.warn("Unable to parse line {}", line);


Pass exception as well

Since this is initialization, you might as well fail with a runtime exception.

pramodin · 2017-05-14T02:12:01Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

@@ -145,9 +136,44 @@ private Object getKey(Map<String, Object> tuple)
  @Override
  public Object get(Object key)
  {
+    try (
+      FSDataInputStream in = openFsInputStream(filePath);


Close input stream when done

pramodin · 2017-05-14T02:12:26Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

+            return getValue(tuple);
+          }
+        } catch (Exception parseExp) {
+          logger.warn("Unable to parse line {}", line);


Fail with runtime exception if unable to parse line.

pramodin · 2017-05-14T02:16:00Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

    return null;
  }

+  private FSDataInputStream openFsInputStream(Path filePath)


Maybe call it check open or something that indicates it is not a mere open but will try to handle failures and retry

How about openWithRetryFsInputStream(..)?

pramodin · 2017-05-14T02:21:59Z

library/src/main/java/com/datatorrent/lib/db/cache/CacheStore.java

    if (entryExpiryStrategy == ExpiryType.EXPIRE_AFTER_ACCESS) {
      cacheBuilder.expireAfterAccess(entryExpiryDurationInMillis, TimeUnit.MILLISECONDS);
    } else if (entryExpiryStrategy == ExpiryType.EXPIRE_AFTER_WRITE) {
      cacheBuilder.expireAfterWrite(entryExpiryDurationInMillis, TimeUnit.MILLISECONDS);
    }
    cache = cacheBuilder.build();
+
+    if (entryExpiryStrategy == ExpiryType.NO_EVICTION) {
+      return;


In disconnect, cleanupSchedule is referenced as well, in disconnect check for not null

pramodin · 2017-05-14T02:25:15Z

library/src/test/java/com/datatorrent/lib/db/cache/CacheManagerTest.java

+    @Override
+    public Map<Object, Object> loadInitialData()
+    {
+      Map<Object, Object> result = new HashedMap();


Why not use the standard HashMap?

pramodin · 2017-05-18T06:06:21Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

@@ -137,7 +137,7 @@ private Object getKey(Map<String, Object> tuple)
  public Object get(Object key)
  {
    try (
-      FSDataInputStream in = openFsInputStream(filePath);
+      FSDataInputStream in = openWithRetryFsInputStream(filePath);


Looks like stream is still not being closed.

Try-with-resources should close it automatically.

pramodin · 2017-05-18T06:06:41Z

squash commits and rebase

pramodin · 2017-05-18T17:54:50Z

contrib/src/main/java/com/datatorrent/contrib/enrich/FSLoader.java

+    } catch (IOException e) {
+      try {
+        this.connect();
+        fs.open(filePath);


in = fs.open(filePath)

…Manager 1. Uses Component interface and newly implemented CacheContext to pass properties to the Stores by calling setup(CacheContext). 2. APEXMALHAR-2474: FSLoader implements Component to get numInitLinesToCache from CacheManager and use it in initial load. Add implementation of get() function so data will also be loaded after initial load. 3. APEXMALHAR-2475: CacheStore implements Component for passing readOnly and numInitLinesToCache. Added NO_EVICTION expire strategy. This strategy will be set in setup(CacheContext) if readOnly is true.

oliverwnk force-pushed the APEXMALHAR-2473.cache-meta-info-support-in-CacheManager branch from 948fc9d to 206b78e Compare April 18, 2017 01:52

pramodin reviewed Apr 30, 2017

View reviewed changes

oliverwnk force-pushed the APEXMALHAR-2473.cache-meta-info-support-in-CacheManager branch 2 times, most recently from 34e9f3d to 5d39b4b Compare May 12, 2017 18:59

pramodin reviewed May 14, 2017

View reviewed changes

pramodin reviewed May 18, 2017

View reviewed changes

oliverwnk force-pushed the APEXMALHAR-2473.cache-meta-info-support-in-CacheManager branch from fff7bac to 36b44d0 Compare May 18, 2017 17:52

pramodin reviewed May 18, 2017

View reviewed changes

oliverwnk force-pushed the APEXMALHAR-2473.cache-meta-info-support-in-CacheManager branch from 36b44d0 to 6be30b7 Compare May 18, 2017 17:58

oliverwnk force-pushed the APEXMALHAR-2473.cache-meta-info-support-in-CacheManager branch from 6be30b7 to 118a75f Compare May 18, 2017 18:01

asfgit merged commit 118a75f into apache:master May 18, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

APEXMALHAR-2473 Support for global cache meta information in db Cache… #605

APEXMALHAR-2473 Support for global cache meta information in db Cache… #605

oliverwnk commented Apr 17, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017

oliverwnk May 8, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017

oliverwnk May 2, 2017

oliverwnk May 8, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017

pramodin Apr 30, 2017 •

edited

pramodin May 14, 2017

oliverwnk May 18, 2017

pramodin May 14, 2017

oliverwnk May 18, 2017

pramodin May 14, 2017

pramodin May 14, 2017

pramodin May 14, 2017

pramodin May 14, 2017

pramodin May 14, 2017

oliverwnk May 18, 2017

pramodin May 14, 2017

pramodin May 14, 2017

pramodin May 18, 2017

oliverwnk May 18, 2017

pramodin commented May 18, 2017

pramodin May 18, 2017

		while ((line = bin.readLine()) != null) {
		if ( numInitCachedLines > 0 && linesCount >= numInitCachedLines) {

APEXMALHAR-2473 Support for global cache meta information in db Cache… #605

APEXMALHAR-2473 Support for global cache meta information in db Cache… #605

Conversation

oliverwnk commented Apr 17, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pramodin Apr 30, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pramodin commented May 18, 2017

Choose a reason for hiding this comment

pramodin Apr 30, 2017 •

edited