You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by di...@apache.org on 2023/12/21 12:47:28 UTC
(kafka) branch trunk updated: KAFKA-13950: Fix resource leak in error scenarios (#12228)
This is an automated email from the ASF dual-hosted git repository.
divijv pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6250049e107 KAFKA-13950: Fix resource leak in error scenarios (#12228)
6250049e107 is described below
commit 6250049e107f2279fa866bc61e5150187f7c1a7f
Author: Divij Vaidya <di...@amazon.com>
AuthorDate: Thu Dec 21 13:47:22 2023 +0100
KAFKA-13950: Fix resource leak in error scenarios (#12228)
We are not properly closing Closeable resources in the code base at multiple places especially when we have an exception. This code change fixes multiple of these leaks.
Reviewers: Ismael Juma <is...@juma.me.uk>, Luke Chen <sh...@gmail.com>, Mickael Maison <mi...@gmail.com>
---
.../kafka/common/network/ChannelBuilder.java | 1 +
.../common/network/PlaintextChannelBuilder.java | 9 +++--
.../kafka/common/network/SaslChannelBuilder.java | 11 ++++--
.../apache/kafka/common/network/Selectable.java | 6 ++++
.../org/apache/kafka/common/network/Selector.java | 8 +++--
.../kafka/common/network/SslChannelBuilder.java | 9 +++--
.../apache/kafka/common/record/LegacyRecord.java | 3 +-
.../apache/kafka/common/record/MemoryRecords.java | 32 ++++++++---------
.../apache/kafka/common/network/SelectorTest.java | 40 ++++++++++++++--------
.../server/builders/ReplicaManagerBuilder.java | 5 ++-
.../examples/ExactlyOnceMessageProcessor.java | 13 ++++++-
.../org/apache/kafka/streams/KafkaStreams.java | 4 ++-
.../processor/internals/StateDirectory.java | 3 +-
.../AbstractRocksDBSegmentedBytesStore.java | 5 ++-
14 files changed, 102 insertions(+), 47 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
index 0cf1d74f85c..f262b2ed1d9 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
@@ -34,6 +34,7 @@ public interface ChannelBuilder extends AutoCloseable, Configurable {
* @param key SelectionKey
* @param maxReceiveSize max size of a single receive buffer to allocate
* @param memoryPool memory pool from which to allocate buffers, or null for none
+ * @param metadataRegistry registry which stores the metadata about the channels
* @return KafkaChannel
*/
KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize,
diff --git a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
index 8a39cdfa3b2..0b4738afa2c 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
@@ -54,12 +54,17 @@ public class PlaintextChannelBuilder implements ChannelBuilder {
@Override
public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize,
MemoryPool memoryPool, ChannelMetadataRegistry metadataRegistry) throws KafkaException {
+ PlaintextTransportLayer transportLayer = null;
try {
- PlaintextTransportLayer transportLayer = buildTransportLayer(key);
- Supplier<Authenticator> authenticatorCreator = () -> new PlaintextAuthenticator(configs, transportLayer, listenerName);
+ transportLayer = buildTransportLayer(key);
+ final PlaintextTransportLayer finalTransportLayer = transportLayer;
+ Supplier<Authenticator> authenticatorCreator = () -> new PlaintextAuthenticator(configs, finalTransportLayer, listenerName);
return buildChannel(id, transportLayer, authenticatorCreator, maxReceiveSize,
memoryPool != null ? memoryPool : MemoryPool.NONE, metadataRegistry);
} catch (Exception e) {
+ // Ideally these resources are closed by the KafkaChannel but this builder should close the resources instead
+ // if an error occurs due to which KafkaChannel is not created.
+ Utils.closeQuietly(transportLayer, "transport layer for channel Id: " + id);
throw new KafkaException(e);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
index fbe4cf0f28d..3f5cbdb588f 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
@@ -210,16 +210,18 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
@Override
public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize,
MemoryPool memoryPool, ChannelMetadataRegistry metadataRegistry) throws KafkaException {
+ TransportLayer transportLayer = null;
try {
SocketChannel socketChannel = (SocketChannel) key.channel();
Socket socket = socketChannel.socket();
- TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel, metadataRegistry);
+ transportLayer = buildTransportLayer(id, key, socketChannel, metadataRegistry);
+ final TransportLayer finalTransportLayer = transportLayer;
Supplier<Authenticator> authenticatorCreator;
if (mode == Mode.SERVER) {
authenticatorCreator = () -> buildServerAuthenticator(configs,
Collections.unmodifiableMap(saslCallbackHandlers),
id,
- transportLayer,
+ finalTransportLayer,
Collections.unmodifiableMap(subjects),
Collections.unmodifiableMap(connectionsMaxReauthMsByMechanism),
metadataRegistry);
@@ -230,12 +232,15 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
id,
socket.getInetAddress().getHostName(),
loginManager.serviceName(),
- transportLayer,
+ finalTransportLayer,
subjects.get(clientSaslMechanism));
}
return new KafkaChannel(id, transportLayer, authenticatorCreator, maxReceiveSize,
memoryPool != null ? memoryPool : MemoryPool.NONE, metadataRegistry);
} catch (Exception e) {
+ // Ideally these resources are closed by the KafkaChannel but this builder should close the resources instead
+ // if an error occurs due to which KafkaChannel is not created.
+ Utils.closeQuietly(transportLayer, "transport layer for channel Id: " + id);
throw new KafkaException(e);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selectable.java b/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
index afdd42e4a9c..425464f7e22 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selectable.java
@@ -17,6 +17,8 @@
package org.apache.kafka.common.network;
+import org.apache.kafka.common.memory.MemoryPool;
+
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Collection;
@@ -78,6 +80,10 @@ public interface Selectable {
/**
* The collection of receives that completed on the last {@link #poll(long) poll()} call.
+ *
+ * Note that the caller of this method assumes responsibility to close the NetworkReceive resources which may be
+ * backed by a {@link MemoryPool}. In such scenarios (when NetworkReceive uses a {@link MemoryPool}, it is necessary
+ * to close the {@link NetworkReceive} to prevent any memory leaks.
*/
Collection<NetworkReceive> completedReceives();
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index 195b48459d6..b63904ab3a3 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -334,9 +334,10 @@ public class Selector implements Selectable, AutoCloseable {
}
private KafkaChannel buildAndAttachKafkaChannel(SocketChannel socketChannel, String id, SelectionKey key) throws IOException {
+ ChannelMetadataRegistry metadataRegistry = null;
try {
- KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize, memoryPool,
- new SelectorChannelMetadataRegistry());
+ metadataRegistry = new SelectorChannelMetadataRegistry();
+ KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize, memoryPool, metadataRegistry);
key.attach(channel);
return channel;
} catch (Exception e) {
@@ -345,6 +346,9 @@ public class Selector implements Selectable, AutoCloseable {
} finally {
key.cancel();
}
+ // Ideally, these resources are closed by the KafkaChannel but if the KafkaChannel is not created to an
+ // error, this builder should close the resources it has created instead.
+ Utils.closeQuietly(metadataRegistry, "metadataRegistry");
throw new IOException("Channel could not be created for socket " + socketChannel, e);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
index caaceb1b519..1c841e55eb9 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
@@ -101,13 +101,18 @@ public class SslChannelBuilder implements ChannelBuilder, ListenerReconfigurable
@Override
public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize,
MemoryPool memoryPool, ChannelMetadataRegistry metadataRegistry) throws KafkaException {
+ SslTransportLayer transportLayer = null;
try {
- SslTransportLayer transportLayer = buildTransportLayer(sslFactory, id, key, metadataRegistry);
+ transportLayer = buildTransportLayer(sslFactory, id, key, metadataRegistry);
+ final SslTransportLayer finalTransportLayer = transportLayer;
Supplier<Authenticator> authenticatorCreator = () ->
- new SslAuthenticator(configs, transportLayer, listenerName, sslPrincipalMapper);
+ new SslAuthenticator(configs, finalTransportLayer, listenerName, sslPrincipalMapper);
return new KafkaChannel(id, transportLayer, authenticatorCreator, maxReceiveSize,
memoryPool != null ? memoryPool : MemoryPool.NONE, metadataRegistry);
} catch (Exception e) {
+ // Ideally these resources are closed by the KafkaChannel but this builder should close the resources instead
+ // if an error occurs due to which KafkaChannel is not created.
+ Utils.closeQuietly(transportLayer, "transport layer for channel Id: " + id);
throw new KafkaException(e);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java b/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java
index f016cbcbc76..014a5f4dfb1 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java
@@ -385,8 +385,7 @@ public final class LegacyRecord {
ByteBuffer value,
CompressionType compressionType,
TimestampType timestampType) {
- try {
- DataOutputStream out = new DataOutputStream(new ByteBufferOutputStream(buffer));
+ try (DataOutputStream out = new DataOutputStream(new ByteBufferOutputStream(buffer))) {
write(out, magic, timestamp, key, value, compressionType, timestampType);
} catch (IOException e) {
throw new KafkaException(e);
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
index 888bcdf2cb1..3aa233c34e9 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
@@ -194,25 +194,24 @@ public class MemoryRecords extends AbstractRecords {
batch.writeTo(bufferOutputStream);
filterResult.updateRetainedBatchMetadata(batch, retainedRecords.size(), false);
} else {
- final MemoryRecordsBuilder builder;
long deleteHorizonMs;
if (needToSetDeleteHorizon)
deleteHorizonMs = filter.currentTime + filter.deleteRetentionMs;
else
deleteHorizonMs = batch.deleteHorizonMs().orElse(RecordBatch.NO_TIMESTAMP);
- builder = buildRetainedRecordsInto(batch, retainedRecords, bufferOutputStream, deleteHorizonMs);
-
- MemoryRecords records = builder.build();
- int filteredBatchSize = records.sizeInBytes();
- if (filteredBatchSize > batch.sizeInBytes() && filteredBatchSize > maxRecordBatchSize)
- log.warn("Record batch from {} with last offset {} exceeded max record batch size {} after cleaning " +
- "(new size is {}). Consumers with version earlier than 0.10.1.0 may need to " +
- "increase their fetch sizes.",
+ try (final MemoryRecordsBuilder builder = buildRetainedRecordsInto(batch, retainedRecords, bufferOutputStream, deleteHorizonMs)) {
+ MemoryRecords records = builder.build();
+ int filteredBatchSize = records.sizeInBytes();
+ if (filteredBatchSize > batch.sizeInBytes() && filteredBatchSize > maxRecordBatchSize)
+ log.warn("Record batch from {} with last offset {} exceeded max record batch size {} after cleaning " +
+ "(new size is {}). Consumers with version earlier than 0.10.1.0 may need to " +
+ "increase their fetch sizes.",
partition, batch.lastOffset(), maxRecordBatchSize, filteredBatchSize);
- MemoryRecordsBuilder.RecordsInfo info = builder.info();
- filterResult.updateRetainedBatchMetadata(info.maxTimestamp, info.shallowOffsetOfMaxTimestamp,
+ MemoryRecordsBuilder.RecordsInfo info = builder.info();
+ filterResult.updateRetainedBatchMetadata(info.maxTimestamp, info.shallowOffsetOfMaxTimestamp,
maxOffset, retainedRecords.size(), filteredBatchSize);
+ }
}
} else if (batchRetention == BatchRetention.RETAIN_EMPTY) {
if (batchMagic < RecordBatch.MAGIC_VALUE_V2)
@@ -677,12 +676,13 @@ public class MemoryRecords extends AbstractRecords {
long logAppendTime = RecordBatch.NO_TIMESTAMP;
if (timestampType == TimestampType.LOG_APPEND_TIME)
logAppendTime = System.currentTimeMillis();
- MemoryRecordsBuilder builder = new MemoryRecordsBuilder(bufferStream, magic, compressionType, timestampType,
+ try (final MemoryRecordsBuilder builder = new MemoryRecordsBuilder(bufferStream, magic, compressionType, timestampType,
initialOffset, logAppendTime, producerId, producerEpoch, baseSequence, isTransactional, false,
- partitionLeaderEpoch, sizeEstimate);
- for (SimpleRecord record : records)
- builder.append(record);
- return builder.build();
+ partitionLeaderEpoch, sizeEstimate)) {
+ for (SimpleRecord record : records)
+ builder.append(record);
+ return builder.build();
+ }
}
public static MemoryRecords withEndTransactionMarker(long producerId, short producerEpoch, EndTransactionMarker marker) {
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
index 8c8ab97b863..e8fb83f017b 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.common.network;
-import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.memory.MemoryPool;
import org.apache.kafka.common.memory.SimpleMemoryPool;
@@ -32,6 +31,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.mockito.MockedConstruction;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
@@ -66,9 +66,13 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -424,20 +428,26 @@ public class SelectorTest {
@Test
public void registerFailure() throws Exception {
- ChannelBuilder channelBuilder = new PlaintextChannelBuilder(null) {
- @Override
- public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize,
- MemoryPool memoryPool, ChannelMetadataRegistry metadataRegistry) throws KafkaException {
- throw new RuntimeException("Test exception");
- }
- };
- Selector selector = new Selector(CONNECTION_MAX_IDLE_MS, new Metrics(), new MockTime(), "MetricGroup", channelBuilder, new LogContext());
- SocketChannel socketChannel = SocketChannel.open();
- socketChannel.configureBlocking(false);
- IOException e = assertThrows(IOException.class, () -> selector.register("1", socketChannel));
- assertTrue(e.getCause().getMessage().contains("Test exception"), "Unexpected exception: " + e);
- assertFalse(socketChannel.isOpen(), "Socket not closed");
- selector.close();
+ final String channelId = "1";
+
+ final ChannelBuilder channelBuilder = mock(ChannelBuilder.class);
+
+ when(channelBuilder.buildChannel(eq(channelId), any(SelectionKey.class), anyInt(), any(MemoryPool.class),
+ any(ChannelMetadataRegistry.class))).thenThrow(new RuntimeException("Test exception"));
+
+ try (MockedConstruction<Selector.SelectorChannelMetadataRegistry> mockedMetadataRegistry =
+ mockConstruction(Selector.SelectorChannelMetadataRegistry.class)) {
+ Selector selector = new Selector(CONNECTION_MAX_IDLE_MS, new Metrics(), new MockTime(), "MetricGroup", channelBuilder, new LogContext());
+ final SocketChannel socketChannel = SocketChannel.open();
+ socketChannel.configureBlocking(false);
+ IOException e = assertThrows(IOException.class, () -> selector.register(channelId, socketChannel));
+ assertTrue(e.getCause().getMessage().contains("Test exception"), "Unexpected exception: " + e);
+ assertFalse(socketChannel.isOpen(), "Socket not closed");
+ // Ideally, metadataRegistry is closed by the KafkaChannel but if the KafkaChannel is not created due to
+ // an error such as in a case like this, the Selector should be closing the metadataRegistry instead.
+ verify(mockedMetadataRegistry.constructed().get(0)).close();
+ selector.close();
+ }
}
@Test
diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
index 6859900100e..13b60a539f9 100644
--- a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
@@ -181,12 +181,15 @@ public class ReplicaManagerBuilder {
public ReplicaManager build() {
if (config == null) config = new KafkaConfig(Collections.emptyMap());
- if (metrics == null) metrics = new Metrics();
if (logManager == null) throw new RuntimeException("You must set logManager");
if (metadataCache == null) throw new RuntimeException("You must set metadataCache");
if (logDirFailureChannel == null) throw new RuntimeException("You must set logDirFailureChannel");
if (alterPartitionManager == null) throw new RuntimeException("You must set alterIsrManager");
if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(Optional.of(config));
+ // Initialize metrics in the end just before passing it to ReplicaManager to ensure ReplicaManager closes the
+ // metrics correctly. There might be a resource leak if it is initialized and an exception occurs between
+ // its initialization and creation of ReplicaManager.
+ if (metrics == null) metrics = new Metrics();
return new ReplicaManager(config,
metrics,
time,
diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java
index fa3a5b81a2e..cf12cc9f158 100644
--- a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java
+++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java
@@ -48,7 +48,7 @@ import static java.util.Collections.singleton;
/**
* This class implements a read-process-write application.
*/
-public class ExactlyOnceMessageProcessor extends Thread implements ConsumerRebalanceListener {
+public class ExactlyOnceMessageProcessor extends Thread implements ConsumerRebalanceListener, AutoCloseable {
private final String bootstrapServers;
private final String inputTopic;
private final String outputTopic;
@@ -214,4 +214,15 @@ public class ExactlyOnceMessageProcessor extends Thread implements ConsumerRebal
}
}).sum();
}
+
+ @Override
+ public void close() throws Exception {
+ if (producer != null) {
+ producer.close();
+ }
+
+ if (consumer != null) {
+ consumer.close();
+ }
+ }
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index 2fdb1e3a8f0..ff928e6ee81 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -42,6 +42,7 @@ import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.errors.InvalidStateStorePartitionException;
import org.apache.kafka.streams.errors.ProcessorStateException;
@@ -167,7 +168,6 @@ public class KafkaStreams implements AutoCloseable {
private final Metrics metrics;
protected final StreamsConfig applicationConfigs;
protected final List<StreamThread> threads;
- protected final StateDirectory stateDirectory;
protected final StreamsMetadataState streamsMetadataState;
private final ScheduledExecutorService stateDirCleaner;
private final ScheduledExecutorService rocksDBMetricsRecordingService;
@@ -184,6 +184,7 @@ public class KafkaStreams implements AutoCloseable {
private final DelegatingStandbyUpdateListener delegatingStandbyUpdateListener;
GlobalStreamThread globalStreamThread;
+ protected StateDirectory stateDirectory = null;
private KafkaStreams.StateListener stateListener;
private boolean oldHandler;
private BiConsumer<Throwable, Boolean> streamsUncaughtExceptionHandler;
@@ -974,6 +975,7 @@ public class KafkaStreams implements AutoCloseable {
stateDirectory = new StateDirectory(applicationConfigs, time, topologyMetadata.hasPersistentStores(), topologyMetadata.hasNamedTopologies());
processId = stateDirectory.initializeProcessId();
} catch (final ProcessorStateException fatal) {
+ Utils.closeQuietly(stateDirectory, "streams state directory");
throw new StreamsException(fatal);
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
index ecb7dd168a3..6be9f87f3c5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
@@ -60,7 +60,7 @@ import static org.apache.kafka.streams.processor.internals.StateManagerUtil.pars
* stored. Handles creation/locking/unlocking/cleaning of the Task Directories. This class is not
* thread-safe.
*/
-public class StateDirectory {
+public class StateDirectory implements AutoCloseable {
private static final Pattern TASK_DIR_PATH_NAME = Pattern.compile("\\d+_\\d+");
private static final Pattern NAMED_TOPOLOGY_DIR_PATH_NAME = Pattern.compile("__.+__"); // named topology dirs follow '__Topology-Name__'
@@ -377,6 +377,7 @@ public class StateDirectory {
}
}
+ @Override
public void close() {
if (hasPersistentStores) {
try {
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
index 1cb0e107a31..79f4f460989 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.ProcessorStateException;
@@ -394,10 +395,12 @@ public class AbstractRocksDBSegmentedBytesStore<S extends Segment> implements Se
consistencyEnabled,
position
);
+ WriteBatch batch = null;
try {
- final WriteBatch batch = writeBatchMap.computeIfAbsent(segment, s -> new WriteBatch());
+ batch = writeBatchMap.computeIfAbsent(segment, s -> new WriteBatch());
segment.addToBatch(new KeyValue<>(record.key(), record.value()), batch);
} catch (final RocksDBException e) {
+ Utils.closeQuietly(batch, "rocksdb write batch");
throw new ProcessorStateException("Error restoring batch to store " + this.name, e);
}
}