This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new a44572bb4e45 fix(flink): close lookup reader after cache reload
attempts (#19503)
a44572bb4e45 is described below
commit a44572bb4e45689bc1fa0ab9e0c4daca17caaf6d
Author: Danny Chan <[email protected]>
AuthorDate: Thu Aug 6 10:36:01 2026 +0800
fix(flink): close lookup reader after cache reload attempts (#19503)
---
.../hudi/table/lookup/HoodieLookupFunction.java | 17 ++--
.../hudi/table/lookup/HoodieLookupTableReader.java | 36 +++++--
.../table/lookup/TestHoodieLookupFunction.java | 44 ++++++++-
.../table/lookup/TestHoodieLookupTableReader.java | 110 +++++++++++++++++++++
4 files changed, 189 insertions(+), 18 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
index 10c47b2fe1c9..7b02f9da826b 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupFunction.java
@@ -155,15 +155,16 @@ public class HoodieLookupFunction extends LookupFunction
implements Serializable
try {
long count = 0;
GenericRowData reuse = new GenericRowData(rowType.getFieldCount());
- partitionReader.open();
- RowData row;
- while ((row = partitionReader.read(reuse)) != null) {
- count++;
- RowData rowData = serializer.copy(row);
- RowData key = extractLookupKey(rowData);
- cache.addRow(key, rowData);
+ try (HoodieLookupTableReader reader = partitionReader) {
+ reader.open();
+ RowData row;
+ while ((row = reader.read(reuse)) != null) {
+ count++;
+ RowData rowData = serializer.copy(row);
+ RowData key = extractLookupKey(rowData);
+ cache.addRow(key, rowData);
+ }
}
- partitionReader.close();
currentCommit = latestCommitInstant.get();
scheduleNextLoad();
log.info("Loaded {} row(s) into lookup join cache", count);
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java
index 31fdf6d85d2a..c2c3fbf0472f 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/lookup/HoodieLookupTableReader.java
@@ -28,16 +28,19 @@ import org.apache.flink.table.data.RowData;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.util.CloseableUtils.closeSuppressing;
+
/**
* Hudi look up table reader.
*/
-public class HoodieLookupTableReader implements Serializable {
+public class HoodieLookupTableReader implements Serializable, Closeable {
private static final long serialVersionUID = 1L;
private final SerializableSupplier<InputFormat<RowData, ?>>
inputFormatSupplier;
@@ -53,11 +56,17 @@ public class HoodieLookupTableReader implements
Serializable {
}
public void open() throws IOException {
+ close();
this.inputFormat = inputFormatSupplier.get();
- inputFormat.configure(conf);
- this.inputSplits =
Arrays.stream(inputFormat.createInputSplits(1)).collect(Collectors.toList());
- ((RichInputFormat) inputFormat).openInputFormat();
- inputFormat.open(inputSplits.remove(0));
+ try {
+ inputFormat.configure(conf);
+ this.inputSplits =
Arrays.stream(inputFormat.createInputSplits(1)).collect(Collectors.toList());
+ ((RichInputFormat) inputFormat).openInputFormat();
+ inputFormat.open(inputSplits.remove(0));
+ } catch (IOException | RuntimeException e) {
+ closeSuppressing(this, e);
+ throw e;
+ }
}
@Nullable
@@ -77,12 +86,21 @@ public class HoodieLookupTableReader implements
Serializable {
return null;
}
+ @Override
public void close() throws IOException {
- if (this.inputFormat != null) {
- inputFormat.close();
+ InputFormat format = this.inputFormat;
+ this.inputFormat = null;
+ this.inputSplits = null;
+ if (format == null) {
+ return;
}
- if (inputFormat instanceof RichInputFormat) {
- ((RichInputFormat) inputFormat).closeInputFormat();
+
+ if (format instanceof RichInputFormat) {
+ try (Closeable ignored = ((RichInputFormat) format)::closeInputFormat) {
+ format.close();
+ }
+ } else {
+ format.close();
}
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
index 750f11f7956c..e2af300219cd 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupFunction.java
@@ -104,6 +104,25 @@ class TestHoodieLookupFunction {
}
}
+ @Test
+ void testReaderIsClosedWhenCacheReloadFails() throws Exception {
+ Configuration conf = getConf();
+ TestData.writeData(TestData.DATA_SET_SINGLE_INSERT, conf);
+
+ FailingLookupTableReader reader = new FailingLookupTableReader(conf);
+ HoodieLookupFunction function = newLookupFunction(reader, conf);
+ function.open(null);
+
+ Thread.currentThread().interrupt();
+ try {
+ assertThrows(RuntimeException.class, () -> function.lookup(lookupKey()));
+ assertEquals(1, reader.closeCount, "The failed reload attempt should
close the reader");
+ } finally {
+ Thread.interrupted();
+ function.close();
+ }
+ }
+
@Test
void testRocksDBCacheLifecycleAndLookupFailure() throws Exception {
Configuration conf = getConf();
@@ -130,7 +149,7 @@ class TestHoodieLookupFunction {
function.close();
}
- private HoodieLookupFunction newLookupFunction(CountingLookupTableReader
reader, Configuration conf) {
+ private HoodieLookupFunction newLookupFunction(HoodieLookupTableReader
reader, Configuration conf) {
return new HoodieLookupFunction(
reader,
TestConfigurations.ROW_TYPE,
@@ -205,4 +224,27 @@ class TestHoodieLookupFunction {
// no-op
}
}
+
+ private static class FailingLookupTableReader extends
HoodieLookupTableReader {
+ private int closeCount;
+
+ private FailingLookupTableReader(Configuration conf) {
+ super(() -> null, conf);
+ }
+
+ @Override
+ public void open() {
+ // no-op
+ }
+
+ @Override
+ public RowData read(RowData reuse) throws IOException {
+ throw new IOException("expected");
+ }
+
+ @Override
+ public void close() {
+ closeCount++;
+ }
+ }
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupTableReader.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupTableReader.java
new file mode 100644
index 000000000000..35c650d34d0a
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/lookup/TestHoodieLookupTableReader.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.lookup;
+
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.flink.api.common.io.RichInputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.table.data.RowData;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link HoodieLookupTableReader}.
+ */
+class TestHoodieLookupTableReader {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testOpenRollsBackPartiallyOpenedInputFormat() throws Exception {
+ RichInputFormat<RowData, InputSplit> inputFormat =
mock(RichInputFormat.class);
+ InputSplit inputSplit = mock(InputSplit.class);
+ when(inputFormat.createInputSplits(1)).thenReturn(new InputSplit[]
{inputSplit});
+ IOException openException = new IOException("expected open failure");
+ doThrow(openException).when(inputFormat).open(inputSplit);
+
+ HoodieLookupTableReader reader =
+ new HoodieLookupTableReader(() -> inputFormat, new Configuration());
+
+ assertSame(openException, assertThrows(IOException.class, reader::open));
+ verify(inputFormat).close();
+ verify(inputFormat).closeInputFormat();
+
+ reader.close();
+ verify(inputFormat, times(1)).close();
+ verify(inputFormat, times(1)).closeInputFormat();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testOpenPreservesFailureWhenRuntimeRollbackFails() throws Exception {
+ RichInputFormat<RowData, InputSplit> inputFormat =
mock(RichInputFormat.class);
+ InputSplit inputSplit = mock(InputSplit.class);
+ when(inputFormat.createInputSplits(1)).thenReturn(new InputSplit[]
{inputSplit});
+ IOException openException = new IOException("expected open failure");
+ HoodieIOException splitCloseException =
+ new HoodieIOException("expected runtime split close failure");
+ doThrow(openException).when(inputFormat).open(inputSplit);
+ doThrow(splitCloseException).when(inputFormat).close();
+
+ HoodieLookupTableReader reader =
+ new HoodieLookupTableReader(() -> inputFormat, new Configuration());
+
+ IOException exception = assertThrows(IOException.class, reader::open);
+ assertSame(openException, exception);
+ assertEquals(1, exception.getSuppressed().length);
+ assertSame(splitCloseException, exception.getSuppressed()[0]);
+ verify(inputFormat).closeInputFormat();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testCloseReleasesInputFormatWhenRuntimeSplitCloseFails() throws
Exception {
+ RichInputFormat<RowData, InputSplit> inputFormat =
mock(RichInputFormat.class);
+ InputSplit inputSplit = mock(InputSplit.class);
+ when(inputFormat.createInputSplits(1)).thenReturn(new InputSplit[]
{inputSplit});
+ HoodieIOException splitCloseException =
+ new HoodieIOException("expected runtime split close failure");
+ IOException formatCloseException = new IOException("expected format close
failure");
+ doThrow(splitCloseException).when(inputFormat).close();
+ doThrow(formatCloseException).when(inputFormat).closeInputFormat();
+
+ HoodieLookupTableReader reader =
+ new HoodieLookupTableReader(() -> inputFormat, new Configuration());
+ reader.open();
+
+ HoodieIOException exception = assertThrows(HoodieIOException.class,
reader::close);
+ assertSame(splitCloseException, exception);
+ assertEquals(1, exception.getSuppressed().length);
+ assertSame(formatCloseException, exception.getSuppressed()[0]);
+ verify(inputFormat).closeInputFormat();
+ }
+}
