You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/06/23 16:10:00 UTC
(flink) branch master updated (bf6333cfa22 -> df5bdc6b31f)
This is an automated email from the ASF dual-hosted git repository.
ron pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from bf6333cfa22 [FLINK-37978][table] Change buffer-capacity to max-concurrent-operations for ml_predict (#26706)
new c62878a3aad Revert "[Flink-37703][hadoop] Add more log info to locate the root cause of failed HadoopRecoverableWriterTest#testRecoverFromIntermWithoutAdditionalState in azure cron pipeline (#26646)"
new df5bdc6b31f [FLINK-37703][hadoop] Splitting the non-local write unit test to seperated class to fix the unstable tests
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../core/fs/AbstractRecoverableWriterTest.java | 47 +++++----------
...WriterTest.java => HadoopNoLocalWriteTest.java} | 39 +++++--------
.../fs/hdfs/HadoopRecoverableWriterTest.java | 66 +---------------------
.../src/test/resources/log4j2-test.properties | 10 +---
tools/maven/suppressions.xml | 2 +-
5 files changed, 29 insertions(+), 135 deletions(-)
copy flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/{HadoopRecoverableWriterTest.java => HadoopNoLocalWriteTest.java} (94%)
(flink) 01/02: Revert "[Flink-37703][hadoop] Add more log info to locate the root cause of failed HadoopRecoverableWriterTest#testRecoverFromIntermWithoutAdditionalState in azure cron pipeline (#26646)"
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c62878a3aadf109ab664b28ea920e0a10728f97b
Author: 余良 <yu...@chinaunicom.cn>
AuthorDate: Mon Jun 23 14:15:08 2025 +0800
Revert "[Flink-37703][hadoop] Add more log info to locate the root cause of failed HadoopRecoverableWriterTest#testRecoverFromIntermWithoutAdditionalState in azure cron pipeline (#26646)"
This reverts commit e7e1bb9704f6b39ae44758c534a9cb0ee207d42c.
---
.../core/fs/AbstractRecoverableWriterTest.java | 47 ++++++----------------
.../src/test/resources/log4j2-test.properties | 10 +----
2 files changed, 14 insertions(+), 43 deletions(-)
diff --git a/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java b/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
index 7cefb7e5bf8..949b67f594d 100644
--- a/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/fs/AbstractRecoverableWriterTest.java
@@ -25,8 +25,6 @@ import org.apache.flink.util.StringUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
@@ -44,8 +42,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
*/
public abstract class AbstractRecoverableWriterTest {
- private static final Logger LOG = LoggerFactory.getLogger(AbstractRecoverableWriterTest.class);
-
private static final Random RND = new Random();
private static final String testData1 = "THIS IS A TEST 1.";
@@ -200,29 +196,16 @@ public abstract class AbstractRecoverableWriterTest {
final Map<String, RecoverableWriter.ResumeRecoverable> recoverables = new HashMap<>(4);
RecoverableFsDataOutputStream stream = null;
try {
- // This is just to provide diagnostics to locate the root cause:
- // https://issues.apache.org/jira/browse/FLINK-37703
- // After the fix, this logic should be reverted.
- int times = 0;
- try {
- times++;
- stream = initWriter.open(path);
- recoverables.put(INIT_EMPTY_PERSIST, stream.persist());
-
- times++;
- stream.write(testData1.getBytes(StandardCharsets.UTF_8));
-
- recoverables.put(INTERM_WITH_STATE_PERSIST, stream.persist());
- recoverables.put(INTERM_WITH_NO_ADDITIONAL_STATE_PERSIST, stream.persist());
-
- // and write some more data
- times++;
- stream.write(testData2.getBytes(StandardCharsets.UTF_8));
-
- } catch (IOException e) {
- LOG.warn("{} execution failed, err message{}: ", times, e.getMessage());
- throw e;
- }
+ stream = initWriter.open(path);
+ recoverables.put(INIT_EMPTY_PERSIST, stream.persist());
+
+ stream.write(testData1.getBytes(StandardCharsets.UTF_8));
+
+ recoverables.put(INTERM_WITH_STATE_PERSIST, stream.persist());
+ recoverables.put(INTERM_WITH_NO_ADDITIONAL_STATE_PERSIST, stream.persist());
+
+ // and write some more data
+ stream.write(testData2.getBytes(StandardCharsets.UTF_8));
recoverables.put(FINAL_WITH_EXTRA_STATE, stream.persist());
} finally {
@@ -253,13 +236,9 @@ public abstract class AbstractRecoverableWriterTest {
assertThat(fileContents.getKey().getName()).startsWith(".part-0.inprogress.");
assertThat(fileContents.getValue()).isEqualTo(expectedPostRecoveryContents);
}
- try {
- recoveredStream.write(testData3.getBytes(StandardCharsets.UTF_8));
- recoveredStream.closeForCommit().commit();
- } catch (IOException e) {
- LOG.warn("Final write failed: {}", e.getMessage());
- throw e;
- }
+
+ recoveredStream.write(testData3.getBytes(StandardCharsets.UTF_8));
+ recoveredStream.closeForCommit().commit();
files = getFileContentByPath(testDir);
assertThat(files).hasSize(1);
diff --git a/flink-filesystems/flink-hadoop-fs/src/test/resources/log4j2-test.properties b/flink-filesystems/flink-hadoop-fs/src/test/resources/log4j2-test.properties
index cc0425b120f..835c2ec9a3d 100644
--- a/flink-filesystems/flink-hadoop-fs/src/test/resources/log4j2-test.properties
+++ b/flink-filesystems/flink-hadoop-fs/src/test/resources/log4j2-test.properties
@@ -18,11 +18,7 @@
# Set root logger level to OFF to not flood build logs
# set manually to INFO for debugging purposes
-# ----------------------
-# This is just for locate the root cause:
-# https://issues.apache.org/jira/browse/FLINK-37703
-# After the fix, this logic should be reverted.
-rootLogger.level = WARN
+rootLogger.level = OFF
rootLogger.appenderRef.test.ref = TestLogger
appender.testlogger.name = TestLogger
@@ -30,7 +26,3 @@ appender.testlogger.type = CONSOLE
appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
-
-logger.DataStreamer.name = org.apache.hadoop.hdfs.DataStreamer
-logger.DataStreamer.level = DEBUG
-
(flink) 02/02: [FLINK-37703][hadoop] Splitting the non-local write unit test to seperated class to fix the unstable tests
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit df5bdc6b31fd9c42b0ae712a1e1eb61d630f9e49
Author: 余良 <yu...@chinaunicom.cn>
AuthorDate: Mon Jun 23 16:44:34 2025 +0800
[FLINK-37703][hadoop] Splitting the non-local write unit test to seperated class to fix the unstable tests
---
...WriterTest.java => HadoopNoLocalWriteTest.java} | 39 +++++--------
.../fs/hdfs/HadoopRecoverableWriterTest.java | 66 +---------------------
tools/maven/suppressions.xml | 2 +-
3 files changed, 15 insertions(+), 92 deletions(-)
diff --git a/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java b/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopNoLocalWriteTest.java
similarity index 94%
copy from flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java
copy to flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopNoLocalWriteTest.java
index 3a045992163..b6827e2d4e5 100644
--- a/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java
+++ b/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopNoLocalWriteTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.fs.hdfs;
-import org.apache.flink.core.fs.AbstractRecoverableWriterTest;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
@@ -51,8 +50,7 @@ import static org.assertj.core.api.Assumptions.assumeThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
-/** Tests for the {@link HadoopRecoverableWriter}. */
-class HadoopRecoverableWriterTest extends AbstractRecoverableWriterTest {
+public class HadoopNoLocalWriteTest {
@TempDir private static java.nio.file.Path tempFolder;
@@ -87,27 +85,10 @@ class HadoopRecoverableWriterTest extends AbstractRecoverableWriterTest {
hdfsCluster = builder.build();
final org.apache.hadoop.fs.FileSystem hdfs = hdfsCluster.getFileSystem();
-
fileSystem = new HadoopFileSystem(hdfs);
basePath = new Path(hdfs.getUri() + "/tests");
}
- @AfterAll
- static void destroyHDFS() throws Exception {
- if (hdfsCluster != null) {
- hdfsCluster
- .getFileSystem()
- .delete(new org.apache.hadoop.fs.Path(basePath.toUri()), true);
- hdfsCluster.shutdown();
- }
- }
-
- private RecoverableWriter getNoLocalWriteFileSystemWriter() throws Exception {
- Map<String, String> conf = new HashMap<>();
- conf.put("fs.hdfs.no-local-write", "true");
- return fileSystem.createRecoverableWriter(conf);
- }
-
@Test
void testNoLocalWriteFlag() throws Exception {
createHDFS();
@@ -147,13 +128,19 @@ class HadoopRecoverableWriterTest extends AbstractRecoverableWriterTest {
Assertions.assertEquals(1, 3 - numDataNodesWithData);
}
- @Override
- public Path getBasePath() {
- return basePath;
+ @AfterAll
+ static void destroyHDFS() throws Exception {
+ if (hdfsCluster != null) {
+ hdfsCluster
+ .getFileSystem()
+ .delete(new org.apache.hadoop.fs.Path(basePath.toUri()), true);
+ hdfsCluster.shutdown();
+ }
}
- @Override
- public FileSystem initializeFileSystem() {
- return fileSystem;
+ private RecoverableWriter getNoLocalWriteFileSystemWriter() throws Exception {
+ Map<String, String> conf = new HashMap<>();
+ conf.put("fs.hdfs.no-local-write", "true");
+ return fileSystem.createRecoverableWriter(conf);
}
}
diff --git a/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java b/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java
index 3a045992163..e5a4ccafb9a 100644
--- a/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java
+++ b/flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/HadoopRecoverableWriterTest.java
@@ -21,35 +21,19 @@ package org.apache.flink.runtime.fs.hdfs;
import org.apache.flink.core.fs.AbstractRecoverableWriterTest;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
-import org.apache.flink.core.fs.RecoverableWriter;
-import org.apache.flink.mock.Whitebox;
import org.apache.flink.runtime.util.HadoopUtils;
import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.OperatingSystem;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
-import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
-import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
-import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
-import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.File;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Random;
import static org.assertj.core.api.Assumptions.assumeThat;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.spy;
/** Tests for the {@link HadoopRecoverableWriter}. */
class HadoopRecoverableWriterTest extends AbstractRecoverableWriterTest {
@@ -79,11 +63,8 @@ class HadoopRecoverableWriterTest extends AbstractRecoverableWriterTest {
final Configuration hdConf = new Configuration();
hdConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath());
- hdConf.set("dfs.replication", "2");
- hdConf.set("dfs.blocksize", String.valueOf(512));
- hdConf.set("dfs.namenode.fs-limits.min-block-size", String.valueOf(512));
- final MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf).numDataNodes(3);
+ final MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(hdConf);
hdfsCluster = builder.build();
final org.apache.hadoop.fs.FileSystem hdfs = hdfsCluster.getFileSystem();
@@ -102,51 +83,6 @@ class HadoopRecoverableWriterTest extends AbstractRecoverableWriterTest {
}
}
- private RecoverableWriter getNoLocalWriteFileSystemWriter() throws Exception {
- Map<String, String> conf = new HashMap<>();
- conf.put("fs.hdfs.no-local-write", "true");
- return fileSystem.createRecoverableWriter(conf);
- }
-
- @Test
- void testNoLocalWriteFlag() throws Exception {
- createHDFS();
- final HadoopRecoverableWriter writer =
- (HadoopRecoverableWriter) getNoLocalWriteFileSystemWriter();
- BlockManager bm = hdfsCluster.getNameNode().getNamesystem().getBlockManager();
- DatanodeManager dm = bm.getDatanodeManager();
- try (RecoverableFsDataOutputStream os = writer.open(new Path("/tests/test-no-local"))) {
- // Inject a DatanodeManager that returns one DataNode as local node for
- // the client.
- DatanodeManager spyDm = spy(dm);
- DatanodeDescriptor dn1 =
- dm.getDatanodeListForReport(HdfsConstants.DatanodeReportType.LIVE).get(0);
- doReturn(dn1).when(spyDm).getDatanodeByHost("127.0.0.1");
- Whitebox.setInternalState(bm, "datanodeManager", spyDm);
- byte[] buf = new byte[512 * 16];
- new Random().nextBytes(buf);
- os.write(buf);
- } finally {
- Whitebox.setInternalState(bm, "datanodeManager", dm);
- }
- hdfsCluster.triggerBlockReports();
- final String bpid = hdfsCluster.getNamesystem().getBlockPoolId();
- // Total number of DataNodes is 3.
- Assertions.assertEquals(3, hdfsCluster.getAllBlockReports(bpid).size());
- int numDataNodesWithData = 0;
- for (Map<DatanodeStorage, BlockListAsLongs> dnBlocks :
- hdfsCluster.getAllBlockReports(bpid)) {
- for (BlockListAsLongs blocks : dnBlocks.values()) {
- if (blocks.getNumberOfBlocks() > 0) {
- numDataNodesWithData++;
- break;
- }
- }
- }
- // Verify that only one DN has no data.
- Assertions.assertEquals(1, 3 - numDataNodesWithData);
- }
-
@Override
public Path getBasePath() {
return basePath;
diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml
index 95a931c4dcb..b9a8c129d52 100644
--- a/tools/maven/suppressions.xml
+++ b/tools/maven/suppressions.xml
@@ -43,7 +43,7 @@ under the License.
<!-- Legacy mockito usages -->
<!-- Updated this suppress files section due to FLINK-36063, which migrated many classes from flink-streaming-java to flink-runtime. This allows runtime access to stream graph. -->
- <suppress files="BufferDataOverWindowOperatorTest.java|CEPOperatorTest.java|CepRuntimeContextTest.java|CliFrontendListTest.java|CliFrontendPackageProgramTest.java|CliFrontendSavepointTest.java|DemultiplexingRecordDeserializerTest.java|DropwizardMeterWrapperTest.java|DynamicEventTimeSessionWindowsTest.java|DynamicProcessingTimeSessionWindowsTest.java|EmbeddedRocksDBStateBackendTest.java|EventTimeSessionWindowsTest.java|FlinkCalciteCatalogReaderTest.java|FlinkMeterWrapperTest.java|Hadoop [...]
+ <suppress files="BufferDataOverWindowOperatorTest.java|CEPOperatorTest.java|CepRuntimeContextTest.java|CliFrontendListTest.java|CliFrontendPackageProgramTest.java|CliFrontendSavepointTest.java|DemultiplexingRecordDeserializerTest.java|DropwizardMeterWrapperTest.java|DynamicEventTimeSessionWindowsTest.java|DynamicProcessingTimeSessionWindowsTest.java|EmbeddedRocksDBStateBackendTest.java|EventTimeSessionWindowsTest.java|FlinkCalciteCatalogReaderTest.java|FlinkMeterWrapperTest.java|Hadoop [...]
checks="IllegalImport"/>
<suppress files="org[\\/]apache[\\/]flink[\\/]formats[\\/]avro[\\/]generated[\\/].*.java" checks="[a-zA-Z0-9]*"/>