Table of Contents |
---|
Status
Current state: Under DiscussionAccepted (2.2.0)
Discussion thread: Link here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: KAFKA-7402
Motivation
...
"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.
...
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.
...
- org.apache.kafka.streams.KafkaStreams
- org.apache.kafka.streams.processor.internals.RecordCollector
- org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup
- org.apache.kafka.connect.transforms.TimestampRouter
- org.apache.kafka.tools.VerifiableProducer
Additions provided by Chia-pingChia-Ping Tsai:
- org.apache.kafka.connect.runtime.WorkerConnector.ConnectorMetricsGroup
- org.apache.kafka.common.record.MemoryRecordsBuilder
- org.apache.kafka.common.metrics.MetricsReporter
- org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin
- org.apache.kafka.common.network.KafkaChannel
- org.apache.kafka.clients.consumer.ConsumerInterceptor
- org.apache.kafka.common.network.Selector.SelectorMetrics
- org.apache.kafka.clients.consumer.internals.AbstractCoordinator.HeartbeatThread
Proposed Changes
Here are some examples for the proposed changesan example of how it would look like:
org.apache.kafka.streams.KafkaStreams
Code Block |
---|
public class KafkaStreams implements AutoCloseable{
...
public void close()
...
} |
org.apache.kafka.streams.processor.internals.RecordCollector
Code Block |
---|
public interface RecordCollector extends AutoCloseable {
...
} |
org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup
Code Block |
---|
public class MetricGroup implements AutoCloseable {
...
} |
org.apache.kafka.connect.transforms.TimestampRouter
Code Block |
---|
public class TimestampRouter<R extends ConnectRecord<R>> implements Transformation<R>, AutoCloseable {
...
} |
org.apache.kafka.tools.VerifiableProducer
Code Block |
---|
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
Compatibility, Deprecation, and Migration Plan
- This change will not have backward compatibility issues. It is legal to implement this AutoClosable interface without declaring "throws Exception" on close(), so the close() method signature won't be changed.
Rejected Alternatives
- We decided not to choose Closeable(). Closeable extends AutoCloseable, and AutoCloseable throws a broader exception than Closeable, so we think that AutoCloseable is a more generic and a more compatible option.N/A