ableasdale

import subprocess
from threading import current_thread
from threading import get_ident
from threading import get_native_id
from concurrent.futures import ThreadPoolExecutor

def process_topic(filepath):
    thread = current_thread()
    #print(f'Worker thread: name={thread.name}, ident={get_ident()}, id={get_native_id()}')
    result = subprocess.run(["docker", "exec", "-t", "broker1", "/bin/bash", "-c", f'kafka-topics --bootstrap-server broker1:9092 --topic topic-{filepath} --replication-factor 3 --partitions 3 --create --config min.insync.replicas=2'], text=True)

ableasdale

Kafka and mTLS Walkthrough

Creating a Certificate Authority (CA)

Build the Certificate

openssl req -new -newkey rsa:4096 -days 365 -x509 -subj "/CN=Kafka-Security-CA" -keyout ca-key -out ca-cert -nodes

ableasdale

all:
  vars:
    ksql_service_environment_overrides:
      KSQL_HEAP_OPTS: "-Xmx7g"
    kafka_broker_service_environment_overrides:
      KAFKA_HEAP_OPTS: "-Xmx5g"
    ansible_connection: ssh
    ansible_user: centos
    ansible_become: true
    # ansible_ssh_private_key_file: /home/centos/.ssh/key.pem

ableasdale

xquery version "1.0-ml";

xdmp:save("/Users/ableasdale/javadoc28.html",
element html { attribute lang {"en"},
element head {
  element meta {attribute charset {"utf-8"}},
  element meta {attribute name {"viewport"}, attribute content {"width=device-width, initial-scale=1"}},
  element link {
    attribute href {"https://cdn.jsdelivr.net/npm/bootstrap@5.0.1/dist/css/bootstrap.min.css"},
    attribute rel {"stylesheet"},

ableasdale

xquery version "1.0-ml";

xdmp:save("/Users/ableasdale/javadoc.html",
element html { attribute lang {"en"},
element head {
  element meta {attribute charset {"utf-8"}},
  element meta {attribute name {"viewport"}, attribute content {"width=device-width, initial-scale=1"}},
  element link {
    attribute href {"https://cdn.jsdelivr.net/npm/bootstrap@5.0.1/dist/css/bootstrap.min.css"},
    attribute rel {"stylesheet"},

ableasdale

import com.marklogic.client.DatabaseClient;
import com.marklogic.client.DatabaseClientFactory;
import com.marklogic.client.datamovement.DataMovementManager;
import com.marklogic.client.datamovement.QueryBatcher;
import com.marklogic.client.query.StructuredQueryBuilder;
import com.marklogic.client.query.StructuredQueryDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.invoke.MethodHandles;

ableasdale

xquery version "1.0-ml";

declare variable $TARGET-FILENAME as xs:string := "/tmp/"||xdmp:random()||".zip";
declare variable $URIS as xs:string* := cts:uris((), ("limit=100"));

declare function local:write-zipfile() {
  let $zip := xdmp:zip-create(
    <parts xmlns="xdmp:zip">{$URIS ! element part {.}}</parts>,
    ($URIS ! doc(.))
  )

ableasdale

xquery version "1.0-ml";
(: 

  A Utility module to allow the deletion of a directory fragment in a given forest in the event of 
  an XDMP-DBDUPURI exception.  
  
:)

let $doc := "/" (: URI for the conflicting directory that is causing an XDMP-DBDUPURI to be reported in 2 forests :)
let $forest-name := "forest_00" (: The name of the forest from which the problem directory fragment should be removed :)

ableasdale

xquery version "1.0-ml";
(:

  A Utility module to allow the detection of duplicate URIs for a given database
  This can be used in any situation where XDMP-DBDUPURI messages are found in the ErrorLogs

  URIs are sorted by frequency order so higher frequencies (e.g. > 1) indicate that the URI 
  has been found in more than one forest

  This module requires the uri lexicon to be enabled on your database

ableasdale

xquery version "1.0-ml";
(: 

  A Utility module to allow the deletion of a fragment in a given forest in the event of 
  an XDMP-DBDUPURI exception.  
  
  Given the following exception:
  XDMP-DBDUPURI: URI /problem/uri.xml found in forests Library06 and Library07
  
  $doc would be set to "/problem/uri.xml"