upgrade-jetty-version: removed deprecated dependecy on monitoring-interceptors

Author: chirag-wadhwa5

pom.xml

@@ -178,17 +178,6 @@
             <artifactId>jakarta.xml.bind-api</artifactId>
             <version>4.0.1</version>
         </dependency>
-        <!--
-            This dependency is included to demonstrate the use of
-            Confluent Monitoring Interceptors in conjunction with
-            Kafka Streams but isn't a required
-            dependency for running Kafka Streams.
-        -->
-        <dependency>
-            <groupId>io.confluent</groupId>
-            <artifactId>monitoring-interceptors</artifactId>
-            <version>${io.confluent.blueway.version}</version>
-        </dependency>
 
         <dependency>
             <groupId>com.twitter</groupId>

src/main/java/io/confluent/examples/streams/WikipediaFeedAvroExample.java

@@ -16,7 +16,6 @@
 package io.confluent.examples.streams;
 
 import io.confluent.examples.streams.avro.WikiFeed;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -124,11 +123,6 @@ static KafkaStreams buildWikipediaFeed(final String bootstrapServers,
     // in order to keep this example interactive.
     streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
 
-    // If Confluent monitoring interceptors are on the classpath,
-    // then the producer and consumer interceptors are added to the
-    // streams application.
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsStreams(streamsConfiguration);
-
     final Serde<String> stringSerde = Serdes.String();
     final Serde<Long> longSerde = Serdes.Long();
 

src/main/java/io/confluent/examples/streams/microservices/AddInventory.java

@@ -3,7 +3,6 @@
 import io.confluent.examples.streams.avro.microservices.Product;
 import io.confluent.examples.streams.microservices.domain.Schemas;
 import io.confluent.examples.streams.microservices.domain.Schemas.Topics;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import org.apache.commons.cli.*;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -35,7 +34,6 @@ private static void sendInventory(final List<KeyValue<Product, Integer>> invento
         producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
         producerConfig.put(ProducerConfig.RETRIES_CONFIG, 1);
         producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "inventory-generator");
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(producerConfig);
 
         final ProductTypeSerde productSerde = new ProductTypeSerde();
 

src/main/java/io/confluent/examples/streams/microservices/OrderDetailsService.java

@@ -12,7 +12,6 @@
 import io.confluent.examples.streams.avro.microservices.OrderValidation;
 import io.confluent.examples.streams.avro.microservices.OrderValidationResult;
 import io.confluent.examples.streams.microservices.domain.Schemas;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 
 import java.io.IOException;
 import java.time.Duration;
@@ -133,7 +132,6 @@ private void startProducer(final String bootstrapServers, final Properties defau
     producerConfig.put(ProducerConfig.RETRIES_CONFIG, String.valueOf(Integer.MAX_VALUE));
     producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
     producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "order-details-service-producer");
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(producerConfig);
 
     producer = new KafkaProducer<>(producerConfig,
         Topics.ORDER_VALIDATIONS.keySerde().serializer(),
@@ -148,7 +146,6 @@ private void startConsumer(final String bootstrapServers, final Properties defau
     consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
     consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, !eosEnabled);
     consumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, "order-details-service-consumer");
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsConsumer(consumerConfig);
 
     consumer = new KafkaConsumer<>(consumerConfig,
         Topics.ORDERS.keySerde().deserializer(),

src/main/java/io/confluent/examples/streams/microservices/PostOrdersAndPayments.java

@@ -6,7 +6,6 @@
 import io.confluent.examples.streams.microservices.domain.Schemas;
 import io.confluent.examples.streams.microservices.domain.beans.OrderBean;
 import io.confluent.examples.streams.microservices.util.Paths;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer;
 import org.apache.commons.cli.*;
@@ -53,7 +52,6 @@ private static KafkaProducer<String, Payment> buildPaymentProducer(final String
         producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
         producerConfig.put(ProducerConfig.RETRIES_CONFIG, 1);
         producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "payment-generator");
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(producerConfig);
 
         return new KafkaProducer<>(producerConfig, new StringSerializer(), paymentSerializer);
     }

src/main/java/io/confluent/examples/streams/microservices/util/ConsumeCustomers.java

@@ -1,7 +1,6 @@
 package io.confluent.examples.streams.microservices.util;
 
 import io.confluent.examples.streams.microservices.domain.Schemas.Topics;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.DefaultParser;
@@ -60,7 +59,6 @@ public static void main(final String[] args) throws Exception {
         props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Topics.CUSTOMERS.keySerde().deserializer().getClass());
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Topics.CUSTOMERS.valueSerde().deserializer().getClass());
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsConsumer(props);
 
         try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
             consumer.subscribe(Collections.singletonList("customers"));

src/main/java/io/confluent/examples/streams/microservices/util/ConsumeOrders.java

@@ -1,7 +1,6 @@
 package io.confluent.examples.streams.microservices.util;
 
 import io.confluent.examples.streams.microservices.domain.Schemas.Topics;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.DefaultParser;
@@ -60,7 +59,6 @@ public static void main(final String[] args) throws Exception {
         props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Topics.ORDERS.keySerde().deserializer().getClass());
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Topics.ORDERS.valueSerde().deserializer().getClass());
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsConsumer(props);
 
         try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
             consumer.subscribe(Collections.singletonList("orders"));

src/main/java/io/confluent/examples/streams/microservices/util/ConsumePayments.java

@@ -1,7 +1,6 @@
 package io.confluent.examples.streams.microservices.util;
 
 import io.confluent.examples.streams.microservices.domain.Schemas.Topics;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.DefaultParser;
