Kafka topicをStormでの処理 LineageをApache Atlasで表示

## Log

<details><summary>実行ログ</summary><div>

```bash
[centos@zzeng-hdp-1 ~/sandbox/crosscomponent_demo/crosscomponent_scripts/storm-demo]$ ./004-run-storm-job.sh
Running: /usr/jdk64/jdk1.8.0_112/bin/java -server -Ddaemon.name= -Dstorm.options= -Dstorm.home=/usr/hdp/3.1.0.0-78/storm -Dstorm.log.dir=/var/log/storm -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/hdp/3.1.0.0-78/storm/*:/usr/hdp/3.1.0.0-78/storm/lib/*:/usr/hdp/3.1.0.0-78/storm/extlib/* org.apache.storm.daemon.ClientJarTransformerRunner org.apache.storm.hack.StormShadeTransformer /home/centos/sandbox/crosscomponent_demo/crosscomponent_scripts/storm-demo/lib/storm-samples-1.0-jar-with-dependencies.jar /tmp/bdd4785432d311e9b3aefa163e67d972.jar
998  [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/state/StateFactory to org/apache/storm/trident/state/StateFactory in storm/kafka/trident/TridentKafkaStateFactory.class. please modify your code to use the new namespace
1004 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/task/IMetricsContext to org/apache/storm/task/IMetricsContext in storm/kafka/trident/TridentKafkaStateFactory.class. please modify your code to use the new namespace
1005 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/state/State to org/apache/storm/trident/state/State in storm/kafka/trident/TridentKafkaStateFactory.class. please modify your code to use the new namespace
1017 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/IOpaquePartitionedTridentSpout$Emitter to org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout$Emitter in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1018 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/IOpaquePartitionedTridentSpout to org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1018 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/IPartitionedTridentSpout$Emitter to org/apache/storm/trident/spout/IPartitionedTridentSpout$Emitter in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1018 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/IPartitionedTridentSpout to org/apache/storm/trident/spout/IPartitionedTridentSpout in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1018 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/ReducedMetric to org/apache/storm/metric/api/ReducedMetric in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1019 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/CombinedMetric to org/apache/storm/metric/api/CombinedMetric in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1019 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/task/TopologyContext to org/apache/storm/task/TopologyContext in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1020 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/IMetric to org/apache/storm/metric/api/IMetric in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1021 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/MeanReducer to org/apache/storm/metric/api/MeanReducer in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1022 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/IReducer to org/apache/storm/metric/api/IReducer in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1022 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/ICombiner to org/apache/storm/metric/api/ICombiner in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1023 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/topology/TransactionAttempt to org/apache/storm/trident/topology/TransactionAttempt in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1023 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/operation/TridentCollector to org/apache/storm/trident/operation/TridentCollector in storm/kafka/trident/TridentKafkaEmitter.class. please modify your code to use the new namespace
1042 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/IOpaquePartitionedTridentSpout$Coordinator to org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout$Coordinator in storm/kafka/trident/OpaqueTridentKafkaSpout.class. please modify your code to use the new namespace
1043 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/tuple/Fields to org/apache/storm/tuple/Fields in storm/kafka/trident/OpaqueTridentKafkaSpout.class. please modify your code to use the new namespace
1044 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/spout/MultiScheme to org/apache/storm/spout/MultiScheme in storm/kafka/trident/OpaqueTridentKafkaSpout.class. please modify your code to use the new namespace
1045 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/state/BaseStateUpdater to org/apache/storm/trident/state/BaseStateUpdater in storm/kafka/trident/TridentKafkaUpdater.class. please modify your code to use the new namespace
1045 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/tuple/TridentTuple to org/apache/storm/trident/tuple/TridentTuple in storm/kafka/trident/TridentKafkaUpdater.class. please modify your code to use the new namespace
1050 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/ISpoutPartition to org/apache/storm/trident/spout/ISpoutPartition in storm/kafka/trident/TridentKafkaEmitter$1.class. please modify your code to use the new namespace
1054 [main] WARN  o.a.s.h.DefaultShader - Relocating storm/trident/spout/IPartitionedTridentSpout$Coordinator to org/apache/storm/trident/spout/IPartitionedTridentSpout$Coordinator in storm/kafka/trident/TransactionalTridentKafkaSpout.class. please modify your code to use the new namespace
1062 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/task/OutputCollector to org/apache/storm/task/OutputCollector in storm/kafka/trident/TridentKafkaState.class. please modify your code to use the new namespace
1064 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/FailedException to org/apache/storm/topology/FailedException in storm/kafka/trident/TridentKafkaState.class. please modify your code to use the new namespace
1072 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/tuple/Values to org/apache/storm/tuple/Values in storm/kafka/StringKeyValueScheme.class. please modify your code to use the new namespace
1074 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/utils/Utils to org/apache/storm/utils/Utils in storm/kafka/ZkState.class. please modify your code to use the new namespace
1079 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/spout/SpoutOutputCollector to org/apache/storm/spout/SpoutOutputCollector in storm/kafka/KafkaSpout$2.class. please modify your code to use the new namespace
1082 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/tuple/Tuple to org/apache/storm/tuple/Tuple in storm/kafka/bolt/mapper/FieldNameBasedTupleToKafkaMapper.class. please modify your code to use the new namespace
1084 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/base/BaseRichBolt to org/apache/storm/topology/base/BaseRichBolt in storm/kafka/bolt/KafkaBolt.class. please modify your code to use the new namespace
1085 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/utils/TupleUtils to org/apache/storm/utils/TupleUtils in storm/kafka/bolt/KafkaBolt.class. please modify your code to use the new namespace
1086 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/OutputFieldsDeclarer to org/apache/storm/topology/OutputFieldsDeclarer in storm/kafka/bolt/KafkaBolt.class. please modify your code to use the new namespace
1092 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/spout/Scheme to org/apache/storm/spout/Scheme in storm/kafka/KeyValueScheme.class. please modify your code to use the new namespace
1120 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/base/BaseRichSpout to org/apache/storm/topology/base/BaseRichSpout in storm/kafka/KafkaSpout.class. please modify your code to use the new namespace
1127 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/spout/RawMultiScheme to org/apache/storm/spout/RawMultiScheme in storm/kafka/KafkaConfig.class. please modify your code to use the new namespace
1129 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/spout/SchemeAsMultiScheme to org/apache/storm/spout/SchemeAsMultiScheme in storm/kafka/KeyValueSchemeAsMultiScheme.class. please modify your code to use the new namespace
1137 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/metric/api/CountMetric to org/apache/storm/metric/api/CountMetric in storm/kafka/PartitionManager.class. please modify your code to use the new namespace
5151 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/security/auth/IAutoCredentials to org/apache/storm/security/auth/IAutoCredentials in org/apache/storm/hdfs/common/security/AutoHDFS.class. please modify your code to use the new namespace
5152 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/security/auth/ICredentialsRenewer to org/apache/storm/security/auth/ICredentialsRenewer in org/apache/storm/hdfs/common/security/AutoHDFS.class. please modify your code to use the new namespace
5152 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/security/INimbusCredentialPlugin to org/apache/storm/security/INimbusCredentialPlugin in org/apache/storm/hdfs/common/security/AutoHDFS.class. please modify your code to use the new namespace
5154 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/security/auth/kerberos/AutoTGT to org/apache/storm/security/auth/kerberos/AutoTGT in org/apache/storm/hdfs/common/security/HdfsSecurityUtil.class. please modify your code to use the new namespace
5170 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/Config to org/apache/storm/Config in org/apache/storm/hdfs/bolt/AbstractHdfsBolt.class. please modify your code to use the new namespace
5184 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/codedistributor/ICodeDistributor to org/apache/storm/codedistributor/ICodeDistributor in org/apache/storm/hdfs/ha/codedistributor/HDFSCodeDistributor.class. please modify your code to use the new namespace
60125 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/base/BaseBasicBolt to org/apache/storm/topology/base/BaseBasicBolt in com/dsinpractice/storm/samples/WordCountTopology$JavaSplitSentence.class. please modify your code to use the new namespace
60126 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/BasicOutputCollector to org/apache/storm/topology/BasicOutputCollector in com/dsinpractice/storm/samples/WordCountTopology$JavaSplitSentence.class. please modify your code to use the new namespace
60127 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/TopologyBuilder to org/apache/storm/topology/TopologyBuilder in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60128 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/generated/StormTopology to org/apache/storm/generated/StormTopology in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60128 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/StormSubmitter to org/apache/storm/StormSubmitter in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60128 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/IRichSpout to org/apache/storm/topology/IRichSpout in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60128 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/SpoutDeclarer to org/apache/storm/topology/SpoutDeclarer in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60129 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/IBasicBolt to org/apache/storm/topology/IBasicBolt in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60129 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/BoltDeclarer to org/apache/storm/topology/BoltDeclarer in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60129 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/InputDeclarer to org/apache/storm/topology/InputDeclarer in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60129 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/topology/IRichBolt to org/apache/storm/topology/IRichBolt in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60130 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/generated/AlreadyAliveException to org/apache/storm/generated/AlreadyAliveException in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60130 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/generated/InvalidTopologyException to org/apache/storm/generated/InvalidTopologyException in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60130 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/Testing to org/apache/storm/Testing in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
60130 [main] WARN  o.a.s.h.DefaultShader - Relocating backtype/storm/ILocalCluster to org/apache/storm/ILocalCluster in com/dsinpractice/storm/samples/WordCountTopology.class. please modify your code to use the new namespace
Running: /usr/jdk64/jdk1.8.0_112/bin/java -Ddaemon.name= -Dstorm.options= -Dstorm.home=/usr/hdp/3.1.0.0-78/storm -Dstorm.log.dir=/var/log/storm -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/hdp/3.1.0.0-78/storm/*:/usr/hdp/3.1.0.0-78/storm/lib/*:/usr/hdp/3.1.0.0-78/storm/extlib/*:/tmp/bdd4785432d311e9b3aefa163e67d972.jar:/usr/hdp/current/storm-client/conf:/usr/hdp/3.1.0.0-78/storm/bin -Dstorm.jar=/tmp/bdd4785432d311e9b3aefa163e67d972.jar -Dstorm.dependency.jars= -Dstorm.dependency.artifacts={} com.dsinpractice.storm.samples.WordCountTopology --cluster true --name storm-demo-topology-01 --path /user/storm/storm-hdfs-test-01 --topic my-topic-01
1418 [main] INFO  o.a.s.h.b.AbstractHdfsBolt - Enabling tick tuple with interval [1]
1703 [main] WARN  o.a.s.u.Utils - STORM-VERSION new 1.2.1.3.1.0.0-78 old null
1741 [main] INFO  o.a.s.StormSubmitter - Generated ZooKeeper secret payload for MD5-digest: -6887613048810109930:-7282845338058760974
1918 [main] INFO  o.a.s.u.NimbusClient - Found leader nimbus : zzeng-hdp-1.field.hortonworks.com:6627
1956 [main] INFO  o.a.s.s.a.AuthUtils - Got AutoCreds []
1967 [main] INFO  o.a.s.u.NimbusClient - Found leader nimbus : zzeng-hdp-1.field.hortonworks.com:6627
2067 [main] INFO  o.a.s.StormSubmitter - Uploading dependencies - jars...
2068 [main] INFO  o.a.s.StormSubmitter - Uploading dependencies - artifacts...
2069 [main] INFO  o.a.s.StormSubmitter - Dependency Blob keys - jars : [] / artifacts : []
2075 [main] INFO  o.a.s.StormSubmitter - Uploading topology jar /tmp/bdd4785432d311e9b3aefa163e67d972.jar to assigned location: /hadoop/storm/nimbus/inbox/stormjar-7f8a776d-65bd-42d7-a210-0ceca964b773.jar
3485 [main] INFO  o.a.s.StormSubmitter - Successfully uploaded topology jar to assigned location: /hadoop/storm/nimbus/inbox/stormjar-7f8a776d-65bd-42d7-a210-0ceca964b773.jar
3485 [main] INFO  o.a.s.StormSubmitter - Submitting topology storm-demo-topology-01 in distributed mode with conf {"storm.zookeeper.topology.auth.scheme":"digest","storm.zookeeper.topology.auth.payload":"-6887613048810109930:-7282845338058760974","topology.workers":3,"topology.debug":true}
3485 [main] WARN  o.a.s.u.Utils - STORM-VERSION new 1.2.1.3.1.0.0-78 old 1.2.1.3.1.0.0-78
4109 [main] INFO  o.a.s.StormSubmitter - Finished submitting topology: storm-demo-topology-01
4109 [main] INFO  o.a.s.StormSubmitter - Initializing the registered ISubmitterHook [org.apache.atlas.storm.hook.StormAtlasHook]
4217 [main] INFO  o.a.a.ApplicationProperties - Looking for atlas-application.properties in classpath
4217 [main] INFO  o.a.a.ApplicationProperties - Loading atlas-application.properties from file:/etc/storm/3.1.0.0-78/0/atlas-application.properties
log4j:ERROR A "org.apache.log4j.ConsoleAppender" object is not assignable to a "org.apache.log4j.Appender" variable.
log4j:ERROR The class "org.apache.log4j.Appender" was loaded by
log4j:ERROR [org.apache.atlas.plugin.classloader.AtlasPluginClassLoader@7c7d3c46] whereas object of type
log4j:ERROR "org.apache.log4j.ConsoleAppender" was loaded by [sun.misc.Launcher$AppClassLoader@6ff3c5b5].
log4j:ERROR Could not instantiate appender named "stdout".
4296 [main] INFO  o.a.a.ApplicationProperties - No graphdb backend specified. Will use 'janus'
4296 [main] INFO  o.a.a.ApplicationProperties - Using storage backend 'hbase2'
4297 [main] INFO  o.a.a.ApplicationProperties - Using index backend 'solr'
4297 [main] INFO  o.a.a.ApplicationProperties - Setting solr-wait-searcher property 'true'
4297 [main] INFO  o.a.a.ApplicationProperties - Setting index.search.map-name property 'false'
4305 [main] INFO  o.a.a.ApplicationProperties - Property (set to default) atlas.graph.cache.db-cache = true
4306 [main] INFO  o.a.a.ApplicationProperties - Property (set to default) atlas.graph.cache.db-cache-clean-wait = 20
4306 [main] INFO  o.a.a.ApplicationProperties - Property (set to default) atlas.graph.cache.db-cache-size = 0.5
4306 [main] INFO  o.a.a.ApplicationProperties - Property (set to default) atlas.graph.cache.tx-cache-size = 15000
4306 [main] INFO  o.a.a.ApplicationProperties - Property (set to default) atlas.graph.cache.tx-dirty-size = 120
4680 [main] WARN  o.a.a.h.AtlasHook - Error in determining keytab for KafkaClient-JAAS config
java.lang.RuntimeException: java.lang.RuntimeException: class org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not org.apache.hadoop.security.GroupMappingServiceProvider
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2628) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.<init>(Groups.java:107) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.<init>(Groups.java:102) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:451) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:337) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:304) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.doSubjectLogin(UserGroupInformation.java:1828) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.createLoginUser(UserGroupInformation.java:710) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:660) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.isLoginKeytabBased(UserGroupInformation.java:1236) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.atlas.hook.AtlasHook.isLoginKeytabBased(AtlasHook.java:288) [atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.hook.AtlasHook.<clinit>(AtlasHook.java:92) [atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at java.lang.Class.forName0(Native Method) ~[?:1.8.0_112]
	at java.lang.Class.forName(Class.java:348) [?:1.8.0_112]
	at org.apache.atlas.storm.hook.StormAtlasHook.initialize(StormAtlasHook.java:78) [storm-bridge-shim-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.storm.hook.StormAtlasHook.<init>(StormAtlasHook.java:46) [storm-bridge-shim-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:1.8.0_112]
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) [?:1.8.0_112]
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) [?:1.8.0_112]
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423) [?:1.8.0_112]
	at java.lang.Class.newInstance(Class.java:442) [?:1.8.0_112]
	at org.apache.storm.StormSubmitter.invokeSubmitterHook(StormSubmitter.java:358) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopologyAs(StormSubmitter.java:275) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:387) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:159) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at com.dsinpractice.storm.samples.WordCountTopology.main(WordCountTopology.java:140) [bdd4785432d311e9b3aefa163e67d972.jar:?]
Caused by: java.lang.RuntimeException: class org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not org.apache.hadoop.security.GroupMappingServiceProvider
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2622) ~[bdd4785432d311e9b3aefa163e67d972.jar:?]
	... 25 more
4698 [main] WARN  o.a.a.h.AtlasHook - Error in determining ticket-cache for KafkaClient-JAAS config
java.lang.RuntimeException: java.lang.RuntimeException: class org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not org.apache.hadoop.security.GroupMappingServiceProvider
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2628) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.<init>(Groups.java:107) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.<init>(Groups.java:102) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:451) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:337) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:304) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.doSubjectLogin(UserGroupInformation.java:1828) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.createLoginUser(UserGroupInformation.java:710) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:660) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.isLoginTicketBased(UserGroupInformation.java:1244) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.atlas.hook.AtlasHook.isLoginTicketBased(AtlasHook.java:300) [atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.hook.AtlasHook.<clinit>(AtlasHook.java:93) [atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at java.lang.Class.forName0(Native Method) ~[?:1.8.0_112]
	at java.lang.Class.forName(Class.java:348) [?:1.8.0_112]
	at org.apache.atlas.storm.hook.StormAtlasHook.initialize(StormAtlasHook.java:78) [storm-bridge-shim-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.storm.hook.StormAtlasHook.<init>(StormAtlasHook.java:46) [storm-bridge-shim-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:1.8.0_112]
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) [?:1.8.0_112]
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) [?:1.8.0_112]
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423) [?:1.8.0_112]
	at java.lang.Class.newInstance(Class.java:442) [?:1.8.0_112]
	at org.apache.storm.StormSubmitter.invokeSubmitterHook(StormSubmitter.java:358) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopologyAs(StormSubmitter.java:275) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:387) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:159) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at com.dsinpractice.storm.samples.WordCountTopology.main(WordCountTopology.java:140) [bdd4785432d311e9b3aefa163e67d972.jar:?]
Caused by: java.lang.RuntimeException: class org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not org.apache.hadoop.security.GroupMappingServiceProvider
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2622) ~[bdd4785432d311e9b3aefa163e67d972.jar:?]
	... 25 more
4716 [main] INFO  o.a.a.k.KafkaNotification - ==> KafkaNotification()
4719 [main] INFO  o.a.a.k.KafkaNotification - <== KafkaNotification()
4728 [main] WARN  o.a.a.h.AtlasHook - Error in determining current user
java.lang.RuntimeException: java.lang.RuntimeException: class org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not org.apache.hadoop.security.GroupMappingServiceProvider
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2628) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.<init>(Groups.java:107) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.<init>(Groups.java:102) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:451) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:337) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:304) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.doSubjectLogin(UserGroupInformation.java:1828) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.createLoginUser(UserGroupInformation.java:710) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:660) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:571) ~[hadoop-common-3.1.1.3.1.0.0-78.jar:?]
	at org.apache.atlas.hook.AtlasHook.getUser(AtlasHook.java:277) ~[atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.hook.AtlasHook.getUser(AtlasHook.java:246) ~[atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.hook.AtlasHook.<clinit>(AtlasHook.java:105) [atlas-notification-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at java.lang.Class.forName0(Native Method) ~[?:1.8.0_112]
	at java.lang.Class.forName(Class.java:348) [?:1.8.0_112]
	at org.apache.atlas.storm.hook.StormAtlasHook.initialize(StormAtlasHook.java:78) [storm-bridge-shim-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at org.apache.atlas.storm.hook.StormAtlasHook.<init>(StormAtlasHook.java:46) [storm-bridge-shim-1.1.0.3.1.0.0-78.jar:1.1.0.3.1.0.0-78]
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:1.8.0_112]
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) [?:1.8.0_112]
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) [?:1.8.0_112]
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423) [?:1.8.0_112]
	at java.lang.Class.newInstance(Class.java:442) [?:1.8.0_112]
	at org.apache.storm.StormSubmitter.invokeSubmitterHook(StormSubmitter.java:358) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopologyAs(StormSubmitter.java:275) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:387) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at org.apache.storm.StormSubmitter.submitTopology(StormSubmitter.java:159) [storm-core-1.2.1.3.1.0.0-78.jar:1.2.1.3.1.0.0-78]
	at com.dsinpractice.storm.samples.WordCountTopology.main(WordCountTopology.java:140) [bdd4785432d311e9b3aefa163e67d972.jar:?]
Caused by: java.lang.RuntimeException: class org.apache.hadoop.security.JniBasedUnixGroupsMappingWithFallback not org.apache.hadoop.security.GroupMappingServiceProvider
	at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2622) ~[bdd4785432d311e9b3aefa163e67d972.jar:?]
	... 26 more
4741 [main] INFO  o.a.a.h.AtlasHook - Created Atlas Hook
4751 [main] INFO  o.a.s.u.NimbusClient - Found leader nimbus : zzeng-hdp-1.field.hortonworks.com:6627
5093 [main] INFO  o.a.s.StormSubmitter - Invoking the registered ISubmitterHook [org.apache.atlas.storm.hook.StormAtlasHook]
5094 [main] INFO  o.a.a.s.h.StormAtlasHook - Collecting metadata for a new storm topology: storm-demo-topology-01
5695 [Atlas Notifier 0] INFO  o.a.k.c.p.ProducerConfig - ProducerConfig values:
	acks = 1
	batch.size = 16384
	bootstrap.servers = [zzeng-hdp-3.field.hortonworks.com:6667, zzeng-hdp-1.field.hortonworks.com:6667, zzeng-hdp-2.field.hortonworks.com:6667]
	buffer.memory = 33554432
	client.id =
	compression.type = none
	connections.max.idle.ms = 540000
	enable.idempotence = false
	interceptor.classes = []
	key.serializer = class org.apache.kafka.common.serialization.StringSerializer
	linger.ms = 0
	max.block.ms = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 0
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.timeout.ms = 60000
	transactional.id = null
	value.serializer = class org.apache.kafka.common.serialization.StringSerializer

5803 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'key.deserializer' was supplied but isn't a known config.
5803 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'value.deserializer' was supplied but isn't a known config.
5805 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'hook.group.id' was supplied but isn't a known config.
5806 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'zookeeper.connection.timeout.ms' was supplied but isn't a known config.
5806 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'zookeeper.session.timeout.ms' was supplied but isn't a known config.
5806 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'enable.auto.commit' was supplied but isn't a known config.
5807 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'zookeeper.connect' was supplied but isn't a known config.
5807 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'zookeeper.sync.time.ms' was supplied but isn't a known config.
5807 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'session.timeout.ms' was supplied but isn't a known config.
5807 [Atlas Notifier 0] WARN  o.a.k.c.p.ProducerConfig - The configuration 'auto.offset.reset' was supplied but isn't a known config.
5811 [Atlas Notifier 0] INFO  o.a.k.c.u.AppInfoParser - Kafka version : 2.0.0.3.1.0.0-78
5811 [Atlas Notifier 0] INFO  o.a.k.c.u.AppInfoParser - Kafka commitId : 0f47b27cde30d177
6039 [kafka-producer-network-thread | producer-1] INFO  o.a.k.c.Metadata - Cluster ID: 6qrlmnM0TSanZmrZy_eGHQ
[centos@zzeng-hdp-1 ~/sandbox/crosscomponent_demo/crosscomponent_scripts/storm-demo]$
```

