Month: November 2024

Kafka Metadata Mismatch

I’ll prefix this with a caveat — when Zookeeper and the Kafka metadata properties topic ID values don’t match up … there’s something wrong beyond “hey, make this string the same as that string and try again”. Look for communication issues, disk issues, etc that might lead to bad data. And, lacking any cause, the general recommendation is to drop and recreate the topic.

But! “Hey, I am going to delete all the data in this Kafka topic. Everyone good with that?” is seldom answered with a resounding “of course, we don’t actually want to keep the data we specifically configured a system to keep”. And you need to keep the topic around even though something has clearly gone sideways. Fortunately, you can manually update either the zookeeper topic ID or the one on disk. I find it easier to update the one on disk in Kafka because this change can be reverted. Stop all of the Kafka servers. The following script runs on each Kafka server – identifies all of the folders within the Kafka data for the partition and changes the wrong topic ID to the right one. Then start the Kafka servers again, and the partition should be functional. For now.

#!/bin/bash

# Function to check if Kafka is stopped
check_kafka_stopped() {
    if systemctl is-active --quiet kafka; then
        echo "Error: Kafka service is still running. Please stop Kafka before proceeding."
        exit 1
    else
        echo "Kafka service is stopped. Proceeding with backup and update."
    fi
}

# Define the base directory where Kafka stores its data
DATA_DIR="/path/todata-kafka"

# Define the topic prefix to search for
TOPIC_PREFIX="MY_TOPIC_NAME-"

# Old and new topic IDs
OLD_TOPIC_ID="2xSmlPMBRv2_ihuWgrvQNA"
NEW_TOPIC_ID="57kBnenRTo21_VhEhWFOWg"

# Date string for backup file naming
DATE_STR=$(date +%Y%m%d)

# Check if Kafka is stopped
check_kafka_stopped

# Find all directories matching the topic pattern and process each one
for dir in $DATA_DIR/${TOPIC_PREFIX}*; do
    # Construct the path to the partition.metadata file
    metadata_file="$dir/partition.metadata"

    # Check if the metadata file exists
    if [[ -f "$metadata_file" ]]; then
        # Backup the existing partition.metadata file
        backup_file="$dir/partmetadata.$DATE_STR"
        echo "Backing up $metadata_file to $backup_file"
        cp "$metadata_file" "$backup_file"

        # Use sed to replace the line containing the old topic_id with the new one
        echo "Updating topic_id in $metadata_file"
        sed -i "s/^topic_id: $OLD_TOPIC_ID/topic_id: $NEW_TOPIC_ID/" "$metadata_file"
    else
        echo "No partition.metadata file found in $dir"
    fi
done

echo "Backup and topic ID update complete."

How do you get the Kafka metadata and Zookeeper topic IDs? In Zookeeper, ask it:

kafkaserver::fixTopicID # bin/zookeeper-shell.sh $(hostname):2181
get /brokers/topics/MY_TOPIC_NAME
{"removing_replicas":{},"partitions":{"2":[251,250],"5":[249,250],"27":[248,247],"12":[248,251],"8":[247,248],"15":[251,250],"21":[247,250],"18":[249,248],"24":[250,248],"7":[251,247],"1":[250,249],"17":[248,247],"23":[249,247],"26":[247,251],"4":[248,247],"11":[247,250],"14":[250,248],"20":[251,249],"29":[250,249],"6":[250,251],"28":[249,248],"9":[248,249],"0":[249,248],"22":[248,251],"16":[247,251],"19":[250,249],"3":[247,251],"10":[251,249],"25":[251,250],"13":[249,247]},"topic_id":"57kBnenRTo21_VhEhWFOWg","adding_replicas":{},"version":3}

Kafka’s is stored on disk in each folder for a topic partition:

kafkaserver::fixTopicID # cat /path/to/data-kafka/MY_TOPIC_NAME-29/partition.metadata
version: 0
topic_id: 2xSmlPMBRv2_ihuWgrvQNA

Once the data is processed, it would be a good idea to recreate the topic … but ensuring the two topic ID values match up will get you back online.

Ohio Voter Turnout

