ADB to Kafka Connector usage examples

$ bin/kafka-topics.sh --create --topic topic_adb_to_kafka --bootstrap-server bds-ads1:9092,bds-ads2:9092,bds-ads3:9092

$ sudo su - gpadmin

$ psql adb

CREATE WRITABLE EXTERNAL TABLE ext_adb_to_kafka (a INT, b TEXT)
LOCATION (
    'pxf://topic_adb_to_kafka?PROFILE=kafka&BOOTSTRAP_SERVERS=bds-ads1:9092,bds-ads2:9092,bds-ads3:9092'
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

INSERT INTO ext_adb_to_kafka VALUES
(1,'test1'),
(2,'test2'),
(3,'test3'),
(4,'test4'),
(5,'test5');

$ bin/kafka-console-consumer.sh --topic topic_adb_to_kafka --from-beginning --bootstrap-server bds-ads1:9092,bds-ads2:9092,bds-ads3:9092 -value-deserializer org.apache.kafka.common.serialization.BytesDeserializer --property print.key=false --property print.value=true --property print.offset=true

CREATE WRITABLE EXTERNAL TABLE ext_adb_to_kafka2 (a INT, b TEXT)
LOCATION (
    'pxf://topic_adb_to_kafka?PROFILE=kafka&SERVER=kafka_test'
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

INSERT INTO ext_adb_to_kafka2 VALUES
(1,'test1');

$ bin/kafka-console-consumer.sh --topic topic_adb_to_kafka --from-beginning --bootstrap-server bds-ads1:9092,bds-ads2:9092,bds-ads3:9092 -value-deserializer org.apache.kafka.common.serialization.BytesDeserializer --property print.key=false --property print.value=true --property print.offset=true

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
# Uncomment and set necessary username/password
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="adb-to-kafka" \
  password="123";

$ sudo su - gpadmin

$ psql adb

CREATE WRITABLE EXTERNAL TABLE ext_adb_to_kafka_sasl (a INT, b TEXT)
LOCATION (
    'pxf://topic_adb_to_kafka_sasl?PROFILE=kafka&BOOTSTRAP_SERVERS=bds-ads1:9092,bds-ads2:9092,bds-ads3:9092&SECURITY_PROTOCOL=SASL_PLAINTEXT&SASL_MECHANISM=PLAIN&SASL_USER=adb-to-kafka&SASL_USER_PASSWORD=123'
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

INSERT INTO ext_adb_to_kafka_sasl VALUES
(1,'test1');

$ bin/kafka-console-consumer.sh --topic topic_adb_to_kafka_sasl --from-beginning --bootstrap-server bds-ads1:9092,bds-ads2:9092,bds-ads3:9092 -value-deserializer org.apache.kafka.common.serialization.BytesDeserializer --property print.key=false --property print.value=true --property print.offset=true --consumer.config /etc/kafka/conf/client.properties

$ sudo kadmin.local -q "add_principal -pw 123 adb-to-kafka@ADS-KAFKA.LOCAL"

$ sudo yum install -y krb5-libs krb5-server krb5-workstation openldap-clients cyrus-sasl-gssapi

$ sudo vi /etc/krb5.conf

[logging]
 default = FILE:/var/log/krb5libs.log
 kdc = FILE:/var/log/krb5kdc.log
 admin_server = FILE:/var/log/kadmind.log

[libdefaults]
 default_realm = ADS-KAFKA.LOCAL
 kdc_timesync = 1
 ticket_lifetime = 24h

[realms]
 ADS-KAFKA.LOCAL = {
 admin_server = bds-mit-kerberos.ru-central1.internal
 kdc = bds-mit-kerberos.ru-central1.internal
 }

$ sudo vi /var/kerberos/krb5kdc/kdc.conf

[kdcdefaults]
 kdc_ports = 88
 kdc_tcp_ports = 88
 default_realm = ADS-KAFKA.LOCAL

[realms]
 ADS-KAFKA.LOCAL = {
  #master_key_type = aes256-cts
  acl_file = /var/kerberos/krb5kdc/kadm5.acl
  dict_file = /usr/share/dict/words
  admin_keytab = /var/kerberos/krb5kdc/kadm5.keytab
  supported_enctypes = aes256-cts:normal aes128-cts:normal des3-hmac-sha1:normal arcfour-hmac:normal camellia256-cts:normal camellia128-cts:normal des-hmac-sha1:normal des-cbc-md5:normal des-cbc-crc:normal
 }

$ sudo mkdir /var/lib/pxf/conf/kerberos
$ sudo chmod 744 /var/lib/pxf/conf/kerberos
$ sudo chown gpadmin:gpadmin /var/lib/pxf/conf/kerberos

$ sudo su - gpadmin

$ ktutil

add_entry -password -p adb-to-kafka@ADS-KAFKA.LOCAL -k 1 -e aes128-cts-hmac-sha1-96
add_entry -password -p adb-to-kafka@ADS-KAFKA.LOCAL -k 1 -e aes256-cts-hmac-sha1-96
add_entry -password -p adb-to-kafka@ADS-KAFKA.LOCAL -k 1 -e arcfour-hmac
add_entry -password -p adb-to-kafka@ADS-KAFKA.LOCAL -k 1 -e camellia256-cts-cmac
add_entry -password -p adb-to-kafka@ADS-KAFKA.LOCAL -k 1  -e camellia128-cts-cmac

write_kt /var/lib/pxf/conf/kerberos/adb-to-kafka.service.keytab

$ kinit adb-to-kafka@ADS-KAFKA.LOCAL

$ klist

$ psql adb

CREATE WRITABLE EXTERNAL TABLE ext_adb_to_kafka_kerberos (a INT, b TEXT)
LOCATION (
    'pxf://topic_adb_to_kafka_kerberos?PROFILE=kafka&BOOTSTRAP_SERVERS=bds-ads1:9092,bds-ads2:9092,bds-ads3:9092&SECURITY_PROTOCOL=SASL_PLAINTEXT&SASL_MECHANISM=GSSAPI&KERBEROS_KEYTAB=/var/lib/pxf/conf/kerberos/adb-to-kafka.service.keytab&KERBEROS_PRINCIPAL=adb-to-kafka&KERBEROS_SERVICE_NAME=kafka'
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

INSERT INTO ext_adb_to_kafka_kerberos VALUES
(1,'kerberos');

$ kinit adb-to-kafka@ADS-KAFKA.LOCAL
$ export KAFKA_OPTS="-Djava.security.auth.login.config=/tmp/client.jaas"
$ bin/kafka-console-consumer.sh --topic topic_adb_to_kafka_kerberos --from-beginning --bootstrap-server bds-ads1:9092,bds-ads2:9092,bds-ads3:9092 -value-deserializer org.apache.kafka.common.serialization.BytesDeserializer --property print.key=false --property print.value=true --property print.offset=true --consumer.config /tmp/client.properties

$ openssl req -new -newkey rsa:4096 -days 3650 -x509 -nodes -subj "/CN=Kafka-CA" -keyout ca.key -out ca.crt

$ openssl req -newkey rsa:4096 -nodes -subj "/CN=val-ads-1.ru-central1.internal" -addext "subjectAltName = IP:192.0.2.124, DNS:localhost, DNS:val-ads-1.ru-central1.internal" -keyout server.key -out server.csr

$ openssl x509 -req -CA ca.crt -CAkey ca.key -in server.csr -out server.crt -days 3650 -CAcreateserial -copy_extensions copyall

$ openssl pkcs12 -export -in server.crt -inkey server.key -chain -CAfile ca.crt -name kafka-server -out server.p12 -passout pass:serverpass

$ keytool -importkeystore -srckeystore server.p12 -srcstoretype PKCS12 -srcstorepass serverpass -destkeystore server.keystore.jks -deststoretype JKS -deststorepass serverpass -noprompt

$ keytool -keystore truststore.jks -alias CARoot -import -file ca.crt -storepass 123456 -noprompt

$ openssl req -newkey rsa:4096 -nodes -subj "/CN=adb-kafka-client" -keyout client.key -out client.csr
$ openssl x509 -req -CA ca.crt -CAkey ca.key -in client.csr -out client.crt -days 3650 -CAcreateserial
$ openssl pkcs12 -export -in client.crt -inkey client.key -name kafka-client -out client.p12 -passout pass:clientpass
$ keytool -importkeystore -srckeystore client.p12 -srcstoretype PKCS12 -srcstorepass clientpass -destkeystore client.keystore.jks -deststoretype JKS -deststorepass clientpass -noprompt

# The address the socket server listens on.
listeners=PLAINTEXT://:9092,CONTROLLER://:9093,SSL://:9094

# Listener name, hostname and port the broker or the controller will advertise to clients.
advertised.listeners=PLAINTEXT://192.0.2.124:9092,CONTROLLER://192.0.2.124:9093,SSL://192.0.2.124:9094

ssl.keystore.location=/home/user/kafka_certs/server.keystore.jks
ssl.keystore.password=serverpass
ssl.key.password=serverpass
ssl.truststore.location=/home/user/kafka_certs/truststore.jks
ssl.truststore.password=123456

$ bin/kafka-server-start.sh config/server.properties

CREATE WRITABLE EXTERNAL TABLE ext_kafka_ssl (a INT, b TEXT)
LOCATION (
    'pxf://test_ssl?PROFILE=kafka&BOOTSTRAP_SERVERS=val-ads-1.ru-central1.internal:9094&SECURITY_PROTOCOL=SSL&TRUSTSTORE_LOCATION=/home/gpadmin/kafka_certs/truststore.jks&TRUSTSTORE_PASSWORD=123456'
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

INSERT INTO ext_kafka_ssl VALUES
(1,'new1'),
(2,'new2'),
(3,'new3'),
(4,'new4'),
(5,'new5');

$ bin/kafka-run-class.sh org.apache.kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic test_ssl --time -1

ssl.client.auth=required
ssl.truststore.location=/home/user/kafka_certs/truststore.jks
ssl.truststore.password=123456

$ bin/kafka-server-stop.sh
$ bin/kafka-server-start.sh -daemon config/server.properties

CREATE WRITABLE EXTERNAL TABLE ext_kafka_mtls (a INT, b TEXT)
LOCATION (
    'pxf://test_mtls?PROFILE=kafka&BOOTSTRAP_SERVERS=val-ads-1.ru-central1.internal:9094&SECURITY_PROTOCOL=SSL&TRUSTSTORE_LOCATION=/home/gpadmin/kafka_certs/truststore.jks&TRUSTSTORE_PASSWORD=123456&KEYSTORE_LOCATION=/home/gpadmin/kafka_certs/client.keystore.jks&KEYSTORE_PASSWORD=clientpass'
)
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');

INSERT INTO ext_kafka_mtls VALUES
(1,'new1'),
(2,'new2'),
(3,'new3');

$ bin/kafka-run-class.sh org.apache.kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic test_mtls --time -1