</div>
</details>

Hook:

```bash
4741 [main] INFO  o.a.a.h.AtlasHook - Created Atlas Hook
4751 [main] INFO  o.a.s.u.NimbusClient - Found leader nimbus : zzeng-hdp-1.field.hortonworks.com:6627
5093 [main] INFO  o.a.s.StormSubmitter - Invoking the registered ISubmitterHook [org.apache.atlas.storm.hook.StormAtlasHook]
5094 [main] INFO  o.a.a.s.h.StormAtlasHook - Collecting metadata for a new storm topology: storm-demo-topology-01
5695 [Atlas Notifier 0] INFO  o.a.k.c.p.ProducerConfig - ProducerConfig values:
	acks = 1
	batch.size = 16384
	bootstrap.servers = [zzeng-hdp-3.field.hortonworks.com:6667, zzeng-hdp-1.field.hortonworks.com:6667, zzeng-hdp-2.field.hortonworks.com:6667]
	buffer.memory = 33554432
	client.id =
	compression.type = none
	connections.max.idle.ms = 540000
	enable.idempotence = false
	interceptor.classes = []
	key.serializer = class org.apache.kafka.common.serialization.StringSerializer
	linger.ms = 0
	max.block.ms = 60000
	max.in.flight.requests.per.connection = 5
	max.request.size = 1048576
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 0
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	transaction.timeout.ms = 60000
	transactional.id = null
	value.serializer = class org.apache.kafka.common.serialization.StringSerializer
```
## Atlas lineage : Kafka + Storm

