def METHOD_NAME(self):
    factory = self.replay_flight_data('test_kafka_tag')
    p = self.load_policy({
        'name': 'kafka',
        'resource': 'aws.kafka',
        'filters': [
            {'tag:App': 'absent'},
            {'tag:Env': 'Dev'}],
        'actions': [
            {'type': 'tag',
             'tags': {'App': 'Custodian'}},
            {'type': 'remove-tag',
             'tags': ['Env']}]},
        session_factory=factory,
        config={'region': 'ap-northeast-2'}
    )
    resources = p.run()
    assert len(resources) == 1
    assert resources[0]['ClusterName'] == 'demo-cluster-1'
    client = factory().client('kafka', region_name='ap-northeast-2')
    assert client.list_tags_for_resource(
        ResourceArn=resources[0]['ClusterArn'])['Tags'] == {
            'App': 'Custodian'}