{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/camus/pulls/comments/25224319","id":25224319,"diff_hunk":"@@ -1,87 +1,156 @@\n package com.linkedin.camus.schemaregistry;\n \n+import java.util.HashMap;\n+import java.util.Map;\n import java.util.Properties;\n import java.util.concurrent.ConcurrentHashMap;\n+import java.util.concurrent.ConcurrentMap;\n+import java.util.concurrent.TimeUnit;\n+import java.util.concurrent.atomic.AtomicInteger;\n+\n \n public class CachedSchemaRegistry<S> implements SchemaRegistry<S> {\n-\tprivate final SchemaRegistry<S> registry;\n-\tprivate final ConcurrentHashMap<CachedSchemaTuple, S> cachedById;\n-\tprivate final ConcurrentHashMap<String, SchemaDetails<S>> cachedLatest;\n-\t\n-\tpublic void init(Properties props) {}\n-\n-\tpublic CachedSchemaRegistry(SchemaRegistry<S> registry) {\n-\t\tthis.registry = registry;\n-\t\tthis.cachedById = new ConcurrentHashMap<CachedSchemaTuple, S>();\n-\t\tthis.cachedLatest = new ConcurrentHashMap<String, SchemaDetails<S>>();\n-\t}\n-\n-\tpublic String register(String topic, S schema) {\n-\t\treturn registry.register(topic, schema);\n-\t}\n-\n-\tpublic S getSchemaByID(String topic, String id) {\n-\t\tCachedSchemaTuple cacheKey = new CachedSchemaTuple(topic, id);\n-\t\tS schema = cachedById.get(cacheKey);\n-\t\tif (schema == null) {\n-\t\t\tschema = registry.getSchemaByID(topic, id);\n-\t\t\tcachedById.putIfAbsent(cacheKey, schema);\n-\t\t}\n-\t\treturn schema;\n-\t}\n-\n-\tpublic SchemaDetails<S> getLatestSchemaByTopic(String topicName) {\n-\t\tSchemaDetails<S> schema = cachedLatest.get(topicName);\n-\t\tif (schema == null) {\n-\t\t\tschema = registry.getLatestSchemaByTopic(topicName);\n-\t\t\tcachedLatest.putIfAbsent(topicName, schema);\n-\t\t}\n-\t\treturn schema;\n-\t}\n-\n-\tpublic static class CachedSchemaTuple {\n-\t\tprivate final String topic;\n-\t\tprivate final String id;\n-\n-\t\tpublic CachedSchemaTuple(String topic, String id) {\n-\t\t\tthis.topic = topic;\n-\t\t\tthis.id = id;\n-\t\t}\n-\n-\t\t@Override\n-\t\tpublic int hashCode() {\n-\t\t\tfinal int prime = 31;\n-\t\t\tint result = 1;\n-\t\t\tresult = prime * result + ((id == null) ? 0 : id.hashCode());\n-\t\t\tresult = prime * result + ((topic == null) ? 0 : topic.hashCode());\n-\t\t\treturn result;\n-\t\t}\n-\n-\t\t@Override\n-\t\tpublic boolean equals(Object obj) {\n-\t\t\tif (this == obj)\n-\t\t\t\treturn true;\n-\t\t\tif (obj == null)\n-\t\t\t\treturn false;\n-\t\t\tif (getClass() != obj.getClass())\n-\t\t\t\treturn false;\n-\t\t\tCachedSchemaTuple other = (CachedSchemaTuple) obj;\n-\t\t\tif (id == null) {\n-\t\t\t\tif (other.id != null)\n-\t\t\t\t\treturn false;\n-\t\t\t} else if (!id.equals(other.id))\n-\t\t\t\treturn false;\n-\t\t\tif (topic == null) {\n-\t\t\t\tif (other.topic != null)\n-\t\t\t\t\treturn false;\n-\t\t\t} else if (!topic.equals(other.topic))\n-\t\t\t\treturn false;\n-\t\t\treturn true;\n-\t\t}\n-\n-\t\t@Override\n-\t\tpublic String toString() {\n-\t\t\treturn \"CachedSchemaTuple [topic=\" + topic + \", id=\" + id + \"]\";\n-\t\t}\n-\t}\n+  private static final String GET_SCHEMA_BY_ID_MAX_RETIRES = \"get.schema.by.id.max.retries\";\n+  private static final String GET_SCHEMA_BY_ID_MAX_RETIRES_DEFAULT = \"3\";\n+  private static final String GET_SCHEMA_BY_ID_MIN_INTERVAL_SECONDS = \"get.schema.by.id.min.interval.seconds\";\n+  private static final String GET_SCHEMA_BY_ID_MIN_INTERVAL_SECONDS_DEFAULT = \"1\";\n+\n+  private final SchemaRegistry<S> registry;\n+  private final ConcurrentHashMap<CachedSchemaTuple, S> cachedById;\n+  private final ConcurrentHashMap<String, SchemaDetails<S>> cachedLatest;\n+  private int getByIdMaxRetries;\n+  private int getByIdMinIntervalSeconds;\n+  private Map<String, FailedFetchHistory> failedFetchHistories;\n+\n+  private class FailedFetchHistory {\n+    private int numOfAttempts;\n+    private long previousAttemptTime;\n+\n+    private FailedFetchHistory(int numOfnumOfAttempts, long previousAttemptTime) {\n+      this.numOfAttempts = numOfnumOfAttempts;\n+      this.previousAttemptTime = previousAttemptTime;\n+    }\n+  }\n+\n+  public void init(Properties props) {\n+    this.getByIdMaxRetries =\n+        Integer.parseInt(props.getProperty(GET_SCHEMA_BY_ID_MAX_RETIRES, GET_SCHEMA_BY_ID_MAX_RETIRES_DEFAULT));\n+    this.getByIdMinIntervalSeconds =\n+        Integer.parseInt(props.getProperty(GET_SCHEMA_BY_ID_MIN_INTERVAL_SECONDS,\n+            GET_SCHEMA_BY_ID_MIN_INTERVAL_SECONDS_DEFAULT));\n+  }\n+\n+  public CachedSchemaRegistry(SchemaRegistry<S> registry, Properties props) {\n+    this.registry = registry;\n+    this.cachedById = new ConcurrentHashMap<CachedSchemaTuple, S>();\n+    this.cachedLatest = new ConcurrentHashMap<String, SchemaDetails<S>>();\n+    this.failedFetchHistories = new HashMap<String, FailedFetchHistory>();\n+    this.init(props);\n+  }\n+\n+  public String register(String topic, S schema) {\n+    return registry.register(topic, schema);\n+  }\n+\n+  public S getSchemaByID(String topic, String id) {\n+    CachedSchemaTuple cacheKey = new CachedSchemaTuple(topic, id);\n+    S schema = cachedById.get(cacheKey);\n+    if (schema == null) {\n+      synchronized (this) {\n+        if (shouldFetchFromSchemaRegistry(id)) {\n+          schema = fetchFromSchemaRegistry(topic, id);\n+        } else {\n+          throw new SchemaNotFoundException();\n+        }\n+      }\n+    }\n+    return schema;\n+  }\n+\n+  private boolean shouldFetchFromSchemaRegistry(String id) {","path":"camus-api/src/main/java/com/linkedin/camus/schemaregistry/CachedSchemaRegistry.java","position":150,"original_position":150,"commit_id":"86d5d39eab61173ed649b98a9a7f5e3bb415381c","original_commit_id":"86d5d39eab61173ed649b98a9a7f5e3bb415381c","user":{"login":"HenryCaiHaiying","id":7378943,"avatar_url":"https://avatars.githubusercontent.com/u/7378943?v=3","gravatar_id":"","url":"https://api.github.com/users/HenryCaiHaiying","html_url":"https://github.com/HenryCaiHaiying","followers_url":"https://api.github.com/users/HenryCaiHaiying/followers","following_url":"https://api.github.com/users/HenryCaiHaiying/following{/other_user}","gists_url":"https://api.github.com/users/HenryCaiHaiying/gists{/gist_id}","starred_url":"https://api.github.com/users/HenryCaiHaiying/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HenryCaiHaiying/subscriptions","organizations_url":"https://api.github.com/users/HenryCaiHaiying/orgs","repos_url":"https://api.github.com/users/HenryCaiHaiying/repos","events_url":"https://api.github.com/users/HenryCaiHaiying/events{/privacy}","received_events_url":"https://api.github.com/users/HenryCaiHaiying/received_events","type":"User","site_admin":false},"body":"Do you want to synchronize on this method?","created_at":"2015-02-24T02:35:43Z","updated_at":"2015-02-24T02:35:43Z","html_url":"https://github.com/linkedin/camus/pull/172#discussion_r25224319","pull_request_url":"https://api.github.com/repos/linkedin/camus/pulls/172","_links":{"self":{"href":"https://api.github.com/repos/linkedin/camus/pulls/comments/25224319"},"html":{"href":"https://github.com/linkedin/camus/pull/172#discussion_r25224319"},"pull_request":{"href":"https://api.github.com/repos/linkedin/camus/pulls/172"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/camus/pulls/172","id":29491041,"html_url":"https://github.com/linkedin/camus/pull/172","diff_url":"https://github.com/linkedin/camus/pull/172.diff","patch_url":"https://github.com/linkedin/camus/pull/172.patch","issue_url":"https://api.github.com/repos/linkedin/camus/issues/172","number":172,"state":"open","locked":false,"title":"etl-2572: ignoring kafka events with missing schema id","user":{"login":"zliu41","id":6342538,"avatar_url":"https://avatars.githubusercontent.com/u/6342538?v=3","gravatar_id":"","url":"https://api.github.com/users/zliu41","html_url":"https://github.com/zliu41","followers_url":"https://api.github.com/users/zliu41/followers","following_url":"https://api.github.com/users/zliu41/following{/other_user}","gists_url":"https://api.github.com/users/zliu41/gists{/gist_id}","starred_url":"https://api.github.com/users/zliu41/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zliu41/subscriptions","organizations_url":"https://api.github.com/users/zliu41/orgs","repos_url":"https://api.github.com/users/zliu41/repos","events_url":"https://api.github.com/users/zliu41/events{/privacy}","received_events_url":"https://api.github.com/users/zliu41/received_events","type":"User","site_admin":false},"body":"1. Added the ability to limit the number of retries for a missing schema id\r\n2. Added the ability to limit the min interval between two retries for a missing schema id.\r\n3. Added hadoop counters 'MSG_SKIPPED.SCHEMA_NOT_FOUND' and 'MSG_SKIPPED.OTHER'.\r\n4. Added unit test for CachedSchemaRegistry (CachedSchemaRegistryTest.java)","created_at":"2015-02-18T00:19:45Z","updated_at":"2015-02-24T02:35:43Z","closed_at":null,"merged_at":null,"merge_commit_sha":"042c53d0605c3fdf65317f365ac25b811c0a37a7","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/camus/pulls/172/commits","review_comments_url":"https://api.github.com/repos/linkedin/camus/pulls/172/comments","review_comment_url":"https://api.github.com/repos/linkedin/camus/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/camus/issues/172/comments","statuses_url":"https://api.github.com/repos/linkedin/camus/statuses/86d5d39eab61173ed649b98a9a7f5e3bb415381c","head":{"label":"zliu41:master","ref":"master","sha":"86d5d39eab61173ed649b98a9a7f5e3bb415381c","user":{"login":"zliu41","id":6342538,"avatar_url":"https://avatars.githubusercontent.com/u/6342538?v=3","gravatar_id":"","url":"https://api.github.com/users/zliu41","html_url":"https://github.com/zliu41","followers_url":"https://api.github.com/users/zliu41/followers","following_url":"https://api.github.com/users/zliu41/following{/other_user}","gists_url":"https://api.github.com/users/zliu41/gists{/gist_id}","starred_url":"https://api.github.com/users/zliu41/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zliu41/subscriptions","organizations_url":"https://api.github.com/users/zliu41/orgs","repos_url":"https://api.github.com/users/zliu41/repos","events_url":"https://api.github.com/users/zliu41/events{/privacy}","received_events_url":"https://api.github.com/users/zliu41/received_events","type":"User","site_admin":false},"repo":{"id":30930337,"name":"camus","full_name":"zliu41/camus","owner":{"login":"zliu41","id":6342538,"avatar_url":"https://avatars.githubusercontent.com/u/6342538?v=3","gravatar_id":"","url":"https://api.github.com/users/zliu41","html_url":"https://github.com/zliu41","followers_url":"https://api.github.com/users/zliu41/followers","following_url":"https://api.github.com/users/zliu41/following{/other_user}","gists_url":"https://api.github.com/users/zliu41/gists{/gist_id}","starred_url":"https://api.github.com/users/zliu41/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zliu41/subscriptions","organizations_url":"https://api.github.com/users/zliu41/orgs","repos_url":"https://api.github.com/users/zliu41/repos","events_url":"https://api.github.com/users/zliu41/events{/privacy}","received_events_url":"https://api.github.com/users/zliu41/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/zliu41/camus","description":"LinkedIn's Kafka to HDFS pipeline.","fork":true,"url":"https://api.github.com/repos/zliu41/camus","forks_url":"https://api.github.com/repos/zliu41/camus/forks","keys_url":"https://api.github.com/repos/zliu41/camus/keys{/key_id}","collaborators_url":"https://api.github.com/repos/zliu41/camus/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/zliu41/camus/teams","hooks_url":"https://api.github.com/repos/zliu41/camus/hooks","issue_events_url":"https://api.github.com/repos/zliu41/camus/issues/events{/number}","events_url":"https://api.github.com/repos/zliu41/camus/events","assignees_url":"https://api.github.com/repos/zliu41/camus/assignees{/user}","branches_url":"https://api.github.com/repos/zliu41/camus/branches{/branch}","tags_url":"https://api.github.com/repos/zliu41/camus/tags","blobs_url":"https://api.github.com/repos/zliu41/camus/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/zliu41/camus/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/zliu41/camus/git/refs{/sha}","trees_url":"https://api.github.com/repos/zliu41/camus/git/trees{/sha}","statuses_url":"https://api.github.com/repos/zliu41/camus/statuses/{sha}","languages_url":"https://api.github.com/repos/zliu41/camus/languages","stargazers_url":"https://api.github.com/repos/zliu41/camus/stargazers","contributors_url":"https://api.github.com/repos/zliu41/camus/contributors","subscribers_url":"https://api.github.com/repos/zliu41/camus/subscribers","subscription_url":"https://api.github.com/repos/zliu41/camus/subscription","commits_url":"https://api.github.com/repos/zliu41/camus/commits{/sha}","git_commits_url":"https://api.github.com/repos/zliu41/camus/git/commits{/sha}","comments_url":"https://api.github.com/repos/zliu41/camus/comments{/number}","issue_comment_url":"https://api.github.com/repos/zliu41/camus/issues/comments{/number}","contents_url":"https://api.github.com/repos/zliu41/camus/contents/{+path}","compare_url":"https://api.github.com/repos/zliu41/camus/compare/{base}...{head}","merges_url":"https://api.github.com/repos/zliu41/camus/merges","archive_url":"https://api.github.com/repos/zliu41/camus/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/zliu41/camus/downloads","issues_url":"https://api.github.com/repos/zliu41/camus/issues{/number}","pulls_url":"https://api.github.com/repos/zliu41/camus/pulls{/number}","milestones_url":"https://api.github.com/repos/zliu41/camus/milestones{/number}","notifications_url":"https://api.github.com/repos/zliu41/camus/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/zliu41/camus/labels{/name}","releases_url":"https://api.github.com/repos/zliu41/camus/releases{/id}","created_at":"2015-02-17T18:14:19Z","updated_at":"2015-02-23T23:14:15Z","pushed_at":"2015-02-23T23:14:14Z","git_url":"git://github.com/zliu41/camus.git","ssh_url":"git@github.com:zliu41/camus.git","clone_url":"https://github.com/zliu41/camus.git","svn_url":"https://github.com/zliu41/camus","homepage":"","size":41650,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"linkedin:master","ref":"master","sha":"017b61c04ff45efde02daec34503fb55993117b3","user":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"repo":{"id":7266306,"name":"camus","full_name":"linkedin/camus","owner":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/linkedin/camus","description":"LinkedIn's Kafka to HDFS pipeline.","fork":false,"url":"https://api.github.com/repos/linkedin/camus","forks_url":"https://api.github.com/repos/linkedin/camus/forks","keys_url":"https://api.github.com/repos/linkedin/camus/keys{/key_id}","collaborators_url":"https://api.github.com/repos/linkedin/camus/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/linkedin/camus/teams","hooks_url":"https://api.github.com/repos/linkedin/camus/hooks","issue_events_url":"https://api.github.com/repos/linkedin/camus/issues/events{/number}","events_url":"https://api.github.com/repos/linkedin/camus/events","assignees_url":"https://api.github.com/repos/linkedin/camus/assignees{/user}","branches_url":"https://api.github.com/repos/linkedin/camus/branches{/branch}","tags_url":"https://api.github.com/repos/linkedin/camus/tags","blobs_url":"https://api.github.com/repos/linkedin/camus/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/linkedin/camus/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/linkedin/camus/git/refs{/sha}","trees_url":"https://api.github.com/repos/linkedin/camus/git/trees{/sha}","statuses_url":"https://api.github.com/repos/linkedin/camus/statuses/{sha}","languages_url":"https://api.github.com/repos/linkedin/camus/languages","stargazers_url":"https://api.github.com/repos/linkedin/camus/stargazers","contributors_url":"https://api.github.com/repos/linkedin/camus/contributors","subscribers_url":"https://api.github.com/repos/linkedin/camus/subscribers","subscription_url":"https://api.github.com/repos/linkedin/camus/subscription","commits_url":"https://api.github.com/repos/linkedin/camus/commits{/sha}","git_commits_url":"https://api.github.com/repos/linkedin/camus/git/commits{/sha}","comments_url":"https://api.github.com/repos/linkedin/camus/comments{/number}","issue_comment_url":"https://api.github.com/repos/linkedin/camus/issues/comments{/number}","contents_url":"https://api.github.com/repos/linkedin/camus/contents/{+path}","compare_url":"https://api.github.com/repos/linkedin/camus/compare/{base}...{head}","merges_url":"https://api.github.com/repos/linkedin/camus/merges","archive_url":"https://api.github.com/repos/linkedin/camus/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/linkedin/camus/downloads","issues_url":"https://api.github.com/repos/linkedin/camus/issues{/number}","pulls_url":"https://api.github.com/repos/linkedin/camus/pulls{/number}","milestones_url":"https://api.github.com/repos/linkedin/camus/milestones{/number}","notifications_url":"https://api.github.com/repos/linkedin/camus/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/linkedin/camus/labels{/name}","releases_url":"https://api.github.com/repos/linkedin/camus/releases{/id}","created_at":"2012-12-20T23:54:42Z","updated_at":"2015-02-22T20:14:42Z","pushed_at":"2015-02-21T18:27:21Z","git_url":"git://github.com/linkedin/camus.git","ssh_url":"git@github.com:linkedin/camus.git","clone_url":"https://github.com/linkedin/camus.git","svn_url":"https://github.com/linkedin/camus","homepage":"","size":43342,"stargazers_count":423,"watchers_count":423,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":261,"mirror_url":null,"open_issues_count":42,"forks":261,"open_issues":42,"watchers":423,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/camus/pulls/172"},"html":{"href":"https://github.com/linkedin/camus/pull/172"},"issue":{"href":"https://api.github.com/repos/linkedin/camus/issues/172"},"comments":{"href":"https://api.github.com/repos/linkedin/camus/issues/172/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/camus/pulls/172/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/camus/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/camus/pulls/172/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/camus/statuses/86d5d39eab61173ed649b98a9a7f5e3bb415381c"}}}}