Status

Current state: Under Discussion

Discussion thread: Link 

JIRA: KAFKA-7402

Motivation

Quoting John John Roesler:

"Various components in Streams have close methods but do not implement AutoCloseable. This means that they can't be used in try-with-resources blocks.

Remedying that would simplify our tests and make life easier for users as well.

KafkaStreams itself is a notable example of this, but we can take the opportunity to look for other components that make sense as AutoCloseable as well."

So these classes can be used in a try-with-resources Statement after implementing AutoCloseable.

From AutoCloseable's Javadoc:

An object that may hold resources (such as file or socket handles) until it is closed. The {@link #close()} method of an {@code AutoCloseable} object is called automatically when exiting a {@code try}-with-resources block for which the object has been declared in the resource specification header. This construction ensures prompt release, avoiding resource exhaustion exceptions and errors that may otherwise occur.

Public Interfaces

By going over the project, here is a list that I found which can implement AutoCloseable. Suggestions are welcome.

1. org.apache.kafka.streams.KafkaStreams
2. org.apache.kafka.streams.processor.internals.RecordCollector
3. org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup
4. org.apache.kafka.connect.transforms.TimestampRouter
5. org.apache.kafka.tools.VerifiableProducer
   
   

Additions provided by Chia-pingChia-Ping Tsai:

1. org.apache.kafka.connect.runtime.WorkerConnector
2. org.apache.kafka.common.record.MemoryRecordsBuilder
3. org.apache.kafka.common.metrics.MetricsReporter
4. org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin
5. org.apache.kafka.common.network.KafkaChannel
6. org.apache.kafka.clients.consumer.ConsumerInterceptor
7. org.apache.kafka.common.network.Selector.SelectorMetrics
8. org.apache.kafka.clients.consumer.internals.AbstractCoordinator.HeartbeatThread

Proposed Changes

Here are some examples for the proposed changes:

org.apache.kafka.streams.KafkaStreams

public class KafkaStreams implements AutoCloseable{
...
	public void close()
...
}

org.apache.kafka.streams.processor.internals.RecordCollector

public interface RecordCollector extends AutoCloseable {
...
}

org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup

public class MetricGroup implements AutoCloseable {
...
}

org.apache.kafka.connect.transforms.TimestampRouter

public class TimestampRouter<R extends ConnectRecord<R>> implements Transformation<R>, AutoCloseable {
...
}

org.apache.kafka.tools.VerifiableProducer

public class VerifiableProducer implements AutoCloseable {
...
}

For Chia-ping's list, the changes are very similar ( ...implements AutoCloseable). Specific changes can be added to this KIP if requested by reviewers.

Compatibility, Deprecation, and Migration Plan

• N/A