Merge branch 'master' into bvolpato/bump-postgres-jdbc-42.7.10

Author: Abacn

.github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 4
+    "modification": 1
 }

.github/workflows/finalize_release.yml

@@ -80,6 +80,12 @@ jobs:
           docker buildx imagetools create --tag "${IMAGE}:latest" "${IMAGE}:${RELEASE}"
         done
 
+        # Flink 2 images have different tagging, add latest tag to latest supported Flink version
+        echo "================Confirming Runner container Release and RC version==========="
+        BEAM_FLINK_REPO=apache/beam_flink_job_server
+        LATEST_FLINK_VERSION=$(wget -qO- https://raw.githubusercontent.com/apache/beam/refs/tags/v${RELEASE}-RC${RC_NUM}/gradle.properties | grep -E flink_versions |  tr ',' '\n' | tail -1)
+        docker buildx imagetools create --tag "${BEAM_FLINK_REPO}:latest" "${BEAM_FLINK_REPO}:${RELEASE}${RC_VERSION}-flink${LATEST_FLINK_VERSION}"
+
   publish_python_artifacts:
     if: ${{github.event.inputs.PUBLISH_PYTHON_ARTIFACTS == 'yes'}}
     runs-on: [self-hosted, ubuntu-20.04, main]

CHANGES.md

@@ -88,6 +88,7 @@
 ## Security Fixes
 
 * Fixed [CVE-2024-1597](https://www.cve.org/CVERecord?id=CVE-2024-1597), [CVE-2022-31197](https://www.cve.org/CVERecord?id=CVE-2022-31197), and [CVE-2022-21724](https://www.cve.org/CVERecord?id=CVE-2022-21724) by upgrading PostgreSQL JDBC Driver from 42.2.16 to 42.7.10 (Java) ([#37942](https://github.com/apache/beam/issues/37942)).
+* Fixed [CVE-2023-46604](https://www.cve.org/CVERecord?id=CVE-2023-46604) (CVSS 10.0) and [CVE-2022-41678](https://www.cve.org/CVERecord?id=CVE-2022-41678) by upgrading ActiveMQ from 5.14.5 to 5.19.2 (Java) ([#37943](https://github.com/apache/beam/issues/37943)).
 
 ## Known Issues
 
@@ -2386,4 +2387,4 @@ Schema Options, it will be removed in version `2.23.0`. ([BEAM-9704](https://iss
 
 ## Highlights
 
-- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/).
+- For versions 2.19.0 and older release notes are available on [Apache Beam Blog](https://beam.apache.org/blog/).

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamDockerPlugin.groovy

@@ -48,6 +48,7 @@ class BeamDockerPlugin implements Plugin<Project> {
     String dockerComposeFile = 'docker-compose.yml'
     Set<Task> dependencies = [] as Set
     Set<String> tags = [] as Set
+    String tagSuffix = null
     Map<String, String> namedTags = [:]
     Map<String, String> labels = [:]
     Map<String, String> buildArgs = [:]
@@ -100,7 +101,11 @@ class BeamDockerPlugin implements Plugin<Project> {
     }
 
     Set<String> getTags() {
-      return this.tags + project.getVersion().toString()
+      def allTags = this.tags + project.getVersion().toString()
+      if (tagSuffix) {
+        allTags = allTags.collect { it.endsWith(tagSuffix) ? it : it + tagSuffix }.toSet()
+      }
+      return allTags
     }
 
     Set<String> getPlatform() {

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -597,7 +597,7 @@ class BeamModulePlugin implements Plugin<Project> {
     //
     // There are a few versions are determined by the BOMs by running scripts/tools/bomupgrader.py
     // marked as [bomupgrader]. See the documentation of that script for detail.
-    def activemq_version = "5.14.5"
+    def activemq_version = "5.19.2"
     def autovalue_version = "1.9"
     def autoservice_version = "1.0.1"
     def aws_java_sdk2_version = "2.20.162"

runners/flink/job-server-container/flink_job_server_container.gradle

@@ -55,8 +55,10 @@ task copyDockerfileDependencies(type: Copy) {
 def pushContainers = project.rootProject.hasProperty(["isRelease"]) || project.rootProject.hasProperty("push-containers")
 def containerName = project.parent.name.startsWith("2") ? "flink_job_server" : "flink${project.parent.name}_job_server"
 def containerTag = project.rootProject.hasProperty(["docker-tag"]) ? project.rootProject["docker-tag"] : project.sdk_version
+String verInSuffix = null
 if (project.parent.name.startsWith("2")) {
   containerTag += "-flink${project.parent.name}"
+  verInSuffix = "-flink${project.parent.name}"
 }
 
 docker {
@@ -68,6 +70,7 @@ docker {
           tag: containerTag)
   // tags used by dockerTag task
   tags containerImageTags()
+  tagSuffix verInSuffix
   files "./build/"
   buildx project.useBuildx()
   platform(*project.containerPlatforms())

sdks/java/core/src/main/java/org/apache/beam/sdk/util/StreamUtils.java

@@ -22,6 +22,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.ref.SoftReference;
+import java.util.Arrays;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.annotations.Internal;
 
 /** Utility functions for stream operations. */
@@ -35,34 +37,67 @@ private StreamUtils() {}
 
   private static final int BUF_SIZE = 8192;
 
-  private static ThreadLocal<SoftReference<byte[]>> threadLocalBuffer = new ThreadLocal<>();
+  private static final ThreadLocal<SoftReference<byte[]>> threadLocalBuffer = new ThreadLocal<>();
 
   /** Efficient converting stream to bytes. */
   public static byte[] getBytesWithoutClosing(InputStream stream) throws IOException {
+    // Unwrap the stream so the below optimizations based upon class type function properly.
+    // We don't use mark or reset in this function.
+    while (stream instanceof UnownedInputStream) {
+      stream = ((UnownedInputStream) stream).getWrappedStream();
+    }
+
     if (stream instanceof ExposedByteArrayInputStream) {
       // Fast path for the exposed version.
       return ((ExposedByteArrayInputStream) stream).readAll();
-    } else if (stream instanceof ByteArrayInputStream) {
+    }
+    if (stream instanceof ByteArrayInputStream) {
       // Fast path for ByteArrayInputStream.
       byte[] ret = new byte[stream.available()];
       stream.read(ret);
       return ret;
     }
-    // Falls back to normal stream copying.
+
+    // Most inputs are fully available so we attempt to first read directly
+    // into a buffer of the right size, assuming available reflects all the bytes.
+    int available = stream.available();
+    @Nullable ByteArrayOutputStream outputStream = null;
+    if (available > 0 && available < 1024 * 1024) {
+      byte[] initialBuffer = new byte[available];
+      int initialReadSize = stream.read(initialBuffer);
+      if (initialReadSize == -1) {
+        return new byte[0];
+      }
+      int nextByte = stream.read();
+      if (nextByte == -1) {
+        if (initialReadSize == available) {
+          // Available reflected the full buffer and we copied directly to the
+          // right size.
+          return initialBuffer;
+        }
+        return Arrays.copyOf(initialBuffer, initialReadSize);
+      }
+      outputStream = new ByteArrayOutputStream();
+      outputStream.write(initialBuffer, 0, initialReadSize);
+      outputStream.write(nextByte);
+    } else {
+      outputStream = new ByteArrayOutputStream();
+    }
+
+    // Normal stream copying using the thread-local buffer.
     SoftReference<byte[]> refBuffer = threadLocalBuffer.get();
     byte[] buffer = refBuffer == null ? null : refBuffer.get();
     if (buffer == null) {
       buffer = new byte[BUF_SIZE];
       threadLocalBuffer.set(new SoftReference<>(buffer));
     }
-    ByteArrayOutputStream outStream = new ByteArrayOutputStream();
     while (true) {
       int r = stream.read(buffer);
       if (r == -1) {
         break;
       }
-      outStream.write(buffer, 0, r);
+      outputStream.write(buffer, 0, r);
     }
-    return outStream.toByteArray();
+    return outputStream.toByteArray();
   }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/util/UnownedInputStream.java

@@ -35,6 +35,10 @@ public UnownedInputStream(InputStream delegate) {
     super(delegate);
   }
 
+  InputStream getWrappedStream() {
+    return in;
+  }
+
   @Override
   public void close() throws IOException {
     throw new UnsupportedOperationException(

sdks/java/core/src/test/java/org/apache/beam/sdk/util/CoderUtilsTest.java

@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.util;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -26,7 +27,9 @@
 
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
 import org.apache.beam.sdk.coders.AtomicCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.Coder.Context;
 import org.apache.beam.sdk.coders.CoderException;
@@ -142,4 +145,11 @@ public void testDecodeFromByteStringWithExtraDataThrows() throws Exception {
         CoderException.class,
         () -> CoderUtils.decodeFromByteString(StringUtf8Coder.of(), byteString, Context.NESTED));
   }
+
+  @Test
+  public void testDecodeByteArrayWithoutCopy() throws Exception {
+    byte[] data = "test data".getBytes(StandardCharsets.UTF_8);
+    byte[] result = CoderUtils.decodeFromByteArray(ByteArrayCoder.of(), data);
+    assertSame(data, result);
+  }
 }

sdks/java/core/src/test/java/org/apache/beam/sdk/util/StreamUtilsTest.java

@@ -23,9 +23,11 @@
 
 import java.io.BufferedInputStream;
 import java.io.ByteArrayInputStream;
+import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.ByteStreams;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -68,4 +70,97 @@ public void testGetBytesFromInputStream() throws IOException {
     assertArrayEquals(testData, bytes);
     assertEquals(0, stream.available());
   }
+
+  @Test
+  public void testGetBytesFromUnownedInputStreamAroundExposed() throws IOException {
+    InputStream stream = new UnownedInputStream(new ExposedByteArrayInputStream(testData));
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertSame(testData, bytes);
+    assertEquals(0, stream.available());
+  }
+
+  @Test
+  public void testGetBytesFromUnownedInputStreamAroundArray() throws IOException {
+    InputStream stream = new UnownedInputStream(new ByteArrayInputStream(testData));
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertEquals(0, stream.available());
+  }
+
+  @Test
+  public void testGetBytesFromLimitedInputStream() throws IOException {
+    InputStream stream = ByteStreams.limit(new ByteArrayInputStream(testData), Integer.MAX_VALUE);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertEquals(0, stream.available());
+  }
+
+  @Test
+  public void testGetBytesFromEmptyLimitedInputStream() throws IOException {
+    InputStream stream = ByteStreams.limit(new ByteArrayInputStream(testData), 0);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(new byte[0], bytes);
+    assertEquals(0, stream.available());
+  }
+
+  @Test
+  public void testGetBytesFromRepeatedInputStream() throws IOException {
+    byte[] largeBytes = new byte[2 * 1024 * 1024];
+    Arrays.fill(largeBytes, (byte) 1);
+    InputStream stream = ByteStreams.limit(new ByteArrayInputStream(largeBytes), Integer.MAX_VALUE);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(largeBytes, bytes);
+    assertEquals(0, stream.available());
+  }
+
+  public static class LyingInputStream extends FilterInputStream {
+    private final int availableLie;
+
+    public LyingInputStream(InputStream in, int availableLie) {
+      super(in);
+      this.availableLie = availableLie;
+    }
+
+    @Override
+    public int available() throws IOException {
+      return availableLie;
+    }
+  }
+
+  @Test
+  public void testGetBytesFromHugeAvailable() throws IOException {
+    InputStream wrappedStream = new ByteArrayInputStream(testData);
+    InputStream stream = new LyingInputStream(wrappedStream, Integer.MAX_VALUE - 1);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertEquals(0, wrappedStream.available());
+  }
+
+  @Test
+  public void testGetBytesFromZeroAvailable() throws IOException {
+    InputStream wrappedStream = new ByteArrayInputStream(testData);
+    InputStream stream = new LyingInputStream(wrappedStream, 0);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertEquals(0, wrappedStream.available());
+  }
+
+  @Test
+  public void testGetBytesFromOneExtraAvailable() throws IOException {
+    InputStream wrappedStream = new ByteArrayInputStream(testData);
+    InputStream stream = new LyingInputStream(wrappedStream, wrappedStream.available() + 1);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertEquals(0, wrappedStream.available());
+  }
+
+  @Test
+  public void testGetBytesFromOneLessAvailable() throws IOException {
+    InputStream wrappedStream = new ByteArrayInputStream(testData);
+    InputStream stream = new LyingInputStream(wrappedStream, wrappedStream.available() - 1);
+    byte[] bytes = StreamUtils.getBytesWithoutClosing(stream);
+    assertArrayEquals(testData, bytes);
+    assertEquals(0, wrappedStream.available());
+  }
 }