Примеры использования ADB to Kafka Connector

$ 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