The current data from https://liveresults.ohiosos.gov/ shows the five biggest counties in Ohio are all in the top ten “bad” voter turnout counties. Ohio’s unofficial turnout is 69.69%. At least so far.

County Registered Voters Ballots Counted Unofficial Voter Turnout Outstanding Absentees Outstanding Provisionals Total Precincts Election Day Precincts Reporting Election Day % Precincts Reporting Metro NonVoter Count
Greene 122,193 49,991 40.91 1,267 2,044 146 146 100 72,204
Lucas 304,907 187,235 61.41 2,856 4,426 303 303 100 Toledo 117,664
Lawrence 43,020 26,586 61.8 239 423 84 84 100 16,434
Cuyahoga 893,801 571,397 63.93 15,697 14,183 967 967 100 Cleveland 322,394
Athens 38,592 24,956 64.67 265 1,537 56 56 100 13,635
Franklin 903,493 592,418 65.57 9,235 19,563 888 888 100 Columbus 311,073
Pike 18,010 12,072 67.03 113 314 22 22 100 5,938
Hamilton 604,178 405,825 67.17 6,360 12,659 562 562 100 Cincinnati 198,352
Montgomery 373,582 251,763 67.39 4,909 6,200 381 381 100 Dayton 121,825
Scioto 45,434 30,666 67.5 247 1,041 77 77 100 14,766

ElasticSearch 7.7.0 with Log4j 2.24.1

Interestingly, I am still able to manually upgrade the log4j components of ElasticSearch 7.7.0 to keep off the “naughty list” of vulnerabilities at work. I am successfully using 2.24.1 with ElasticSearch 7.7.0, OpenDistro, and the S3 backup plug-in.

Same script as before — just downloaded newer JARs, placed them into the /tmp folder on each server, and set the l4jver variable to the current release.

export l4jver=2.24.1
systemctl stop elasticsearch 

# Remove old jars
rm  --interactive=never /opt/elk/elasticsearch/lib/log4j-api-*.jar
rm  --interactive=never /opt/elk/elasticsearch/lib/log4j-core-*.jar
rm  --interactive=never /opt/elk/elasticsearch/modules/x-pack-identity-provider/log4j-slf4j-impl-*.jar
rm  --interactive=never /opt/elk/elasticsearch/modules/x-pack-security/log4j-slf4j-impl-*.jar
rm  --interactive=never /opt/elk/elasticsearch/plugins/opendistro_security/log4j-slf4j-impl-*.jar
rm  --interactive=never /opt/elk/elasticsearch/modules/x-pack-core/log4j-1.2-api-*.jar 
rm  --interactive=never /opt/elk/elasticsearch/plugins/repository-s3/log4j-1.2-api-*.jar 

# Copy in upgraded jars
cp /tmp/log4j-api-$l4jver*.jar /opt/elk/elasticsearch/lib/
cp /tmp/log4j-core-$l4jver*.jar /opt/elk/elasticsearch/lib/
cp /tmp/log4j-slf4j-impl-$l4jver*.jar /opt/elk/elasticsearch/modules/x-pack-identity-provider/
cp /tmp/log4j-slf4j-impl-$l4jver*.jar /opt/elk/elasticsearch/modules/x-pack-security/
cp /tmp/log4j-slf4j-impl-$l4jver*.jar /opt/elk/elasticsearch/plugins/opendistro_security/
cp /tmp/log4j-1.2-api-$l4jver*.jar /opt/elk/elasticsearch/modules/x-pack-core/
cp /tmp/log4j-1.2-api-$l4jver*.jar /opt/elk/elasticsearch/plugins/repository-s3/

# Fix permissions
chown elkadmin:elkadmin /opt/elk/elasticsearch/lib/log4j*
chown elkadmin:elkadmin /opt/elk/elasticsearch/modules/x-pack-core/log4j*
chown elkadmin:elkadmin /opt/elk/elasticsearch/modules/x-pack-identity-provider/log4j*
chown elkadmin:elkadmin /opt/elk/elasticsearch/modules/x-pack-security/log4j*
chown elkadmin:elkadmin /opt/elk/elasticsearch/plugins/repository-s3/log4j*
chown elkadmin:elkadmin /opt/elk/elasticsearch/plugins/opendistro_security/log4j*

systemctl start elasticsearch