[KAFKA] Adding new client code

Signed-off-by: david.mcweeney <david.mcweeney@est.tech>
Change-Id: I38b930b1b5f4233f961d51bbab4b1828d034e67a
Issue-ID: DMAAP-1847

19 files changed, 0 insertions, 1192 deletions

diff --git a/src/main/docker/Dockerfile b/src/main/docker/Dockerfile
deleted file mode 100644
index 930f5ba..0000000
--- a/src/main/docker/Dockerfile
+++ /dev/null
@@ -1,29 +0,0 @@
-FROM confluentinc/cp-kafka:6.2.0
-
-ENV COMPONENT=kafka \
-    KAFKA_USER=mrkafka \
-    KAFKA_GROUP=onap
-
-COPY org.onap.dmaap.mr.trust.jks \
-     org.onap.dmaap.mr.p12 \
-     org.onap.dmaap.mr.keyfile \
-     /etc/${COMPONENT}/secrets/cert/
-
-USER root
-
-RUN userdel -r appuser && groupadd $KAFKA_GROUP && useradd $KAFKA_USER -u 1000 -G 1000,$KAFKA_GROUP
-
-WORKDIR /home/$KAFKA_USER
-
-COPY include/etc/confluent/docker/* /etc/confluent/docker/
-RUN chmod -R +x /etc/confluent/docker \
-&& mkdir -p /etc/${COMPONENT}/data /etc/${COMPONENT}/secrets \
-&& chown -R $KAFKA_USER:$KAFKA_GROUP /var/lib/${COMPONENT} /etc/${COMPONENT} /etc/confluent/docker /var/log/${COMPONENT} /var/lib/${COMPONENT} /var/log/confluent
-
-COPY kafka11aaf.jar /usr/share/java/${COMPONENT}/
-
-USER $KAFKA_USER
-
-EXPOSE 9092 9093
-
-CMD ["/etc/confluent/docker/run"]
\ No newline at end of file
diff --git a/src/main/docker/include/etc/confluent/docker/configure b/src/main/docker/include/etc/confluent/docker/configure
deleted file mode 100644
index f0d8701..0000000
--- a/src/main/docker/include/etc/confluent/docker/configure
+++ /dev/null
@@ -1,123 +0,0 @@
-#!/usr/bin/env bash
-#
-# Copyright 2016 Confluent Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-. /etc/confluent/docker/bash-config
-
-dub ensure KAFKA_ZOOKEEPER_CONNECT
-dub ensure KAFKA_ADVERTISED_LISTENERS
-
-# By default, LISTENERS is derived from ADVERTISED_LISTENERS by replacing
-# hosts with 0.0.0.0. This is good default as it ensures that the broker
-# process listens on all ports.
-if [[ -z "${KAFKA_LISTENERS-}" ]]
-then
-  export KAFKA_LISTENERS
-  KAFKA_LISTENERS=$(cub listeners "$KAFKA_ADVERTISED_LISTENERS")
-fi
-
-dub path /etc/kafka/ writable
-
-if [[ -z "${KAFKA_LOG_DIRS-}" ]]
-then
-  export KAFKA_LOG_DIRS
-  KAFKA_LOG_DIRS="/var/lib/kafka/data"
-fi
-
-# advertised.host, advertised.port, host and port are deprecated. Exit if these properties are set.
-if [[ -n "${KAFKA_ADVERTISED_PORT-}" ]]
-then
-  echo "advertised.port is deprecated. Please use KAFKA_ADVERTISED_LISTENERS instead."
-  exit 1
-fi
-
-if [[ -n "${KAFKA_ADVERTISED_HOST-}" ]]
-then
-  echo "advertised.host is deprecated. Please use KAFKA_ADVERTISED_LISTENERS instead."
-  exit 1
-fi
-
-if [[ -n "${KAFKA_HOST-}" ]]
-then
-  echo "host is deprecated. Please use KAFKA_ADVERTISED_LISTENERS instead."
-  exit 1
-fi
-
-if [[ -n "${KAFKA_PORT-}" ]]
-then
-  echo "port is deprecated. Please use KAFKA_ADVERTISED_LISTENERS instead."
-  exit 1
-fi
-
-# Set if ADVERTISED_LISTENERS has SSL:// or SASL_SSL:// endpoints.
-if [[ $KAFKA_ADVERTISED_LISTENERS == *"SSL://"* ]]
-then
-  echo "SSL is enabled."
-
-  dub ensure KAFKA_SSL_KEYSTORE_FILENAME
-  export KAFKA_SSL_KEYSTORE_LOCATION="/etc/kafka/secrets/$KAFKA_SSL_KEYSTORE_FILENAME"
-  dub path "$KAFKA_SSL_KEYSTORE_LOCATION" exists
-
-  dub ensure KAFKA_SSL_KEY_CREDENTIALS
-  KAFKA_SSL_KEY_CREDENTIALS_LOCATION="/etc/kafka/secrets/$KAFKA_SSL_KEY_CREDENTIALS"
-  dub path "$KAFKA_SSL_KEY_CREDENTIALS_LOCATION" exists
-  export KAFKA_SSL_KEY_PASSWORD
-  KAFKA_SSL_KEY_PASSWORD=$(cat "$KAFKA_SSL_KEY_CREDENTIALS_LOCATION")
-
-  dub ensure KAFKA_SSL_KEYSTORE_CREDENTIALS
-  KAFKA_SSL_KEYSTORE_CREDENTIALS_LOCATION="/etc/kafka/secrets/$KAFKA_SSL_KEYSTORE_CREDENTIALS"
-  dub path "$KAFKA_SSL_KEYSTORE_CREDENTIALS_LOCATION" exists
-  export KAFKA_SSL_KEYSTORE_PASSWORD
-  KAFKA_SSL_KEYSTORE_PASSWORD=$(cat "$KAFKA_SSL_KEYSTORE_CREDENTIALS_LOCATION")
-
-  if [[ -n "${KAFKA_SSL_CLIENT_AUTH-}" ]] && ( [[ $KAFKA_SSL_CLIENT_AUTH == *"required"* ]] || [[ $KAFKA_SSL_CLIENT_AUTH == *"requested"* ]] )
-  then
-      dub ensure KAFKA_SSL_TRUSTSTORE_FILENAME
-      export KAFKA_SSL_TRUSTSTORE_LOCATION="/etc/kafka/secrets/$KAFKA_SSL_TRUSTSTORE_FILENAME"
-      dub path "$KAFKA_SSL_TRUSTSTORE_LOCATION" exists
-
-      dub ensure KAFKA_SSL_TRUSTSTORE_CREDENTIALS
-      KAFKA_SSL_TRUSTSTORE_CREDENTIALS_LOCATION="/etc/kafka/secrets/$KAFKA_SSL_TRUSTSTORE_CREDENTIALS"
-      dub path "$KAFKA_SSL_TRUSTSTORE_CREDENTIALS_LOCATION" exists
-      export KAFKA_SSL_TRUSTSTORE_PASSWORD
-      KAFKA_SSL_TRUSTSTORE_PASSWORD=$(cat "$KAFKA_SSL_TRUSTSTORE_CREDENTIALS_LOCATION")
-  fi
-  
-fi
-
-# Set if KAFKA_ADVERTISED_LISTENERS has SASL_PLAINTEXT:// or SASL_SSL:// endpoints.
-if [[ $KAFKA_ADVERTISED_LISTENERS =~ .*SASL_.*://.* ]]
-then
-  echo "SASL" is enabled.
-
-  dub ensure KAFKA_OPTS
-
-  if [[ ! $KAFKA_OPTS == *"java.security.auth.login.config"*  ]]
-  then
-    echo "KAFKA_OPTS should contain 'java.security.auth.login.config' property."
-  fi
-fi
-
-if [[ -n "${KAFKA_JMX_OPTS-}" ]]
-then
-  if [[ ! $KAFKA_JMX_OPTS == *"com.sun.management.jmxremote.rmi.port"*  ]]
-  then
-    echo "KAFKA_OPTS should contain 'com.sun.management.jmxremote.rmi.port' property. It is required for accessing the JMX metrics externally."
-  fi
-fi
-
-dub template "/etc/confluent/docker/${COMPONENT}.properties.template" "/etc/${COMPONENT}/${COMPONENT}.properties"
-dub template "/etc/confluent/docker/log4j.properties.template" "/etc/${COMPONENT}/log4j.properties"
-dub template "/etc/confluent/docker/tools-log4j.properties.template" "/etc/${COMPONENT}/tools-log4j.properties"
diff --git a/src/main/docker/include/etc/confluent/docker/ensure b/src/main/docker/include/etc/confluent/docker/ensure
deleted file mode 100644
index 09160f0..0000000
--- a/src/main/docker/include/etc/confluent/docker/ensure
+++ /dev/null
@@ -1,29 +0,0 @@
-#!/usr/bin/env bash
-#
-# Copyright 2020 Confluent Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-. /etc/confluent/docker/bash-config
-
-export KAFKA_DATA_DIRS=${KAFKA_DATA_DIRS:-"/var/lib/kafka/data"}
-echo "===> Check if $KAFKA_DATA_DIRS is writable ..."
-dub path "$KAFKA_DATA_DIRS" writable
-
-if [[ -n "${KAFKA_ZOOKEEPER_SSL_CLIENT_ENABLE-}" ]] && [[ $KAFKA_ZOOKEEPER_SSL_CLIENT_ENABLE == "true" ]]
-then
-    echo "===> Skipping Zookeeper health check for SSL connections..."
-else
-    echo "===> Check if Zookeeper is healthy ..."
-    cub zk-ready "$KAFKA_ZOOKEEPER_CONNECT" "${KAFKA_CUB_ZK_TIMEOUT:-40}"
-fi
\ No newline at end of file
diff --git a/src/main/docker/include/etc/confluent/docker/kafka.properties.template b/src/main/docker/include/etc/confluent/docker/kafka.properties.template
deleted file mode 100644
index 5eeaea3..0000000
--- a/src/main/docker/include/etc/confluent/docker/kafka.properties.template
+++ /dev/null
@@ -1,33 +0,0 @@
-{% set excluded_props = ['KAFKA_VERSION',
-                         'KAFKA_HEAP_OPTS'
-                         'KAFKA_LOG4J_OPTS',
-                         'KAFKA_OPTS',
-                         'KAFKA_JMX_OPTS',
-                         'KAFKA_JVM_PERFORMANCE_OPTS',
-                         'KAFKA_GC_LOG_OPTS',
-                         'KAFKA_LOG4J_ROOT_LOGLEVEL',
-                         'KAFKA_LOG4J_LOGGERS',
-                         'KAFKA_TOOLS_LOG4J_LOGLEVEL',
-                         'KAFKA_ZOOKEEPER_CLIENT_CNXN_SOCKET']
--%}
-
-{# properties that don't fit the standard format #}
-{% set other_props = {
-  'KAFKA_ZOOKEEPER_CLIENT_CNXN_SOCKET' : 'zookeeper.clientCnxnSocket'
- } -%}
-
-{% set kafka_props = env_to_props('KAFKA_', '', exclude=excluded_props) -%}
-{% for name, value in kafka_props.items() -%}
-{{name}}={{value}}
-{% endfor -%}
-
-{% for k, property in other_props.items() -%}
-{% if env.get(k) != None -%}
-{{property}}={{env[k]}}
-{% endif -%}
-{% endfor -%}
-
-{% set confluent_support_props = env_to_props('CONFLUENT_SUPPORT_', 'confluent.support.') -%}
-{% for name, value in confluent_support_props.items() -%}
-{{name}}={{value}}
-{% endfor -%}
diff --git a/src/main/docker/include/etc/confluent/docker/launch b/src/main/docker/include/etc/confluent/docker/launch
deleted file mode 100644
index d1eaf56..0000000
--- a/src/main/docker/include/etc/confluent/docker/launch
+++ /dev/null
@@ -1,37 +0,0 @@
-#!/usr/bin/env bash
-#
-# Copyright 2016 Confluent Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# Override this section from the script to include the com.sun.management.jmxremote.rmi.port property.
-if [ -z "$KAFKA_JMX_OPTS" ]; then
-  export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
-fi
-
-# The JMX client needs to be able to connect to java.rmi.server.hostname.
-# The default for bridged n/w is the bridged IP so you will only be able to connect from another docker container.
-# For host n/w, this is the IP that the hostname on the host resolves to.
-
-# If you have more that one n/w configured, hostname -i gives you all the IPs,
-# the default is to pick the first IP (or network).
-export KAFKA_JMX_HOSTNAME=${KAFKA_JMX_HOSTNAME:-$(hostname -i | cut -d" " -f1)}
-
-if [ "$KAFKA_JMX_PORT" ]; then
-  # This ensures that the "if" section for JMX_PORT in kafka launch script does not trigger.
-  export JMX_PORT=$KAFKA_JMX_PORT
-  export KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Djava.rmi.server.hostname=$KAFKA_JMX_HOSTNAME -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT -Dcom.sun.management.jmxremote.port=$JMX_PORT"
-fi
-
-echo "===> Launching ${COMPONENT} ... "
-exec "${COMPONENT}"-server-start /etc/"${COMPONENT}"/"${COMPONENT}".properties
diff --git a/src/main/docker/include/etc/confluent/docker/log4j.properties.template b/src/main/docker/include/etc/confluent/docker/log4j.properties.template
deleted file mode 100644
index 445a05c..0000000
--- a/src/main/docker/include/etc/confluent/docker/log4j.properties.template
+++ /dev/null
@@ -1,26 +0,0 @@
-
-log4j.rootLogger={{ env["KAFKA_LOG4J_ROOT_LOGLEVEL"] | default('INFO') }}, stdout
-
-log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
-
-{% set loggers = {
-  'kafka': 'INFO',
-  'kafka.network.RequestChannel$': 'WARN',
-  'kafka.producer.async.DefaultEventHandler': 'DEBUG',
-  'kafka.request.logger': 'WARN',
-  'kafka.controller': 'TRACE',
-  'kafka.log.LogCleaner': 'INFO',
-  'state.change.logger': 'TRACE',
-  'kafka.authorizer.logger': 'WARN'
-  } -%}
-
-
-{% if env['KAFKA_LOG4J_LOGGERS'] %}
-{% set loggers = parse_log4j_loggers(env['KAFKA_LOG4J_LOGGERS'], loggers) %}
-{% endif %}
-
-{% for logger,loglevel in loggers.items() %}
-log4j.logger.{{logger}}={{loglevel}}
-{% endfor %}
diff --git a/src/main/docker/include/etc/confluent/docker/run b/src/main/docker/include/etc/confluent/docker/run
deleted file mode 100644
index 91ac16b..0000000
--- a/src/main/docker/include/etc/confluent/docker/run
+++ /dev/null
@@ -1,41 +0,0 @@
-#!/usr/bin/env bash
-#
-# Copyright 2016 Confluent Inc.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-. /etc/confluent/docker/bash-config
-
-# Set environment values if they exist as arguments
-if [ $# -ne 0 ]; then
-  echo "===> Overriding env params with args ..."
-  for var in "$@"
-  do
-    export "$var"
-  done
-fi
-
-echo "===> ENV Variables ..."
-env
-
-echo "===> User"
-id
-
-echo "===> Configuring ..."
-/etc/confluent/docker/configure
-
-echo "===> Running preflight checks ... "
-/etc/confluent/docker/ensure
-
-echo "===> Launching ... "
-exec /etc/confluent/docker/launch
diff --git a/src/main/docker/include/etc/confluent/docker/tools-log4j.properties.template b/src/main/docker/include/etc/confluent/docker/tools-log4j.properties.template
deleted file mode 100644
index da06f13..0000000
--- a/src/main/docker/include/etc/confluent/docker/tools-log4j.properties.template
+++ /dev/null
@@ -1,7 +0,0 @@
-
-log4j.rootLogger={{ env["KAFKA_TOOLS_LOG4J_LOGLEVEL"] | default('WARN') }}, stderr
-
-log4j.appender.stderr=org.apache.log4j.ConsoleAppender
-log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
-log4j.appender.stderr.layout.ConversionPattern=[%d] %p %m (%c)%n
-log4j.appender.stderr.Target=System.err
diff --git a/src/main/docker/org.onap.dmaap.mr.keyfile b/src/main/docker/org.onap.dmaap.mr.keyfile
deleted file mode 100644
index 72d3b73..0000000
--- a/src/main/docker/org.onap.dmaap.mr.keyfile
+++ /dev/null
@@ -1,27 +0,0 @@
-yJhToh1HtF7641JOeljPtn4ECRn7dncPKtUh9XN4Hv1GX2q1MSVsDI2qQ7i2Q8hH1G3Ko_x0fl1p
-PLn2bOh9cOOGKxQrWxY0724Cme1MMc_drOl7TNk5DPmiw-teI6BdpS_wPtfDGLql9xuxIMWPdv_P
-Id9LSzdW_Fa4FepgcxAj6jOK7jQBmJIsedxIpAVFU0bjmMwybe_BRe1x8yEBrfQo8Si0cfjYdQYP
-XBTAnJ46zejAPJh2U4MyBhYzz2Zr1nMux2wjHc52z8J7_YpfveNSpr9UwOzSo0VqAEORISQDS7Cb
-Cc9jeYmxPkVCEraHWq5jtOpVdwxwTSh1PJ8_pgnhQ4AgQS-5JyRdHWvzwGa2RW8i3ZF1qfJBP4wb
-lyXiNYKSU5jDd-wNP0b9WDILGFRKuAHjz1wKB1IHyQTBX7dpjouuZysEkZS348NVLfAmb1klKp5Q
-1lq2H4TdQnPaG6tV_wyI0ZrZsf4TCeDxYRxEAZc3ILefM_72Zc-UWEHQ_Q4Qck30fJzoHFgEm5Rm
-8XofzAfHOcjoa7o8mEVi9veNpqTeNa8b2DLqYehcE_rMYU_y1AgWsnWgiIX1AEzLyUyfliS2PxQh
-ZI3HLMrzfV--pOuQp_CnrcHclvkX3u5ZJ01a6jq7ONpr712YNmUEoD6s1UR0wEEeO61Cun8zfty8
-m-qXD3k-re64WDizw-pHXHYmIS9a4jL7D0R4KysJRf6iZTAUy4ROy6aS-wMYGgy0r7sEOZY6zp4h
-MBczN5-3O7r-dKjOrr1RWXS3zt71oJNSGcnG41KKOnUeNpFBmIzVfoIp9KR5zdcg-gGrA8Xi_tBj
--rAqYfaNMBUDDtFCLhCHwuhKpR0Y0gG_-8J43yi3bfWOYMWkgPRHrJIiIO-SNUW_pobbRqQzIT3h
-V5IcEc5pa-lnJnBNS5b_-cHsA0x4g1FPmDmjsGBYZmiaPSvXPzhpZVsww1p9nVL63rYLMOTSps1V
-Lj9_UYWef7jD409vZRqeLQGi7lMNeC7oLv_hi_l05rZUkpegMLYmIWzjhzW8OuYdiHGuboV9TyUR
-QWn5e_bypQBAJhYqaYNADzV9duW0v3SW89zP9TlkJ6tfvhcVXRQn4MUzIC9h_0q3cf_9Wemx7UPW
-tGML52lYWcEbVbGF9yFtOm3qVPi4sVFO61vOtIoQhIIL3oa9gOWO9yPCjNm4lD-C4jAWHjtJdMr8
-cGtBplAHGly63VQs9RRUr42A95JMtsuJIPNDWP5-I-9R77NALdzjKEJE2FngGW8Ud4pJQ1sikPH3
-F4kVn1L2NpAilVrjlyb_y89mskrWaSdHCE2P1_gtkWHXfXIfKK0UFQt17s8hk0MfL6JSUDUE4IKN
-tK70iHwmqY9VbYKYLf-8Gl7CW_Q9MumHjGsktwVZinpH4kOUREr6tyve4rZv8wN6mbNPVJ5gw_PE
-I4bvSiEstMgelbkheMC4l-zc3q9C_fNZmLmdav8PLUrkS_NxnZ4hJQCDTjhbMqLIwknXU9SkDyPb
-Dgh049PyJrYzv2_TpYoS6M_o3HjApMkRKlV_TEcbGoX06gAUYTiEWAQU6wm0TdsIdxjEXAWeTiX7
-ddI_vEioFemoKjE5iRWNaKL85xsTsQj6bQi1eSj1F0lxqnSGRldiMAPMrfqKDJ7xFpXS7nyQfLjY
-m1H-Y3bk0iBBZbU0JKXerE_jlr3s7rcdarpwY1pdODoUJBk-EiKezm6zWuG9o3IisPNSqqOs4Cax
-QAE3dt-1TpCxkw7Rpgm8eTwPMPOD3gj7Szcs2sEh-0UIk8y7uZCSRz0ZCsQj-jJl97WQV1ky89xS
-c9ECqzDTgl2cVrih9aQu863_yHnjm9tNTxMH4DudB5JcmM96BX4CfS9qgVzAqCGvW9KS37wy0bK_
-iSCAhAWNT5L9E3fUyg--V_gmVjxGb8Y020cc4_pkqSbAAC8qjQhDWHLy_M2RzQrPmQMdP2PZ5-AU
-Pw6HdHmVTOLZeYuVS1rXx4AYWXkgKHiSRqO6bal1opzOnSpbw-Q1bQu0wZ1MarXodEtJFaOr
\ No newline at end of file
diff --git a/src/main/docker/org.onap.dmaap.mr.p12 b/src/main/docker/org.onap.dmaap.mr.p12
deleted file mode 100644
index 1a0e8a4..0000000
--- a/src/main/docker/org.onap.dmaap.mr.p12
+++ /dev/null
diff --git a/src/main/docker/org.onap.dmaap.mr.trust.jks b/src/main/docker/org.onap.dmaap.mr.trust.jks
deleted file mode 100644
index aae6d81..0000000
--- a/src/main/docker/org.onap.dmaap.mr.trust.jks
+++ /dev/null
diff --git a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProvider.java b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProvider.java
deleted file mode 100644
index 551cf81..0000000
--- a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProvider.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.commonauth.kafka.base.authorization;
-
-import java.io.IOException;
-
-public interface AuthorizationProvider {
-
-	boolean hasPermission(String userId, String permission, String instance, String action);
-
-	String getId();
-
-	String authenticate(String userId, String password) throws IOException;
-}
diff --git a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactory.java b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactory.java
deleted file mode 100644
index bdced2d..0000000
--- a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/AuthorizationProviderFactory.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.commonauth.kafka.base.authorization;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.ServiceLoader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class AuthorizationProviderFactory {
-	private static final Logger logger = LoggerFactory.getLogger(AuthorizationProviderFactory.class);
-	private static final Map<String, AuthorizationProvider> AUTHORIZATION_PROVIDER_MAP = new HashMap<>();
-	private static final AuthorizationProviderFactory AUTHORIZATION_PROVIDER_FACTORY = new AuthorizationProviderFactory();
-
-	private AuthorizationProviderFactory() {
-		try {
-			ServiceLoader<AuthorizationProvider> serviceLoader = ServiceLoader.load(AuthorizationProvider.class);
-			for (AuthorizationProvider authzProvider : serviceLoader) {
-				AUTHORIZATION_PROVIDER_MAP.put(authzProvider.getId(), authzProvider);
-
-			}
-		} catch (Exception ee) {
-			logger.error(ee.getMessage(), ee);
-			System.exit(0);
-		}
-	}
-
-	public static AuthorizationProviderFactory getProviderFactory() {
-		return AUTHORIZATION_PROVIDER_FACTORY;
-	}
-
-	public AuthorizationProvider getProvider() {
-		return AUTHORIZATION_PROVIDER_MAP.get(System.getProperty("kafka.authorization.provider", "CADI_AAF_PROVIDER"));
-	}
-}
diff --git a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProvider.java b/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProvider.java
deleted file mode 100644
index 92e27b7..0000000
--- a/src/main/java/org/onap/dmaap/commonauth/kafka/base/authorization/Cadi3AAFProvider.java
+++ /dev/null
@@ -1,205 +0,0 @@
-/*******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.commonauth.kafka.base.authorization;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.Map;
-import java.util.Properties;
-import javax.security.auth.login.AppConfigurationEntry;
-import javax.security.auth.login.Configuration;
-import org.onap.aaf.cadi.PropAccess;
-import org.onap.aaf.cadi.aaf.AAFPermission;
-import org.onap.aaf.cadi.aaf.v2_0.AAFAuthn;
-import org.onap.aaf.cadi.aaf.v2_0.AAFCon;
-import org.onap.aaf.cadi.aaf.v2_0.AAFConHttp;
-import org.onap.aaf.cadi.aaf.v2_0.AbsAAFLur;
-import org.onap.aaf.cadi.principal.UnAuthPrincipal;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class Cadi3AAFProvider implements AuthorizationProvider {
-
-	private static PropAccess access;
-	private static AAFCon<?> aafcon;
-	private static final String CADI_PROPERTIES = "/etc/kafka/data/cadi.properties";
-	private static final String AAF_LOCATOR_ENV = "aaf_locate_url";
-	private static String apiKey = null;
-	private static String kafkaUsername = null;
-	private static AAFAuthn<?> aafAuthn;
-	private static AbsAAFLur<AAFPermission> aafLur;
-	private static boolean enableCadi = false;
-	private static final String ENABLE_CADI = "enableCadi";
-	private static final Logger logger = LoggerFactory.getLogger(Cadi3AAFProvider.class);
-
-	static {
-		if (System.getProperty(ENABLE_CADI) != null) {
-			if (System.getProperty(ENABLE_CADI).equals("true")) {
-				enableCadi = true;
-			}
-		}
-         else{
-		if (System.getenv(ENABLE_CADI) != null && System.getenv(ENABLE_CADI).equals("true")) {
-			enableCadi = true;
-		}
-         }
-		Configuration config = Configuration.getConfiguration();
-		try {
-			if (config == null) {
-				logger.error("CRITICAL ERROR|Check java.security.auth.login.config VM argument|");
-			} else {
-				// read the section for KafkaServer
-				AppConfigurationEntry[] entries = config.getAppConfigurationEntry("KafkaServer");
-				if (entries == null) {
-					logger.error(
-							"CRITICAL ERROR|Check config contents passed in java.security.auth.login.config VM argument|");
-					kafkaUsername = "kafkaUsername";
-					apiKey = "apiKey";
-
-				} else {
-					for (AppConfigurationEntry entry : entries) {
-						Map<String, ?> optionsMap = entry.getOptions();
-						kafkaUsername = (String) optionsMap.get("username");
-						apiKey = (String) optionsMap.get("password");
-					}
-				}
-			}
-		} catch (Exception e) {
-			logger.error("CRITICAL ERROR: JAAS configuration incorrectly set: {}", e.getMessage());
-		}
-	}
-
-	public static String getKafkaUsername() {
-		return kafkaUsername;
-	}
-
-	public static boolean isCadiEnabled() {
-
-		return enableCadi;
-	}
-
-	public Cadi3AAFProvider() {
-		setup();
-	}
-
-	private synchronized void setup() {
-		if (access == null) {
-
-			Properties props = new Properties();
-			FileInputStream fis;
-			try {
-				if (System.getProperty("CADI_PROPERTIES") != null) {
-					fis = new FileInputStream(System.getProperty("CADI_PROPERTIES"));
-				} else {
-					fis = new FileInputStream(CADI_PROPERTIES);
-				}
-				try {
-					props.load(fis);
-					if (System.getenv(AAF_LOCATOR_ENV) != null)
-						props.setProperty(AAF_LOCATOR_ENV, System.getenv(AAF_LOCATOR_ENV));
-					access = new PropAccess(props);
-				} finally {
-					fis.close();
-				}
-			} catch (IOException e) {
-				logger.error("Unable to load " + CADI_PROPERTIES);
-				logger.error("Error", e);
-			}
-		}
-
-		if (aafAuthn == null) {
-			try {
-				aafcon = new AAFConHttp(access);
-				aafAuthn = aafcon.newAuthn();
-				aafLur = aafcon.newLur(aafAuthn);
-			} catch (final Exception e) {
-				aafAuthn = null;
-				if (access != null)
-					access.log(e, "Failed to initialize AAF");
-			}
-		}
-
-	}
-
-	/**
-	 * Checks if a user has a particular permission
-	 * <p/>
-	 * Returns true if the permission in found
-	 */
-	public boolean hasPermission(String userId, String permission, String instance, String action) {
-		boolean hasPermission = false;
-		try {
-			logger.info("^ Event at hasPermission to validate userid {} with {} {} {}", userId, permission, instance, action);
-			// AAF Style permissions are in the form
-			// Resource Name, Resource Type, Action
-			if (userId.equals("admin")) {
-				hasPermission = true;
-				return hasPermission;
-			}
-			AAFPermission perm = new AAFPermission(null, permission, instance, action);
-			if (aafLur != null) {
-				hasPermission = aafLur.fish(new UnAuthPrincipal(userId), perm);
-				logger.trace("Permission: {}  for user : {}  found: {}" , perm.getKey(), userId, hasPermission);
-			} else {
-				logger.error("AAF client not initialized. Not able to find permissions.");
-			}
-		} catch (Exception e) {
-			logger.error("AAF client not initialized", e);
-		}
-		return hasPermission;
-	}
-
-	public String getId() {
-		return "CADI_AAF_PROVIDER";
-	}
-
-	public String authenticate(String userId, String password) throws IOException {
-
-		logger.info("^Event received  with username {}", userId);
-
-		if (!enableCadi) {
-			return null;
-		} else {
-			if (userId.equals(kafkaUsername)) {
-				if (password.equals(apiKey)) {
-					logger.info("by passes the authentication for the admin {}", kafkaUsername);
-					return null;
-				} else {
-					String errorMessage = "Authentication failed for user " + kafkaUsername;
-					logger.error(errorMessage);
-					return errorMessage;
-				}
-
-			}
-
-			String aafResponse = aafAuthn.validate(userId, password);
-			logger.info("aafResponse = {} for {}", aafResponse, userId);
-
-			if (aafResponse != null) {
-				logger.error("Authentication failed for user {}", userId);
-			}
-			return aafResponse;
-		}
-
-	}
-
-}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java
deleted file mode 100644
index 4ad10e8..0000000
--- a/src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java
+++ /dev/null
@@ -1,233 +0,0 @@
-/*******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.kafkaAuthorize;
-
-import java.util.EnumSet;
-import java.util.Map;
-
-import org.apache.kafka.common.acl.AclOperation;
-import org.apache.kafka.common.security.auth.KafkaPrincipal;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
-import org.onap.dmaap.commonauth.kafka.base.authorization.Cadi3AAFProvider;
-
-import kafka.network.RequestChannel.Session;
-import kafka.security.auth.Acl;
-import kafka.security.auth.Authorizer;
-import kafka.security.auth.Operation;
-import kafka.security.auth.Resource;
-import scala.collection.immutable.Set;
-
-/**
- * A trivial Kafka Authorizer for use with SSL and AAF
- * Authentication/Authorization.
- * 
- */
-public class KafkaCustomAuthorizer implements Authorizer {
-
-	private final String[] adminPermission = new String[3];
-	protected static final EnumSet<AclOperation> TOPIC_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.DESCRIBE_CONFIGS);
-	protected static final EnumSet<AclOperation> TOPIC_READ_WRITE_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.WRITE,
-			AclOperation.READ, AclOperation.DESCRIBE_CONFIGS);
-	protected static final EnumSet<AclOperation> TOPIC_ADMIN_OPERATIONS = EnumSet.of(AclOperation.ALTER,
-			AclOperation.ALTER_CONFIGS, AclOperation.CREATE);
-	static final String TOPIC = "Topic";
-
-	private static final Logger logger = LoggerFactory.getLogger(KafkaCustomAuthorizer.class);
-
-	@Override
-	public void configure(final Map<String, ?> arg0) {
-		// TODO Auto-generate method stub
-	}
-
-	@Override
-	public void addAcls(final Set<Acl> arg0, final Resource arg1) {
-		// TODO Auto-generated method stub
-
-	}
-
-	private String[] getTopicPermission(String topicName, AclOperation aclOperation) {
-
-		String namspace = topicName.substring(0, topicName.lastIndexOf("."));
-		String[] permission = new String[3];
-		if (TOPIC_READ_WRITE_DESCRIBE_OPERATIONS.contains(aclOperation)) {
-			permission[0] = namspace + ".topic";
-			String instancePart = (System.getenv("pubSubInstPart") != null) ? System.getenv("pubSubInstPart")
-					: ".topic";
-			permission[1] = instancePart + topicName;
-
-			if (aclOperation.equals(AclOperation.WRITE)) {
-				permission[2] = "pub";
-			} else if (aclOperation.equals(AclOperation.READ)) {
-				permission[2] = "sub";
-
-			} else if (TOPIC_DESCRIBE_OPERATIONS.contains(aclOperation)) {
-				permission[2] = "view";
-
-			}
-		} else if (aclOperation.equals(AclOperation.DELETE)) {
-			permission = (System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|destroy").split("\\|");
-
-		} else if (TOPIC_ADMIN_OPERATIONS.contains(aclOperation)) {
-			permission = (System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|create").split("\\|");
-		}
-
-		return permission;
-	}
-
-	private String[] getAdminPermission() {
-
-		if (adminPermission[0] == null) {
-			adminPermission[0] = System.getProperty("namespace") + ".kafka.access";
-			adminPermission[1] = "*";
-			adminPermission[2] = "*";
-		}
-
-		return adminPermission;
-	}
-
-	private String[] getPermission(AclOperation aclOperation, String resource, String topicName) {
-		String[] permission = new String[3];
-		switch (aclOperation) {
-
-		case ALTER:
-		case ALTER_CONFIGS:
-		case CREATE:
-		case DELETE:
-			if (resource.equals(TOPIC)) {
-				permission = getTopicPermission(topicName, aclOperation);
-			} else if (resource.equals("Cluster")) {
-				permission = getAdminPermission();
-			}
-			break;
-		case DESCRIBE_CONFIGS:
-		case READ:
-		case WRITE:
-			if (resource.equals(TOPIC)) {
-				permission = getTopicPermission(topicName, aclOperation);
-			}
-			break;
-		case IDEMPOTENT_WRITE:
-			if (resource.equals("Cluster")) {
-				permission = getAdminPermission();
-			}
-			break;
-		default:
-			break;
-
-		}
-		return permission;
-
-	}
-
-	@Override
-	public boolean authorize(final Session arg0, final Operation arg1, final Resource arg2) {
-		if (arg0.principal() == null) {
-			return false;
-		}
-
-		String fullName = arg0.principal().getName();
-		fullName = fullName != null ? fullName.trim() : fullName;
-		String topicName = null;
-		String[] permission;
-
-		String resource = arg2.resourceType().name();
-
-		if (resource.equals(TOPIC)) {
-			topicName = arg2.name();
-		}
-
-		if (fullName != null && fullName.equals(Cadi3AAFProvider.getKafkaUsername())) {
-			return true;
-		}
-
-		if ((!Cadi3AAFProvider.isCadiEnabled())||(null != topicName && !topicName.startsWith("org.onap"))) {
-			return true;
-		}
-
-		permission = getPermission(arg1.toJava(), resource, topicName);
-
-		if (permission[0] != null) {
-			return !checkPermissions(fullName, topicName, permission);
-		}
-		return true;
-	}
-
-	private boolean checkPermissions(String fullName, String topicName, String[] permission) {
-		try {
-
-			if (null != topicName) {
-				boolean hasResp = AuthorizationProviderFactory.getProviderFactory().getProvider()
-					.hasPermission(fullName, permission[0], permission[1], permission[2]);
-				if (hasResp) {
-					logger.info("Successful Authorization for {} on {} for {} | {} | {}", fullName, topicName,
-						permission[0], permission[1], permission[2]);
-				}
-				if (!hasResp) {
-					logger.info("{} is not allowed in {} | {} | {}", fullName, permission[0], permission[1],
-						permission[2]);
-					return true;
-				}
-			}
-		} catch (final Exception e) {
-			return true;
-		}
-		return false;
-	}
-
-	@Override
-	public void close() {
-		// TODO Auto-generated method stub
-
-	}
-
-	@Override
-	public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls() {
-		// TODO Auto-generated method stub
-		return null;
-	}
-
-	@Override
-	public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls(final KafkaPrincipal arg0) {
-		// TODO Auto-generated method stub
-		return null;
-	}
-
-	@Override
-	public boolean removeAcls(final Resource arg0) {
-		// TODO Auto-generated method stub
-		return false;
-	}
-
-	@Override
-	public boolean removeAcls(final Set<Acl> arg0, final Resource arg1) {
-		// TODO Auto-generated method stub
-		return false;
-	}
-
-	public Set<Acl> getAcls(Resource arg0) {
-		// TODO Auto-generated method stub
-		return null;
-	}
-}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainLoginModule1.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainLoginModule1.java
deleted file mode 100644
index af5aa8f..0000000
--- a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainLoginModule1.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.kafkaAuthorize;
-
-import java.util.Map;
-import javax.security.auth.Subject;
-import javax.security.auth.callback.CallbackHandler;
-import javax.security.auth.spi.LoginModule;
-
-public class PlainLoginModule1 implements LoginModule {
-
-	private static final String USERNAME_CONFIG = "username";
-	private static final String PASSWORD_CONFIG = "password";
-
-	static {
-		PlainSaslServerProvider1.initialize();
-	}
-
-	@Override
-	public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState, Map<String, ?> options) {
-		String username = (String) options.get(USERNAME_CONFIG);
-		if (username != null)
-			subject.getPublicCredentials().add(username);
-		String password = (String) options.get(PASSWORD_CONFIG);
-		if (password != null)
-			subject.getPrivateCredentials().add(password);
-
-	}
-
-	@Override
-	public boolean login() {
-		return true;
-	}
-
-	@Override
-	public boolean logout() {
-		return true;
-	}
-
-	@Override
-	public boolean commit() {
-		return true;
-	}
-
-	@Override
-	public boolean abort() {
-		return false;
-	}
-}
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1.java
deleted file mode 100644
index 7a9bede..0000000
--- a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServer1.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.kafkaAuthorize;
-
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import javax.security.auth.callback.CallbackHandler;
-import javax.security.sasl.Sasl;
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
-import javax.security.sasl.SaslServerFactory;
-import org.apache.kafka.common.errors.SaslAuthenticationException;
-import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
-
-/**
- * Simple SaslServer implementation for SASL/PLAIN. In order to make this
- * implementation fully pluggable, authentication of username/password is fully
- * contained within the server implementation.
- * <p>
- * Valid users with passwords are specified in the Jaas configuration file. Each
- * user is specified with user_<username> as key and <password> as value. This
- * is consistent with Zookeeper Digest-MD5 implementation.
- * <p>
- * To avoid storing clear passwords on disk or to integrate with external
- * authentication servers in production systems, this module can be replaced
- * with a different implementation.
- *
- */
-public class PlainSaslServer1 implements SaslServer {
-
-    public static final String PLAIN_MECHANISM = "PLAIN";
-
-    private boolean complete;
-    private String authorizationId;
-    private static final String AUTH_EXC_NOT_COMPLETE = "Authentication exchange has not completed";
-
-
-    /**
-     * @throws SaslAuthenticationException if username/password combination is invalid or if the requested
-     *         authorization id is not the same as username.
-     * <p>
-     * <b>Note:</b> This method may throw {@link SaslAuthenticationException} to provide custom error messages
-     * to clients. But care should be taken to avoid including any information in the exception message that
-     * should not be leaked to unauthenticated clients. It may be safer to throw {@link SaslException} in
-     * some cases so that a standard error message is returned to clients.
-     * </p>
-     */
-    @Override
-    public byte[] evaluateResponse(byte[] responseBytes) throws SaslAuthenticationException {
-        /*
-         * Message format (from https://tools.ietf.org/html/rfc4616):
-         *
-         * message   = [authzid] UTF8NUL authcid UTF8NUL passwd
-         * authcid   = 1*SAFE ; MUST accept up to 255 octets
-         * authzid   = 1*SAFE ; MUST accept up to 255 octets
-         * passwd    = 1*SAFE ; MUST accept up to 255 octets
-         * UTF8NUL   = %x00 ; UTF-8 encoded NUL character
-         *
-         * SAFE      = UTF1 / UTF2 / UTF3 / UTF4
-         *                ;; any UTF-8 encoded Unicode character except NUL
-         */
-        String response = new String(responseBytes, StandardCharsets.UTF_8);
-        List<String> tokens = extractTokens(response);
-        String authorizationIdFromClient = tokens.get(0);
-        String username = tokens.get(1);
-        String password = tokens.get(2);
-
-        if (username.isEmpty()) {
-            throw new SaslAuthenticationException("Authentication failed: username not specified");
-        }
-        if (password.isEmpty()) {
-            throw new SaslAuthenticationException("Authentication failed: password not specified");
-        }
-
-        String aafResponse = "Not Verified";
-		try {
-			aafResponse = AuthorizationProviderFactory.getProviderFactory().getProvider().authenticate(username,
-					password);
-		} catch (Exception ignored) {
-            throw new SaslAuthenticationException("Authentication failed: " + aafResponse + " User " + username);
-		}
-		if (null != aafResponse) {
-			throw new SaslAuthenticationException("Authentication failed: " + aafResponse + " User " + username);
-		}
-
-        if (!authorizationIdFromClient.isEmpty() && !authorizationIdFromClient.equals(username))
-            throw new SaslAuthenticationException("Authentication failed: Client requested an authorization id that is different from username");
-
-        this.authorizationId = username;
-
-        complete = true;
-        return new byte[0];
-    }
-
-    private List<String> extractTokens(String string) {
-        List<String> tokens = new ArrayList<>();
-        int startIndex = 0;
-        for (int i = 0; i < 4; ++i) {
-            int endIndex = string.indexOf("\u0000", startIndex);
-            if (endIndex == -1) {
-                tokens.add(string.substring(startIndex));
-                break;
-            }
-            tokens.add(string.substring(startIndex, endIndex));
-            startIndex = endIndex + 1;
-        }
-
-        if (tokens.size() != 3)
-            throw new SaslAuthenticationException("Invalid SASL/PLAIN response: expected 3 tokens, got " +
-                tokens.size());
-
-        return tokens;
-    }
-
-    @Override
-    public String getAuthorizationID() {
-        if (!complete)
-            throw new IllegalStateException(AUTH_EXC_NOT_COMPLETE);
-        return authorizationId;
-    }
-
-    @Override
-    public String getMechanismName() {
-        return PLAIN_MECHANISM;
-    }
-
-    @Override
-    public Object getNegotiatedProperty(String propName) {
-        if (!complete)
-            throw new IllegalStateException(AUTH_EXC_NOT_COMPLETE);
-        return null;
-    }
-
-    @Override
-    public boolean isComplete() {
-        return complete;
-    }
-
-    @Override
-    public byte[] unwrap(byte[] incoming, int offset, int len) {
-        if (!complete)
-            throw new IllegalStateException(AUTH_EXC_NOT_COMPLETE);
-        return Arrays.copyOfRange(incoming, offset, offset + len);
-    }
-
-    @Override
-    public byte[] wrap(byte[] outgoing, int offset, int len) {
-        if (!complete)
-            throw new IllegalStateException(AUTH_EXC_NOT_COMPLETE);
-        return Arrays.copyOfRange(outgoing, offset, offset + len);
-    }
-
-    @Override
-    public void dispose() {
-        // TODO Auto-generate method stub
-    }
-
-    public static class PlainSaslServerFactory1 implements SaslServerFactory {
-
-        @Override
-        public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map<String, ?> props, CallbackHandler cbh)
-            throws SaslException {
-
-            if (!PLAIN_MECHANISM.equals(mechanism))
-                throw new SaslException(String.format("Mechanism '%s' is not supported. Only PLAIN is supported.", mechanism));
-
-            return new PlainSaslServer1();
-        }
-
-        @Override
-        public String[] getMechanismNames(Map<String, ?> props) {
-            if (props == null) return new String[]{PLAIN_MECHANISM};
-            String noPlainText = (String) props.get(Sasl.POLICY_NOPLAINTEXT);
-            if ("true".equals(noPlainText))
-                return new String[]{};
-            else
-                return new String[]{PLAIN_MECHANISM};
-        }
-    }
-}
-
diff --git a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServerProvider1.java b/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServerProvider1.java
deleted file mode 100644
index 37b408e..0000000
--- a/src/main/java/org/onap/dmaap/kafkaAuthorize/PlainSaslServerProvider1.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*******************************************************************************
- *  ============LICENSE_START=======================================================
- *  org.onap.dmaap
- *  ================================================================================
- *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- *  Modification copyright (C) 2021 Nordix Foundation.
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *        http://www.apache.org/licenses/LICENSE-2.0
-*  
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *  ============LICENSE_END=========================================================
- *  
- *  
- *******************************************************************************/
-package org.onap.dmaap.kafkaAuthorize;
-
-import java.security.Provider;
-import java.security.Security;
-
-import org.onap.dmaap.kafkaAuthorize.PlainSaslServer1.PlainSaslServerFactory1;
-
-public class PlainSaslServerProvider1 extends Provider {
-
-	private static final long serialVersionUID = 1L;
-
-	protected PlainSaslServerProvider1() {
-		super("Simple SASL/PLAIN Server Provider", 1.0, "Simple SASL/PLAIN Server Provider for Kafka");
-		super.put("SaslServerFactory." + PlainSaslServer1.PLAIN_MECHANISM, PlainSaslServerFactory1.class.getName());
-	}
-
-	public static void initialize() {
-		Security.insertProviderAt(new PlainSaslServerProvider1(),1);
-	}
-}
-
diff --git a/src/main/resources/META-INF/services/org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider b/src/main/resources/META-INF/services/org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider
deleted file mode 100644
index 0388ce7..0000000
--- a/src/main/resources/META-INF/services/org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProvider
+++ /dev/null
@@ -1 +0,0 @@
-org.onap.dmaap.commonauth.kafka.base.authorization.Cadi3AAFProvider
\ No newline at end of file