@@ -59,7 +58,6 @@ public static void main(final String[] args) throws Exception {
         props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Topics.PAYMENTS.keySerde().deserializer().getClass());
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Topics.PAYMENTS.valueSerde().deserializer().getClass());
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsConsumer(props);
 
         try (final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
             consumer.subscribe(Collections.singletonList("payments"));

src/main/java/io/confluent/examples/streams/microservices/util/MicroserviceUtils.java

@@ -3,7 +3,6 @@
 import io.confluent.examples.streams.avro.microservices.Product;
 import io.confluent.examples.streams.microservices.Service;
 import io.confluent.examples.streams.microservices.domain.Schemas;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -85,7 +84,6 @@ public static Properties baseStreamsConfig(final String bootstrapServers,
             enableEOS ? "exactly_once" : "at_least_once");
     config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1); //commit as fast as possible
     config.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), 30000);
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsStreams(config);
     return config;
   }
 
@@ -198,7 +196,6 @@ public static <T> KafkaProducer startProducer(final String bootstrapServers,
     producerConfig.put(ProducerConfig.RETRIES_CONFIG, String.valueOf(Integer.MAX_VALUE));
     producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
     producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "order-sender");
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(producerConfig);
 
     return new KafkaProducer<>(producerConfig,
         topic.keySerde().serializer(),

src/main/java/io/confluent/examples/streams/microservices/util/ProduceCustomers.java

@@ -1,7 +1,6 @@
 package io.confluent.examples.streams.microservices.util;
 
 import io.confluent.examples.streams.avro.microservices.Customer;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer;
 import org.apache.commons.cli.CommandLine;
@@ -62,7 +61,6 @@ public static void main(final String[] args) throws Exception {
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
         props.put(ProducerConfig.ACKS_CONFIG, "all");
         props.put(ProducerConfig.RETRIES_CONFIG, 1);
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(props);
 
         try (final KafkaProducer<Long, Customer> producer = new KafkaProducer<>(props, new LongSerializer(), mySerializer)) {
             while (true) {

src/main/java/io/confluent/examples/streams/microservices/util/ProduceOrders.java

@@ -1,7 +1,6 @@
 package io.confluent.examples.streams.microservices.util;
 
 import io.confluent.examples.streams.avro.microservices.Order;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer;
 import org.apache.commons.cli.CommandLine;
@@ -64,7 +63,6 @@ public static void main(final String[] args) throws Exception {
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
         props.put(ProducerConfig.ACKS_CONFIG, "all");
         props.put(ProducerConfig.RETRIES_CONFIG, 1);
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(props);
 
         try (final KafkaProducer<String, Order> producer = new KafkaProducer<>(props, new StringSerializer(), mySerializer)) {
             while (true) {

src/main/java/io/confluent/examples/streams/microservices/util/ProducePayments.java

@@ -1,7 +1,6 @@
 package io.confluent.examples.streams.microservices.util;
 
 import io.confluent.examples.streams.avro.microservices.Payment;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer;
 import org.apache.commons.cli.CommandLine;
@@ -62,7 +61,6 @@ public static void main(final String[] args) throws Exception {
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
         props.put(ProducerConfig.ACKS_CONFIG, "all");
         props.put(ProducerConfig.RETRIES_CONFIG, 1);
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(props);
 
         try (final KafkaProducer<String, Payment> producer = new KafkaProducer<>(props, new StringSerializer(), mySerializer)) {
             while (true) {

src/main/java/io/confluent/examples/streams/utils/MonitoringInterceptorUtils.java

@@ -20,7 +20,6 @@
 import io.confluent.examples.streams.avro.Order;
 import io.confluent.examples.streams.avro.Product;
 import io.confluent.examples.streams.kafka.EmbeddedSingleNodeKafkaCluster;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
@@ -114,8 +113,6 @@ public void shouldDemonstrateGlobalStoreJoins() throws Exception {
         consumerProps.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, CLUSTER.schemaRegistryUrl());
         consumerProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
 
-        MonitoringInterceptorUtils.maybeConfigureInterceptorsConsumer(consumerProps);
-
         // receive the enriched orders
         final List<EnrichedOrder>
                 enrichedOrders =

src/test/java/io/confluent/examples/streams/AbstractGlobalStoresAndTablesTest.java

@@ -16,7 +16,6 @@
 package io.confluent.examples.streams;
 
 import io.confluent.examples.streams.kafka.EmbeddedSingleNodeKafkaCluster;
-import io.confluent.examples.streams.utils.MonitoringInterceptorUtils;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.GenericRecordBuilder;
@@ -90,8 +89,6 @@ public void shouldProduceTopNArticles() throws Exception {
               io.confluent.kafka.serializers.KafkaAvroSerializer.class);
     props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, CLUSTER.schemaRegistryUrl());
 
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsProducer(props);
-
     final KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);
 
     final GenericRecordBuilder pageViewBuilder =
@@ -122,8 +119,6 @@ public void shouldProduceTopNArticles() throws Exception {
     final Deserializer<Windowed<String>> windowedDeserializer =
             WindowedSerdes.timeWindowedSerdeFrom(String.class, TopArticlesLambdaExample.windowSize.toMillis()).deserializer();
 
-    MonitoringInterceptorUtils.maybeConfigureInterceptorsConsumer(consumerProperties);
-
     final KafkaConsumer<Windowed<String>, String> consumer = new KafkaConsumer<>(consumerProperties,
                                                                                  windowedDeserializer,
                                                                                  Serdes.String().deserializer());