KAFKA-20297: Deprecate Time-accepting constructors in KafkaStreams and Metrics

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -518,6 +518,7 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
         return createInternal(config, timeoutProcessorFactory, null);
     }
 
+    @SuppressWarnings("deprecation")
     static KafkaAdminClient createInternal(
         AdminClientConfig config,
         TimeoutProcessorFactory timeoutProcessorFactory,
@@ -575,6 +576,7 @@ static KafkaAdminClient createInternal(
     }
 
     // Visible for tests
+    @SuppressWarnings("deprecation")
     static KafkaAdminClient createInternal(AdminClientConfig config,
                                            AdminMetadataManager metadataManager,
                                            KafkaClient client,

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

@@ -664,6 +664,7 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
         this.positionsValidator = positionsValidator;
     }
 
+    @SuppressWarnings("deprecation")
     AsyncKafkaConsumer(LogContext logContext,
                        Time time,
                        ConsumerConfig config,

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java

@@ -294,6 +294,7 @@ public class ClassicKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
     }
 
     // visible for testing
+    @SuppressWarnings("deprecation")
     ClassicKafkaConsumer(LogContext logContext,
                          Time time,
                          ConsumerConfig config,

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java

@@ -141,6 +141,7 @@ public static Metrics createMetrics(ConsumerConfig config, Time time) {
                 config.getString(ConsumerConfig.CLIENT_ID_CONFIG), config));
     }
 
+    @SuppressWarnings("deprecation")
     public static Metrics createMetrics(ConsumerConfig config, Time time, List<MetricsReporter> reporters) {
         String clientId = config.getString(ConsumerConfig.CLIENT_ID_CONFIG);
         Map<String, String> metricsTags = Collections.singletonMap(CONSUMER_CLIENT_ID_METRIC_TAG, clientId);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java

@@ -359,6 +359,7 @@ private void process(final ErrorEvent event) {
     }
 
     // Visible for testing
+    @SuppressWarnings("deprecation")
     ShareConsumerImpl(final LogContext logContext,
                       final String clientId,
                       final String groupId,

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -347,7 +347,7 @@ public KafkaProducer(Properties properties, Serializer<K> keySerializer, Seriali
     }
 
     // visible for testing
-    @SuppressWarnings({"unchecked", "this-escape"})
+    @SuppressWarnings({"unchecked", "this-escape", "deprecation"})
     KafkaProducer(ProducerConfig config,
                   Serializer<K> keySerializer,
                   Serializer<V> valueSerializer,

clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java

@@ -89,15 +89,19 @@ public Metrics() {
     /**
      * Create a metrics repository with no metric reporters and default configuration.
      * Expiration of Sensors is disabled.
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public Metrics(Time time) {
         this(new MetricConfig(), new ArrayList<>(0), time);
     }
 
     /**
      * Create a metrics repository with no metric reporters and the given default configuration.
      * Expiration of Sensors is disabled.
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public Metrics(MetricConfig defaultConfig, Time time) {
         this(defaultConfig, new ArrayList<>(0), time);
     }
@@ -118,7 +122,9 @@ public Metrics(MetricConfig defaultConfig) {
      * @param defaultConfig The default config
      * @param reporters The metrics reporters
      * @param time The time instance to use with the metrics
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time time) {
         this(defaultConfig, reporters, time, false);
     }
@@ -130,7 +136,9 @@ public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time
      * @param reporters The metrics reporters
      * @param time The time instance to use with the metrics
      * @param metricsContext The metricsContext to initialize metrics reporter with
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time time, MetricsContext metricsContext) {
         this(defaultConfig, reporters, time, false, metricsContext);
     }
@@ -141,7 +149,9 @@ public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time
      * @param reporters The metrics reporters
      * @param time The time instance to use with the metrics
      * @param enableExpiration true if the metrics instance can garbage collect inactive sensors, false otherwise
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time time, boolean enableExpiration) {
         this(defaultConfig, reporters, time, enableExpiration, new KafkaMetricsContext(""));
     }
@@ -154,7 +164,9 @@ public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time
      * @param time The time instance to use with the metrics
      * @param enableExpiration true if the metrics instance can garbage collect inactive sensors, false otherwise
      * @param metricsContext The metricsContext to initialize metrics reporter with
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public Metrics(MetricConfig defaultConfig, List<MetricsReporter> reporters, Time time, boolean enableExpiration,
                    MetricsContext metricsContext) {
         this.config = defaultConfig;

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -847,7 +847,9 @@ public KafkaStreams(final Topology topology,
      * @param props          properties for {@link StreamsConfig}
      * @param time           {@code Time} implementation; cannot be null
      * @throws StreamsException if any fatal error occurs
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public KafkaStreams(final Topology topology,
                         final Properties props,
                         final Time time) {
@@ -866,7 +868,9 @@ public KafkaStreams(final Topology topology,
      *                       for the new {@code KafkaStreams} instance
      * @param time           {@code Time} implementation; cannot be null
      * @throws StreamsException if any fatal error occurs
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public KafkaStreams(final Topology topology,
                         final Properties props,
                         final KafkaClientSupplier clientSupplier,
@@ -917,7 +921,9 @@ public KafkaStreams(final Topology topology,
      * @param applicationConfigs         configs for Kafka Streams
      * @param time           {@code Time} implementation; cannot be null
      * @throws StreamsException if any fatal error occurs
+     * @deprecated This constructor is intended for internal use only and will be made package-private in version 5.0.
      */
+    @Deprecated(since = "4.4")
     public KafkaStreams(final Topology topology,
                         final StreamsConfig applicationConfigs,
                         final Time time) {
@@ -1097,6 +1103,7 @@ private StreamThread createAndAddStreamThread(final long cacheSizePerThread,
         return streamThread;
     }
 
+    @SuppressWarnings("deprecation")
     private static Metrics createMetrics(final StreamsConfig config, final Time time, final String clientId) {
         final MetricConfig metricConfig = new MetricConfig()
             .samples(config.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

@@ -382,6 +382,7 @@ private static void logIfTaskIdleEnabled(final StreamsConfig streamsConfig) {
         }
     }
 
+    @SuppressWarnings("deprecation")
     private StreamsMetricsImpl setupMetrics(final StreamsConfig streamsConfig) {
         final String threadId = Thread.currentThread().getName();