　 hyperledger-fabricにブロックチェーンを学ぶ（1/9）ブロックチェーンとは
　 hyperledger-fabricにブロックチェーンを学ぶ（2/9）サンプルを動かす
　 [hyperledger-fabricにブロックチェーンを学ぶ（3/9）ブロックを見る]
(https://qiita.com/ctrlzr/items/487e4055151aab9a5591)
⇒ hyperledger-fabricにブロックチェーンを学ぶ（4/9）kafkaを入れる
　 hyperledger-fabricにブロックチェーンを学ぶ（5/9）可視化する

Why kafka？

kafkaは後から入れられない。

kafkaを入れる主目的は単一障害点の回避です。ただ、
ブロックチェーンネットワークを起動するとき、1番目(No.0)のブロックはgenesis.blockで作成され、後から変更することはできません。
genesis.block に orderer Typeが書かれているので、スモールスタートで、あとから追加すればいいやと考えていると詰みます。

環境構成

hyperledger公式サイトに従い、kafka4台、zookeeper3台にしました。ついでにcouch dbもいれてデータを保存できるようにしました。ordererは1つのままです。

kafka, zookeeper, orderer の関係性を理解しておらず、構成図はIBMに頼ります
IBMの図ではordererが2つあるので単一障害点は回避されますが、トランザクションの前後関係を保つ仕組みが理解できていません。トランザクション日時はブロックチェーンネットワークの外から渡されるので、それで前後関係を保つことはできません。
複数ordererでトランザクションの前後関係を保証する仕組みをご存知の方は教えてください。

環境構築

configtxgen コマンドの作成
genesis.blockの作成
base.yamlの修正
docker-compose.yamlの修正

configtxgenコマンドの作成


$ go env|grep GOROOT
# GOROOT="/usr/lib/go"

$ sudo mkdir -p /usr/lib/go/src/github.com/hyperledger/ 
$ cd /usr/lib/go/src/github.com/hyperledger/ 
$ git clone https://github.com/hyperledger/fabric.git
$ cd /usr/lib/go/src/github.com/hyperledger/fabric
$ sudo make configtxgen configtxlator cryptogen # 後で使うので configtxlator cryptogen も追加
# find: `/src/github.com/hyperledger/fabric/core/chaincode/shim': そのようなファイルやディレクトリはありません
# .build/bin/configtxlator
# CGO_CFLAGS=" " GOBIN=/usr/lib/go/src/github.com/hyperledger/fabric/.build/bin go # install -tags "" -ldflags "-X # github.com/hyperledger/fabric/common/tools/configtxlator/metadata.CommitSHA=f5210b5" github.com/hyperledger/fabric/common/tools/configtxlator
# Binary available as .build/bin/configtxlator
# .build/bin/cryptogen
# CGO_CFLAGS=" " GOBIN=/usr/lib/go/src/github.com/hyperledger/fabric/.build/bin go # install -tags "" -ldflags "-X  github.com/hyperledger/fabric/common/tools/cryptogen/metadata.CommitSHA=f5210b5" github.com/hyperledger/fabric/common/tools/cryptogen
# Binary available as .build/bin/cryptogen

$ ll /usr/lib/go/src/github.com/hyperledger/fabric/.build/bin
# -rwxr-xr-x 1 root root 21257692  4月 30 12:22 2019 configtxgen
# -rwxr-xr-x 1 root root 23528734  4月 30 12:30 2019 configtxlator
# -rwxr-xr-x 1 root root 14009802  4月 30 12:30 2019 cryptogen

# コマンドが使えるようにPATHに入れる
$ vi ~/.bash_profile
# 以下を追加
# PATH=$PATH:/usr/lib/go/src/github.com/hyperledger/fabric/.build/bin

$ source ~/.bash_profile

genesis.blockの作成

# configtx.yamlの編集
$ vi /var/www/fabric-samples/balance-transfer/artifacts/channel/configtx.yaml
# 以下を追加
#    TwoOrgsOrdererGenesisKafka:
#        Orderer:
#            <<: *OrdererDefaults
#            OrdererType: kafka
#            Kafka:
#                Brokers:
#                - kafka0.example.com:9092
#                - kafka1.example.com:9092
#                - kafka2.example.com:9092
#                - kafka3.example.com:9092
#            Organizations:
#                <<: *OrdererOrg
#                Policies:
#                    Readers:
#                        Type: Signature
#                        Rule: "OR('OrdererMSP.member')"
#                    Writers:
#                        Type: Signature
#                        Rule: "OR('OrdererMSP.member')"
#                    Admins:
#                        Type: Signature
#                        Rule: "OR('OrdererMSP.admin')"
#        Consortiums:
#            SampleConsortium:
#                Organizations:
#                    - *Org1
#                    - *Org2

$ cd /var/www/fabric-samples/balance-transfer/artifacts/channel/
$ configtxgen -profile=TwoOrgsOrdererGenesisKafka -channelID=mychannel -outputBlock=genesis.block
# [2019-04-19 21:22:20.688 JST [common.tools.configtxgen] main -> INFO 001 Loading configuration
# [2019-04-19 21:22:20.698 JST [common.tools.configtxgen.localconfig] completeInitialization -> INFO 002 orderer type: kafka
# [2019-04-19 21:22:20.699 JST [common.tools.configtxgen.localconfig] Load -> INFO 003 Loaded configuration: C:\Users\ando\ando\blockchain\balance-transfer\artifacts\channel\configtx.yaml
# [2019-04-19 21:22:20.778 JST [common.tools.configtxgen.localconfig] completeInitialization -> INFO 004 orderer type: kafka
# [2019-04-19 21:22:20.779 JST [common.tools.configtxgen.localconfig] LoadTopLevel -> INFO 005 Loaded configuration: C:\Users\ando\ando\blockchain\balance-transfer\artifacts\channel\configtx.yaml
# [2019-04-19 21:22:20.781 JST [common.tools.configtxgen.encoder] NewChannelGroup -> WARN 006 Default policy emission is deprecated, please include policy specifications for the channel group in configtx.yaml
# [2019-04-19 21:22:20.783 JST [common.tools.configtxgen.encoder] NewOrdererGroup -> WARN 007 Default policy emission is deprecated, please include policy specifications for the orderer group in configtx.yaml
# [2019-04-19 21:22:20.838 JST [common.tools.configtxgen.encoder] NewOrdererOrgGroup -> WARN 008 Default policy emission is deprecated, please include policy specifications for the orderer org group Org1MSP in configtx.yaml
# [2019-04-19 21:22:20.842 JST [common.tools.configtxgen.encoder] NewOrdererOrgGroup -> WARN 009 Default policy emission is deprecated, please include policy specifications for the orderer org group Org2MSP in configtx.yaml
# [2019-04-19 21:22:20.844 JST [common.tools.configtxgen] doOutputBlock -> INFO 00a Generating genesis block
# [2019-04-19 21:22:20.850 JST [common.tools.configtxgen] doOutputBlock -> INFO 00b Writing genesis block

# WARNは気になる人はPoliciesを追記すればWARNは消えます。ただし、
# balance-transferサンプルはぺーぺーのJimとBarryで実行するため、サンプルを修正する必要があります。

vi /var/www/fabric-samples/balance-transfer/base.yaml

base.yamlの修正

# Copyright IBM Corp. All Rights Reserved.
#
# SPDX-License-Identifier: Apache-2.0
#

version: '2'
services:

  peer-base:
    image: hyperledger/fabric-peer:1.4
    environment:
      - CORE_VM_ENDPOINT=unix:///host/var/run/docker.sock
      # the following setting starts chaincode containers on the same
      # bridge network as the peers
      # https://docs.docker.com/compose/networking/
      - CORE_VM_DOCKER_HOSTCONFIG_NETWORKMODE=artifacts_default
      - FABRIC_LOGGING_SPEC=DEBUG
      - CORE_PEER_GOSSIP_USELEADERELECTION=true
      - CORE_PEER_GOSSIP_ORGLEADER=false
      # The following setting skips the gossip handshake since we are
      # are not doing mutual TLS
      - CORE_PEER_GOSSIP_SKIPHANDSHAKE=true
      - CORE_PEER_MSPCONFIGPATH=/etc/hyperledger/crypto/peer/msp
      - CORE_PEER_TLS_ENABLED=true
      - CORE_PEER_TLS_KEY_FILE=/etc/hyperledger/crypto/peer/tls/server.key
      - CORE_PEER_TLS_CERT_FILE=/etc/hyperledger/crypto/peer/tls/server.crt
      - CORE_PEER_TLS_ROOTCERT_FILE=/etc/hyperledger/crypto/peer/tls/ca.crt
    working_dir: /opt/gopath/src/github.com/hyperledger/fabric/peer
    command: peer node start
    volumes:
        - /var/run/:/host/var/run/

  ca-base:
    image: hyperledger/fabric-ca:1.4
    environment:
      - FABRIC_CA_HOME=/etc/hyperledger/fabric-ca-server
      - FABRIC_CA_SERVER_TLS_ENABLED=true
    command: sh -c 'fabric-ca-server start -b admin:adminpw -d'

  orderer-base:
    image: hyperledger/fabric-orderer:1.4
    environment:
      - ORDERER_GENERAL_LOGLEVEL=debug
      - ORDERER_GENERAL_LISTENADDRESS=0.0.0.0
      - ORDERER_GENERAL_GENESISMETHOD=file
      - ORDERER_GENERAL_GENESISFILE=/etc/hyperledger/configtx/genesis.block
      - ORDERER_GENERAL_LOCALMSPID=OrdererMSP
      - ORDERER_GENERAL_LOCALMSPDIR=/etc/hyperledger/crypto/orderer/msp
      - ORDERER_GENERAL_TLS_ENABLED=true
      - ORDERER_GENERAL_TLS_PRIVATEKEY=/etc/hyperledger/crypto/orderer/tls/server.key
      - ORDERER_GENERAL_TLS_CERTIFICATE=/etc/hyperledger/crypto/orderer/tls/server.crt
    working_dir: /opt/gopath/src/github.com/hyperledger/fabric/orderers
    command: orderer
    volumes:
      - ./channel:/etc/hyperledger/configtx
      - ./channel/crypto-config/ordererOrganizations/example.com/orderers/orderer.example.com/:/etc/hyperledger/crypto/orderer

  zookeeper-base:
    image: hyperledger/fabric-zookeeper:0.4.15
    ports:
      - 2181
      - 2888
      - 3888
  
  kafka-base:
    image: hyperledger/fabric-kafka:0.4.15
    environment:
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_MESSAGE_MAX_BYTES=1048576 # 1 * 1024 * 1024 B
      - KAFKA_REPLICA_FETCH_MAX_BYTES=1048576 # 1 * 1024 * 1024 B
      - KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE=true
      - KAFKA_LOG_RETENTION_MS=-1
      - KAFKA_MIN_INSYNC_REPLICAS=2
      - KAFKA_DEFAULT_REPLICATION_FACTOR=3
    ports:
      - 9092

  couchdb-base:
    image: hyperledger/fabric-couchdb:0.4.15

base.yamlの修正ポイントは2つ

imageバージョンとnode_moduleバージョンと合わせる

node_moduleのfabric-sdkバージョンはpackage.jsonに記述されている。例は1.4.x
base.yamlに記述するdockerイメージバージョンと同じにすること。

$ cat /var/www/fabric-samples/balance-transfer/package.json|grep fabric-c
# "fabric-client sample app",
# "fabric-ca-client": "~1.4.0",
# "fabric-client": "~1.4.0",

バージョンをbase.yamlで管理する

balance-transferではbase.yamlにpeerしか記述されていない。
まとめてバージョン管理するため、base.yaml に全てのイメージを記載する。

vi /var/www/fabric-samples/balance-transfer/docker-compose.yaml

docker-compose.yamlの修正

#
# Copyright IBM Corp. All Rights Reserved.
#
# SPDX-License-Identifier: Apache-2.0
#
version: '2'

services:

  ca.org1.example.com:
    extends:
      file: base.yaml
      service: ca-base
    environment:
      - FABRIC_CA_SERVER_CA_NAME=ca-org1
      - FABRIC_CA_SERVER_CA_CERTFILE=/etc/hyperledger/fabric-ca-server-config/ca.org1.example.com-cert.pem
      - FABRIC_CA_SERVER_CA_KEYFILE=/etc/hyperledger/fabric-ca-server-config/0e729224e8b3f31784c8a93c5b8ef6f4c1c91d9e6e577c45c33163609fe40011_sk
      - FABRIC_CA_SERVER_TLS_CERTFILE=/etc/hyperledger/fabric-ca-server-config/ca.org1.example.com-cert.pem
      - FABRIC_CA_SERVER_TLS_KEYFILE=/etc/hyperledger/fabric-ca-server-config/0e729224e8b3f31784c8a93c5b8ef6f4c1c91d9e6e577c45c33163609fe40011_sk
    ports:
      - "7054:7054"
    volumes:
      - ./channel/crypto-config/peerOrganizations/org1.example.com/ca/:/etc/hyperledger/fabric-ca-server-config
      - ./data/ca/org1.ca0/:/etc/hyperledger/fabric-ca-server
    container_name: ca_peerOrg1

  ca.org2.example.com:
    extends:
      file: base.yaml
      service: ca-base
    environment:
      - FABRIC_CA_SERVER_CA_NAME=ca-org2
      - FABRIC_CA_SERVER_CA_CERTFILE=/etc/hyperledger/fabric-ca-server-config/ca.org2.example.com-cert.pem
      - FABRIC_CA_SERVER_CA_KEYFILE=/etc/hyperledger/fabric-ca-server-config/a7d47efa46a6ba07730c850fed2c1375df27360d7227f48cdc2f80e505678005_sk
      - FABRIC_CA_SERVER_TLS_CERTFILE=/etc/hyperledger/fabric-ca-server-config/ca.org2.example.com-cert.pem
      - FABRIC_CA_SERVER_TLS_KEYFILE=/etc/hyperledger/fabric-ca-server-config/a7d47efa46a6ba07730c850fed2c1375df27360d7227f48cdc2f80e505678005_sk
    ports:
      - "8054:7054"
    volumes:
      - ./channel/crypto-config/peerOrganizations/org2.example.com/ca/:/etc/hyperledger/fabric-ca-server-config
      - ./data/ca/org2.ca0/:/etc/hyperledger/fabric-ca-server
    container_name: ca_peerOrg2

  orderer.example.com:
    container_name: orderer.example.com
    extends:
      file: base.yaml
      service: orderer-base
    environment:
      - ORDERER_GENERAL_TLS_ROOTCAS=[/etc/hyperledger/crypto/orderer/tls/ca.crt, /etc/hyperledger/crypto/peerOrg1/tls/ca.crt, /etc/hyperledger/crypto/peerOrg2/tls/ca.crt]
    ports:
      - 7050:7050
    volumes:
        - ./channel/crypto-config/peerOrganizations/org1.example.com/peers/peer0.org1.example.com/:/etc/hyperledger/crypto/peerOrg1
        - ./channel/crypto-config/peerOrganizations/org2.example.com/peers/peer0.org2.example.com/:/etc/hyperledger/crypto/peerOrg2
        - ./data/orderer/orderer0:/var/hyperledger/
    depends_on:
      - kafka0.example.com
      - kafka1.example.com
      - kafka2.example.com
      - kafka3.example.com

  zookeeper0.example.com:
    container_name: zookeeper0.example.com
    extends:
      file: base.yaml
      service: zookeeper-base
    environment:
      - ZOO_MY_ID=1
      - ZOO_SERVERS=server.1=zookeeper0.example.com:2888:3888 server.2=zookeeper1.example.com:2888:3888 server.3=zookeeper2.example.com:2888:3888

  zookeeper1.example.com:
    container_name: zookeeper1.example.com
    extends:
      file: base.yaml
      service: zookeeper-base
    environment:
      - ZOO_MY_ID=2
      - ZOO_SERVERS=server.1=zookeeper0.example.com:2888:3888 server.2=zookeeper1.example.com:2888:3888 server.3=zookeeper2.example.com:2888:3888

  zookeeper2.example.com:
    container_name: zookeeper2.example.com
    extends:
      file: base.yaml
      service: zookeeper-base
    environment:
      - ZOO_MY_ID=3
      - ZOO_SERVERS=server.1=zookeeper0.example.com:2888:3888 server.2=zookeeper1.example.com:2888:3888 server.3=zookeeper2.example.com:2888:3888

  kafka0.example.com:
    container_name: kafka0.example.com
    extends:
      file: base.yaml
      service: kafka-base
    depends_on:
      - zookeeper0.example.com
      - zookeeper1.example.com
      - zookeeper2.example.com
    environment:
      - KAFKA_BROKER_ID=0
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper0.example.com:2181,zookeeper1.example.com:2181,zookeeper2.example.com:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka0.example.com:9092
    volumes:
      - ./data/kafka/kafka0/kafka-logs:/tmp/kafka-logs

  kafka1.example.com:
    container_name: kafka1.example.com
    extends:
      file: base.yaml
      service: kafka-base
    depends_on:
      - zookeeper0.example.com
      - zookeeper1.example.com
      - zookeeper2.example.com
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper0.example.com:2181,zookeeper1.example.com:2181,zookeeper2.example.com:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka1.example.com:9092
    volumes:
      - ./data/kafka/kafka1/kafka-logs:/tmp/kafka-logs

  kafka2.example.com:
    container_name: kafka2.example.com
    extends:
      file: base.yaml
      service: kafka-base
    depends_on:
      - zookeeper0.example.com
      - zookeeper1.example.com
      - zookeeper2.example.com
    environment:
      - KAFKA_BROKER_ID=2
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper0.example.com:2181,zookeeper1.example.com:2181,zookeeper2.example.com:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka2.example.com:9092
    volumes:
      - ./data/kafka/kafka2/kafka-logs:/tmp/kafka-logs

  kafka3.example.com:
    container_name: kafka3.example.com
    extends:
      file: base.yaml
      service: kafka-base
    depends_on:
      - zookeeper0.example.com
      - zookeeper1.example.com
      - zookeeper2.example.com
    environment:
      - KAFKA_BROKER_ID=3
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper0.example.com:2181,zookeeper1.example.com:2181,zookeeper2.example.com:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka3.example.com:9092
    volumes:
      - ./data/kafka/kafka3/kafka-logs:/tmp/kafka-logs

  peer0.org1.example.com:
    container_name: peer0.org1.example.com
    extends:
      file:   base.yaml
      service: peer-base
    environment:
      - CORE_PEER_ID=peer0.org1.example.com
      - CORE_PEER_LOCALMSPID=Org1MSP
      - CORE_PEER_ADDRESS=peer0.org1.example.com:7051
      - CORE_PEER_GOSSIP_BOOTSTRAP=peer1.org1.example.com:7051      
      - CORE_PEER_GOSSIP_EXTERNALENDPOINT=peer0.org1.example.com:7051
      - CORE_LEDGER_STATE_STATEDATABASE=CouchDB
      - CORE_LEDGER_STATE_COUCHDBCONFIG_COUCHDBADDRESS=couchdb_peer0.org1.example.com:5984
    ports:
      - 7051:7051
      - 7053:7053
    volumes:
        - ./channel/crypto-config/peerOrganizations/org1.example.com/peers/peer0.org1.example.com/:/etc/hyperledger/crypto/peer
        - ./data/peer/org1.peer0/:/var/hyperledger/
    depends_on:
      - orderer.example.com
      - couchdb_peer1.org2.example.com

  peer1.org1.example.com:
    container_name: peer1.org1.example.com
    extends:
      file:   base.yaml
      service: peer-base
    environment:
      - CORE_PEER_ID=peer1.org1.example.com
      - CORE_PEER_LOCALMSPID=Org1MSP
      - CORE_PEER_ADDRESS=peer1.org1.example.com:7051
      - CORE_PEER_GOSSIP_BOOTSTRAP=peer0.org1.example.com:7051      
      - CORE_PEER_GOSSIP_EXTERNALENDPOINT=peer1.org1.example.com:7051
      - CORE_LEDGER_STATE_STATEDATABASE=CouchDB
      - CORE_LEDGER_STATE_COUCHDBCONFIG_COUCHDBADDRESS=couchdb_peer1.org1.example.com:5984
    ports:
      - 7056:7051
      - 7058:7053
    volumes:
        - ./channel/crypto-config/peerOrganizations/org1.example.com/peers/peer1.org1.example.com/:/etc/hyperledger/crypto/peer
        - ./data/peer/org1.peer1/:/var/hyperledger/
    depends_on:
      - orderer.example.com
      - couchdb_peer1.org1.example.com

  peer0.org2.example.com:
    container_name: peer0.org2.example.com
    extends:
      file:   base.yaml
      service: peer-base
    environment:
      - CORE_PEER_ID=peer0.org2.example.com
      - CORE_PEER_LOCALMSPID=Org2MSP
      - CORE_PEER_ADDRESS=peer0.org2.example.com:7051
      - CORE_PEER_GOSSIP_BOOTSTRAP=peer1.org2.example.com:7051      
      - CORE_PEER_GOSSIP_EXTERNALENDPOINT=peer0.org2.example.com:7051
      - CORE_LEDGER_STATE_STATEDATABASE=CouchDB
      - CORE_LEDGER_STATE_COUCHDBCONFIG_COUCHDBADDRESS=couchdb_peer0.org2.example.com:5984
    ports:
      - 8051:7051
      - 8053:7053
    volumes:
        - ./channel/crypto-config/peerOrganizations/org2.example.com/peers/peer0.org2.example.com/:/etc/hyperledger/crypto/peer
        - ./data/peer/org2.peer0/:/var/hyperledger/
    depends_on:
      - orderer.example.com
      - couchdb_peer0.org2.example.com

  peer1.org2.example.com:
    container_name: peer1.org2.example.com
    extends:
      file:   base.yaml
      service: peer-base
    environment:
      - CORE_PEER_ID=peer1.org2.example.com
      - CORE_PEER_LOCALMSPID=Org2MSP
      - CORE_PEER_ADDRESS=peer1.org2.example.com:7051
      - CORE_PEER_GOSSIP_BOOTSTRAP=peer0.org2.example.com:7051      
      - CORE_PEER_GOSSIP_EXTERNALENDPOINT=peer1.org2.example.com:7051
      - CORE_LEDGER_STATE_STATEDATABASE=CouchDB
      - CORE_LEDGER_STATE_COUCHDBCONFIG_COUCHDBADDRESS=couchdb_peer1.org2.example.com:5984
    ports:
      - 8056:7051
      - 8058:7053
    volumes:
        - ./channel/crypto-config/peerOrganizations/org2.example.com/peers/peer1.org2.example.com/:/etc/hyperledger/crypto/peer
        - ./data/peer/org2.peer1/:/var/hyperledger/
    depends_on:
      - orderer.example.com
      - couchdb_peer1.org2.example.com

  couchdb_peer0.org1.example.com:
    container_name: couchdb_peer0.org1.example.com
    extends:
      file: base.yaml
      service: couchdb-base
    ports:
      - 5984:5984
    volumes:
      - ./data/couch_db/org1.peer0/:/opt/couchdb/data

  couchdb_peer1.org1.example.com:
    container_name: couchdb_peer1.org1.example.com
    extends:
      file: base.yaml
      service: couchdb-base
    ports:
      - 5989:5984
    volumes:
      - ./data/couch_db/org1.peer1/:/opt/couchdb/data

  couchdb_peer0.org2.example.com:
    container_name: couchdb_peer0.org2.example.com
    extends:
      file: base.yaml
      service: couchdb-base
    ports:
      - 5994:5984
    volumes:
      - ./data/couch_db/org2.peer0/:/opt/couchdb/data

  couchdb_peer1.org2.example.com:
    container_name: couchdb_peer1.org2.example.com
    extends:
      file: base.yaml
      service: couchdb-base
    ports:
      - 5999:5984
    volumes:
      - ./data/couch_db/org2.peer1/:/opt/couchdb/data

balance-transferの実行

コマンドは第2回を参照

実行結果

ブロックにkakaが含まれていることが分かる。

KafkaBrokers":{"version":0,"mod_policy":"Admins","value":{}},"ConsensusType":{"version":0,"mod_policy":"Admins","value":{"type":"kafka"}}