feat(kafka): increase kafka message size and enable compression (#9038)

Co-authored-by: Pedro Silva <[email protected]>

.github/workflows/docker-unified.yml

@@ -851,8 +851,14 @@ jobs:
         if: failure()
         run: |
           docker ps -a
-          docker logs datahub-gms >& gms-${{ matrix.test_strategy }}.log
-          docker logs datahub-actions >& actions-${{ matrix.test_strategy }}.log
+          docker logs datahub-gms >& gms-${{ matrix.test_strategy }}.log || true
+          docker logs datahub-actions >& actions-${{ matrix.test_strategy }}.log || true
+          docker logs datahub-mae-consumer >& mae-${{ matrix.test_strategy }}.log || true
+          docker logs datahub-mce-consumer >& mce-${{ matrix.test_strategy }}.log || true
+          docker logs broker >& broker-${{ matrix.test_strategy }}.log || true
+          docker logs mysql >& mysql-${{ matrix.test_strategy }}.log || true
+          docker logs elasticsearch >& elasticsearch-${{ matrix.test_strategy }}.log || true
+          docker logs datahub-frontend-react >& frontend-${{ matrix.test_strategy }}.log || true
       - name: Upload logs
         uses: actions/upload-artifact@v3
         if: failure()

build.gradle

@@ -39,7 +39,7 @@ buildscript {
 plugins {
   id 'com.gorylenko.gradle-git-properties' version '2.4.0-rc2'
   id 'com.github.johnrengelman.shadow' version '6.1.0'
-  id 'com.palantir.docker' version '0.35.0'
+  id 'com.palantir.docker' version '0.35.0' apply false
   // https://blog.ltgt.net/javax-jakarta-mess-and-gradle-solution/
   // TODO id "org.gradlex.java-ecosystem-capabilities" version "1.0"
 }

datahub-frontend/app/client/KafkaTrackingProducer.java

@@ -1,6 +1,8 @@
 package client;
 
+import com.linkedin.metadata.config.kafka.ProducerConfiguration;
 import com.typesafe.config.Config;
+import config.ConfigurationProvider;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -35,12 +37,12 @@ public class KafkaTrackingProducer {
     private final KafkaProducer<String, String> _producer;
 
     @Inject
-    public KafkaTrackingProducer(@Nonnull Config config, ApplicationLifecycle lifecycle) {
+    public KafkaTrackingProducer(@Nonnull Config config, ApplicationLifecycle lifecycle, final ConfigurationProvider configurationProvider) {
         _isEnabled = !config.hasPath("analytics.enabled") || config.getBoolean("analytics.enabled");
 
         if (_isEnabled) {
             _logger.debug("Analytics tracking is enabled");
-            _producer = createKafkaProducer(config);
+            _producer = createKafkaProducer(config, configurationProvider.getKafka().getProducer());
 
             lifecycle.addStopHook(
                     () -> {
@@ -62,13 +64,15 @@ public void send(ProducerRecord<String, String> record) {
         _producer.send(record);
     }
 
-    private static KafkaProducer createKafkaProducer(Config config) {
+    private static KafkaProducer createKafkaProducer(Config config, ProducerConfiguration producerConfiguration) {
         final Properties props = new Properties();
         props.put(ProducerConfig.CLIENT_ID_CONFIG, "datahub-frontend");
         props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, config.getString("analytics.kafka.delivery.timeout.ms"));
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getString("analytics.kafka.bootstrap.server"));
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); // Actor urn.
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); // JSON object.
+        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, producerConfiguration.getMaxRequestSize());
+        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, producerConfiguration.getCompressionType());
 
         final String securityProtocolConfig = "analytics.kafka.security.protocol";
         if (config.hasPath(securityProtocolConfig)

datahub-frontend/app/config/ConfigurationProvider.java

@@ -1,6 +1,7 @@
 package config;
 
 import com.linkedin.metadata.config.cache.CacheConfiguration;
+import com.linkedin.metadata.config.kafka.KafkaConfiguration;
 import com.linkedin.metadata.spring.YamlPropertySourceFactory;
 import lombok.Data;
 
@@ -11,14 +12,17 @@
 
 /**
  * Minimal sharing between metadata-service and frontend
- * Initially for use of client caching configuration.
  * Does not use the factories module to avoid transitive dependencies.
  */
 @EnableConfigurationProperties
 @PropertySource(value = "application.yml", factory = YamlPropertySourceFactory.class)
 @ConfigurationProperties
 @Data
 public class ConfigurationProvider {
+    /**
+     * Kafka related configs.
+     */
+    private KafkaConfiguration kafka;
 
     /**
      * Configuration for caching

docker/broker/env/docker.env

@@ -5,4 +5,6 @@ KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9
 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
 KAFKA_HEAP_OPTS=-Xms256m -Xmx256m
-KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
+KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
+KAFKA_MESSAGE_MAX_BYTES=5242880
+KAFKA_MAX_MESSAGE_BYTES=5242880

docker/datahub-frontend/Dockerfile

@@ -8,10 +8,12 @@ RUN addgroup -S datahub && adduser -S datahub -G datahub
 # Upgrade Alpine and base packages
 # PFP-260: Upgrade Sqlite to >=3.28.0-r0 to fix https://security.snyk.io/vuln/SNYK-ALPINE39-SQLITE-449762
 RUN apk --no-cache --update-cache --available upgrade \
-    && apk --no-cache add curl sqlite \
+    && apk --no-cache add curl sqlite libc6-compat java-snappy \
     && apk --no-cache add openjdk11-jre --repository=http://dl-cdn.alpinelinux.org/alpine/edge/community \
     && apk --no-cache add jattach --repository http://dl-cdn.alpinelinux.org/alpine/edge/community/
 
+ENV LD_LIBRARY_PATH="/lib:/lib64"
+
 FROM base as prod-install
 
 COPY ./datahub-frontend.zip /

docker/datahub-gms/Dockerfile

@@ -18,7 +18,7 @@ FROM alpine:3 AS base
 ENV JMX_VERSION=0.18.0
 # PFP-260: Upgrade Sqlite to >=3.28.0-r0 to fix https://security.snyk.io/vuln/SNYK-ALPINE39-SQLITE-449762
 RUN apk --no-cache --update-cache --available upgrade \
-    && apk --no-cache add curl bash coreutils gcompat sqlite \
+    && apk --no-cache add curl bash coreutils gcompat sqlite libc6-compat java-snappy \
     && apk --no-cache add openjdk11-jre --repository=http://dl-cdn.alpinelinux.org/alpine/edge/community \
     && apk --no-cache add jattach --repository http://dl-cdn.alpinelinux.org/alpine/edge/community/ \
     && curl -sS https://repo1.maven.org/maven2/org/eclipse/jetty/jetty-runner/9.4.46.v20220331/jetty-runner-9.4.46.v20220331.jar --output jetty-runner.jar \
@@ -29,6 +29,8 @@ RUN apk --no-cache --update-cache --available upgrade \
     && cp /usr/lib/jvm/java-11-openjdk/jre/lib/security/cacerts /tmp/kafka.client.truststore.jks
 COPY --from=binary /go/bin/dockerize /usr/local/bin
 
+ENV LD_LIBRARY_PATH="/lib:/lib64"
+
 FROM base as prod-install
 COPY war.war /datahub/datahub-gms/bin/war.war
 COPY metadata-models/src/main/resources/entity-registry.yml /datahub/datahub-gms/resources/entity-registry.yml

docker/datahub-mae-consumer/Dockerfile

@@ -18,14 +18,16 @@ FROM alpine:3 AS base
 ENV JMX_VERSION=0.18.0
 # PFP-260: Upgrade Sqlite to >=3.28.0-r0 to fix https://security.snyk.io/vuln/SNYK-ALPINE39-SQLITE-449762
 RUN apk --no-cache --update-cache --available upgrade \
-    && apk --no-cache add curl bash coreutils sqlite \
+    && apk --no-cache add curl bash coreutils sqlite libc6-compat java-snappy \
     && apk --no-cache add openjdk11-jre --repository=http://dl-cdn.alpinelinux.org/alpine/edge/community \
     && apk --no-cache add jattach --repository http://dl-cdn.alpinelinux.org/alpine/edge/community/ \
     && wget --no-verbose https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/download/v1.24.0/opentelemetry-javaagent.jar \
     && wget --no-verbose https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/${JMX_VERSION}/jmx_prometheus_javaagent-${JMX_VERSION}.jar -O jmx_prometheus_javaagent.jar \
     && cp /usr/lib/jvm/java-11-openjdk/jre/lib/security/cacerts /tmp/kafka.client.truststore.jks
 COPY --from=binary /go/bin/dockerize /usr/local/bin
 
+ENV LD_LIBRARY_PATH="/lib:/lib64"
+
 FROM base as prod-install
 COPY mae-consumer-job.jar /datahub/datahub-mae-consumer/bin/
 COPY metadata-models/src/main/resources/entity-registry.yml /datahub/datahub-mae-consumer/resources/entity-registry.yml

docker/datahub-mce-consumer/Dockerfile

@@ -18,7 +18,7 @@ FROM alpine:3 AS base
 ENV JMX_VERSION=0.18.0
 # PFP-260: Upgrade Sqlite to >=3.28.0-r0 to fix https://security.snyk.io/vuln/SNYK-ALPINE39-SQLITE-449762
 RUN apk --no-cache --update-cache --available upgrade \
-    && apk --no-cache add curl bash sqlite \
+    && apk --no-cache add curl bash sqlite libc6-compat java-snappy \
     && apk --no-cache add openjdk11-jre --repository=http://dl-cdn.alpinelinux.org/alpine/edge/community \
     && apk --no-cache add jattach --repository http://dl-cdn.alpinelinux.org/alpine/edge/community/ \
     && wget --no-verbose https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/download/v1.24.0/opentelemetry-javaagent.jar \
@@ -33,6 +33,8 @@ COPY docker/datahub-mce-consumer/start.sh /datahub/datahub-mce-consumer/scripts/
 COPY docker/monitoring/client-prometheus-config.yaml /datahub/datahub-mce-consumer/scripts/prometheus-config.yaml
 RUN chmod +x /datahub/datahub-mce-consumer/scripts/start.sh
 
+ENV LD_LIBRARY_PATH="/lib:/lib64"
+
 FROM base as dev-install
 # Dummy stage for development. Assumes code is built on your machine and mounted to this image.
 # See this excellent thread https://github.com/docker/cli/issues/1134

docker/datahub-upgrade/Dockerfile

@@ -18,7 +18,7 @@ FROM alpine:3 AS base
 ENV JMX_VERSION=0.18.0
 # PFP-260: Upgrade Sqlite to >=3.28.0-r0 to fix https://security.snyk.io/vuln/SNYK-ALPINE39-SQLITE-449762
 RUN apk --no-cache --update-cache --available upgrade \
-    && apk --no-cache add curl bash coreutils gcompat sqlite \
+    && apk --no-cache add curl bash coreutils gcompat sqlite libc6-compat java-snappy \
     && apk --no-cache add openjdk11-jre --repository=http://dl-cdn.alpinelinux.org/alpine/edge/community \
     && curl -sS https://repo1.maven.org/maven2/org/eclipse/jetty/jetty-runner/9.4.46.v20220331/jetty-runner-9.4.46.v20220331.jar --output jetty-runner.jar \
     && curl -sS https://repo1.maven.org/maven2/org/eclipse/jetty/jetty-jmx/9.4.46.v20220331/jetty-jmx-9.4.46.v20220331.jar --output jetty-jmx.jar \
@@ -28,6 +28,8 @@ RUN apk --no-cache --update-cache --available upgrade \
     && cp /usr/lib/jvm/java-11-openjdk/jre/lib/security/cacerts /tmp/kafka.client.truststore.jks
 COPY --from=binary /go/bin/dockerize /usr/local/bin
 
+ENV LD_LIBRARY_PATH="/lib:/lib64"
+
 FROM base as prod-install
 COPY datahub-upgrade.jar /datahub/datahub-upgrade/bin/
 COPY metadata-models/src/main/resources/entity-registry.yml /datahub/datahub-gms/resources/entity-registry.yml

docker/kafka-setup/kafka-config.sh

@@ -2,6 +2,7 @@
 
 : ${PARTITIONS:=1}
 : ${REPLICATION_FACTOR:=1}
+: ${MAX_MESSAGE_BYTES:=5242880}
 
 : ${KAFKA_PROPERTIES_SECURITY_PROTOCOL:=PLAINTEXT}
 
@@ -12,3 +13,4 @@ export KAFKA_HEAP_OPTS="-Xmx64M"
 CONNECTION_PROPERTIES_PATH=/tmp/connection.properties
 
 WORKERS=4
+DELIMITER=";"

docker/kafka-setup/kafka-setup.sh

@@ -102,24 +102,43 @@ exec 4<&-
 send() {
   work_id=$1
   topic_args=$2
-  echo sending $work_id $topic_args
-  echo "$work_id" "$topic_args" 1>&3 ## the fifo is fd 3
+  topic_config=$3
+
+  echo -e "sending $work_id\n   worker_args: ${topic_args}${DELIMITER}${topic_config}"
+  echo "$work_id" "${topic_args}${DELIMITER}${topic_config}" 1>&3 ## the fifo is fd 3
 }
 
 ## Produce the jobs to run.
-send "$METADATA_AUDIT_EVENT_NAME" "--partitions $PARTITIONS --topic $METADATA_AUDIT_EVENT_NAME"
-send "$METADATA_CHANGE_EVENT_NAME" "--partitions $PARTITIONS --topic $METADATA_CHANGE_EVENT_NAME"
-send "$FAILED_METADATA_CHANGE_EVENT_NAME" "--partitions $PARTITIONS --topic $FAILED_METADATA_CHANGE_EVENT_NAME"
-send "$METADATA_CHANGE_LOG_VERSIONED_TOPIC_NAME" "--partitions $PARTITIONS --topic $METADATA_CHANGE_LOG_VERSIONED_TOPIC_NAME"
+send "$METADATA_AUDIT_EVENT_NAME" "--partitions $PARTITIONS --topic $METADATA_AUDIT_EVENT_NAME" \
+     "--entity-type topics --entity-name $METADATA_AUDIT_EVENT_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
+
+send "$METADATA_CHANGE_EVENT_NAME" "--partitions $PARTITIONS --topic $METADATA_CHANGE_EVENT_NAME" \
+     "--entity-type topics --entity-name $METADATA_CHANGE_EVENT_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
+send "$FAILED_METADATA_CHANGE_EVENT_NAME" "--partitions $PARTITIONS --topic $FAILED_METADATA_CHANGE_EVENT_NAME" \
+     "--entity-type topics --entity-name $FAILED_METADATA_CHANGE_EVENT_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
+
+send "$METADATA_CHANGE_LOG_VERSIONED_TOPIC_NAME" "--partitions $PARTITIONS --topic $METADATA_CHANGE_LOG_VERSIONED_TOPIC_NAME" \
+     "--entity-type topics --entity-name $METADATA_CHANGE_LOG_VERSIONED_TOPIC_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
 
 # Set retention to 90 days
-send "$METADATA_CHANGE_LOG_TIMESERIES_TOPIC_NAME" "--partitions $PARTITIONS --config retention.ms=7776000000 --topic $METADATA_CHANGE_LOG_TIMESERIES_TOPIC_NAME"
-send "$METADATA_CHANGE_PROPOSAL_TOPIC_NAME" "--partitions $PARTITIONS --topic $METADATA_CHANGE_PROPOSAL_TOPIC_NAME"
-send "$FAILED_METADATA_CHANGE_PROPOSAL_TOPIC_NAME" "--partitions $PARTITIONS --topic $FAILED_METADATA_CHANGE_PROPOSAL_TOPIC_NAME"
-send "$PLATFORM_EVENT_TOPIC_NAME" "--partitions $PARTITIONS --topic $PLATFORM_EVENT_TOPIC_NAME"
+send "$METADATA_CHANGE_LOG_TIMESERIES_TOPIC_NAME" "--partitions $PARTITIONS --config retention.ms=7776000000 --topic $METADATA_CHANGE_LOG_TIMESERIES_TOPIC_NAME" \
+     "--entity-type topics --entity-name $METADATA_CHANGE_LOG_TIMESERIES_TOPIC_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
+
+send "$METADATA_CHANGE_PROPOSAL_TOPIC_NAME" "--partitions $PARTITIONS --topic $METADATA_CHANGE_PROPOSAL_TOPIC_NAME" \
+     "--entity-type topics --entity-name $METADATA_CHANGE_PROPOSAL_TOPIC_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
+send "$FAILED_METADATA_CHANGE_PROPOSAL_TOPIC_NAME" "--partitions $PARTITIONS --topic $FAILED_METADATA_CHANGE_PROPOSAL_TOPIC_NAME" \
+     "--entity-type topics --entity-name $FAILED_METADATA_CHANGE_PROPOSAL_TOPIC_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
+
+send "$PLATFORM_EVENT_TOPIC_NAME" "--partitions $PARTITIONS --topic $PLATFORM_EVENT_TOPIC_NAME" \
+     "--entity-type topics --entity-name $PLATFORM_EVENT_TOPIC_NAME --alter --add-config max.message.bytes=$MAX_MESSAGE_BYTES"
 
 # Infinite retention upgrade topic
-send "$DATAHUB_UPGRADE_HISTORY_TOPIC_NAME" "--partitions 1 --config retention.ms=-1 --topic $DATAHUB_UPGRADE_HISTORY_TOPIC_NAME"
+  # Make sure the retention.ms config for $DATAHUB_UPGRADE_HISTORY_TOPIC_NAME is configured to infinite
+  # Please see the bug report below for details
+  # https://github.com/datahub-project/datahub/issues/7882
+send "$DATAHUB_UPGRADE_HISTORY_TOPIC_NAME" "--partitions 1 --config retention.ms=-1 --topic $DATAHUB_UPGRADE_HISTORY_TOPIC_NAME" \
+     "--entity-type topics --entity-name "$DATAHUB_UPGRADE_HISTORY_TOPIC_NAME" --alter --add-config retention.ms=-1"
+
 # Create topic for datahub usage event
 if [[ $DATAHUB_ANALYTICS_ENABLED == true ]]; then
   send "$DATAHUB_USAGE_EVENT_NAME" "--partitions $PARTITIONS --topic $DATAHUB_USAGE_EVENT_NAME"
@@ -150,8 +169,3 @@ if [[ $USE_CONFLUENT_SCHEMA_REGISTRY == "TRUE" ]]; then
       --entity-name _schemas \
       --alter --add-config cleanup.policy=compact
 fi
-
-# Make sure the retention.ms config for $DATAHUB_UPGRADE_HISTORY_TOPIC_NAME is configured to infinite
-# Please see the bug report below for details
-# https://github.com/datahub-project/datahub/issues/7882
-kafka-configs.sh --command-config $CONNECTION_PROPERTIES_PATH --bootstrap-server $KAFKA_BOOTSTRAP_SERVER --entity-type topics --entity-name "$DATAHUB_UPGRADE_HISTORY_TOPIC_NAME" --alter --add-config retention.ms=-1

docker/kafka-setup/kafka-topic-workers.sh

@@ -11,10 +11,18 @@ START_LOCK=$4
 ## the queue workers are supposed to be doing
 job() {
   i=$1
-  topic_args=$2
+  worker_args=$2
+  topic_args=$(echo $worker_args | cut -d "$DELIMITER" -f 1)
+  topic_config=$(echo $worker_args | cut -d "$DELIMITER" -f 2)
+
+  echo "   $i: kafka-topics.sh --create --if-not-exist $topic_args"
   kafka-topics.sh --create --if-not-exists --command-config $CONNECTION_PROPERTIES_PATH --bootstrap-server $KAFKA_BOOTSTRAP_SERVER \
      --replication-factor $REPLICATION_FACTOR \
      $topic_args
+  if [[ ! -z "$topic_config" ]]; then
+    echo "   $i: kafka-configs.sh $topic_config"
+    kafka-configs.sh --command-config $CONNECTION_PROPERTIES_PATH --bootstrap-server $KAFKA_BOOTSTRAP_SERVER $topic_config
+  fi
 }
 
 ## This is the worker to read from the queue.

docker/quickstart/docker-compose-m1.quickstart.yml

@@ -16,6 +16,8 @@ services:
     - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
     - KAFKA_HEAP_OPTS=-Xms256m -Xmx256m
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
+    - KAFKA_MESSAGE_MAX_BYTES=5242880
+    - KAFKA_MAX_MESSAGE_BYTES=5242880
     healthcheck:
       interval: 1s
       retries: 5

docker/quickstart/docker-compose-without-neo4j-m1.quickstart.yml

@@ -16,6 +16,8 @@ services:
     - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
     - KAFKA_HEAP_OPTS=-Xms256m -Xmx256m
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
+    - KAFKA_MESSAGE_MAX_BYTES=5242880
+    - KAFKA_MAX_MESSAGE_BYTES=5242880
     healthcheck:
       interval: 1s
       retries: 5

docker/quickstart/docker-compose-without-neo4j.quickstart.yml

@@ -16,6 +16,8 @@ services:
     - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
     - KAFKA_HEAP_OPTS=-Xms256m -Xmx256m
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
+    - KAFKA_MESSAGE_MAX_BYTES=5242880
+    - KAFKA_MAX_MESSAGE_BYTES=5242880
     healthcheck:
       interval: 1s
       retries: 5

docker/quickstart/docker-compose.quickstart.yml

@@ -16,6 +16,8 @@ services:
     - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
     - KAFKA_HEAP_OPTS=-Xms256m -Xmx256m
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
+    - KAFKA_MESSAGE_MAX_BYTES=5242880
+    - KAFKA_MAX_MESSAGE_BYTES=5242880
     healthcheck:
       interval: 1s
       retries: 5