<img width="1999" alt="image.png" src="https://qiita-image-store.s3.amazonaws.com/0/133516/40ae35d3-4d4a-406e-17c2-1249e25e911e.png">

<img width="629" alt="image.png" src="https://qiita-image-store.s3.amazonaws.com/0/133516/8d5eba35-9c87-4942-f39d-40fc50e620a8.png">

<img width="805" alt="image.png" src="https://qiita-image-store.s3.amazonaws.com/0/133516/708635fe-8521-2f73-d6b4-3f08c0a00e3a.png">

<img width="672" alt="image.png" src="https://qiita-image-store.s3.amazonaws.com/0/133516/216abf39-47df-3df8-d2ff-d5b037459e42.png">


## Kafka topic metadata in Atlas
![image.png](https://qiita-image-store.s3.amazonaws.com/0/133516/b4d589e3-4ea0-999f-cb33-98517e92b40c.png)

### Stormを２回実行したら：
Lineageの線が増える。。。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/133516/c14430bb-4c35-7624-2f1e-fdcecd2395a0.png)

## Atlas画面操作とStormでの操作により、AtlasのMetadataが更新されるが、履歴上では誰が更新したかもわかる

![image.png](https://qiita-image-store.s3.amazonaws.com/0/133516/c06d549d-46f8-6bcb-ba8f-5775b31a4c18.png)
新しい順。

* 一番上は僕がTagをつけた、作業者はAdminの僕。
* ２番目はStorm処理で生成したAtlas Entry
* ３番目もStorm処理で生成したAtlas Entry