You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/12/21 09:32:50 UTC
(pulsar) branch branch-4.0 updated (948b33a3ff0 -> d3d8e507980)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 948b33a3ff0 [Fix][Client] Fix pending message not complete when closeAsync (#23761)
new 2117be75bce [fix][test] Fix flaky KeySharedSubscriptionTest.testNoKeySendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector (#23747)
new 0f15a3667e1 [fix][broker] Fix bug causing loss of migrated information when setting other localPolicies in namespace (#23764)
new 8700d1c2d7e [fix][client] Fix reader message filtering issue during blue-green cluster switch (#23693)
new f961e7572e8 [fix][client] Fix enableRetry for consumers using legacy topic naming where cluster name is included (#23753)
new d3d8e507980 [fix] Fix issues with Pulsar Alpine docker image stability: remove glibc-compat (#23762)
The 5 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:
docker/glibc-package/Dockerfile | 80 ------------
docker/glibc-package/scripts/APKBUILD | 53 --------
docker/glibc-package/scripts/glibc-bin.trigger | 21 ----
docker/glibc-package/scripts/ld.so.conf | 23 ----
docker/kinesis-producer-alpine/Dockerfile | 90 +++++++++++++
.../README.md | 14 +--
docker/kinesis-producer-alpine/build-alpine.sh | 139 +++++++++++++++++++++
.../kinesis_producer_alpine.patch | 127 +++++++++++++++++++
docker/pulsar-all/Dockerfile | 38 +++++-
docker/pulsar-all/pom.xml | 5 +
docker/pulsar/Dockerfile | 11 +-
pom.xml | 3 +-
.../pulsar/broker/admin/impl/NamespacesBase.java | 20 +--
.../pulsar/common/naming/NamespaceBundles.java | 3 +-
.../apache/pulsar/broker/admin/NamespacesTest.java | 27 ++++
.../broker/service/ClusterMigrationTest.java | 111 ++++++++++++++++
.../client/api/KeySharedSubscriptionTest.java | 5 +-
.../pulsar/client/impl/TopicsConsumerImplTest.java | 18 +++
.../pulsar/client/impl/ConsumerBuilderImpl.java | 8 +-
.../apache/pulsar/client/impl/ConsumerImpl.java | 7 ++
.../pulsar/common/policies/data/LocalPolicies.java | 11 +-
pulsar-io/kinesis/pom.xml | 2 +-
.../org/apache/pulsar/io/kinesis/KinesisSink.java | 8 +-
.../pulsar/io/kinesis/KinesisSinkConfig.java | 23 ++++
.../apache/pulsar/io/kinesis/KinesisSinkTest.java | 37 +++---
.../docker-images/latest-version-image/Dockerfile | 24 +---
tests/docker-images/latest-version-image/pom.xml | 1 -
.../integration/io/sinks/KinesisSinkTester.java | 13 +-
28 files changed, 664 insertions(+), 258 deletions(-)
delete mode 100644 docker/glibc-package/Dockerfile
delete mode 100644 docker/glibc-package/scripts/APKBUILD
delete mode 100755 docker/glibc-package/scripts/glibc-bin.trigger
delete mode 100644 docker/glibc-package/scripts/ld.so.conf
create mode 100644 docker/kinesis-producer-alpine/Dockerfile
rename docker/{glibc-package => kinesis-producer-alpine}/README.md (54%)
create mode 100644 docker/kinesis-producer-alpine/build-alpine.sh
create mode 100644 docker/kinesis-producer-alpine/kinesis_producer_alpine.patch
(pulsar) 01/05: [fix][test] Fix flaky KeySharedSubscriptionTest.testNoKeySendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector (#23747)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2117be75bceb97e0e8c64c715e7ba94eb170a565
Author: Philipp Dolif <52...@users.noreply.github.com>
AuthorDate: Fri Dec 20 12:39:05 2024 +0100
[fix][test] Fix flaky KeySharedSubscriptionTest.testNoKeySendAndReceiveWithHashRangeAutoSplitStickyKeyConsumerSelector (#23747)
(cherry picked from commit a54be73e95e02f12e6656acb46c339c81dc0e531)
---
.../java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
index 08efb6d9583..92257c1df53 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java
@@ -391,13 +391,14 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase {
@Cleanup
Producer<Integer> producer = createProducer(topic, enableBatch);
- for (int i = 0; i < 100; i++) {
+ int totalMessages = 300;
+ for (int i = 0; i < totalMessages; i++) {
producer.newMessage()
.value(i)
.send();
}
- receiveAndCheckDistribution(Lists.newArrayList(consumer1, consumer2, consumer3), 100);
+ receiveAndCheckDistribution(Lists.newArrayList(consumer1, consumer2, consumer3), totalMessages);
}
@Test(dataProvider = "batch")
(pulsar) 04/05: [fix][client] Fix enableRetry for consumers using legacy topic naming where cluster name is included (#23753)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f961e7572e83347b7e1201237e6c06fad45b13e4
Author: crossoverJie <cr...@gmail.com>
AuthorDate: Fri Dec 20 20:19:16 2024 +0800
[fix][client] Fix enableRetry for consumers using legacy topic naming where cluster name is included (#23753)
(cherry picked from commit 217ebfbeaab9a33e648912bfae8ed47e9199d41a)
---
.../pulsar/client/impl/TopicsConsumerImplTest.java | 18 ++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerBuilderImpl.java | 8 ++++----
2 files changed, 22 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
index 1d5ac759625..3c7cd16f144 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
@@ -138,6 +138,24 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
}
}
+ @Test(timeOut = testTimeout)
+ public void testRetryClusterTopic() throws Exception {
+ String key = "testRetryClusterTopic";
+ final String topicName = "persistent://prop/use/ns-abc1/topic-1-" + key;
+ TenantInfoImpl tenantInfo = createDefaultTenantInfo();
+ final String namespace = "prop/ns-abc1";
+ admin.tenants().createTenant("prop", tenantInfo);
+ admin.namespaces().createNamespace(namespace, Set.of("test"));
+ Consumer consumer = pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName("my-sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .enableRetry(true)
+ .ackTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS)
+ .subscribe();
+ assertTrue(consumer instanceof MultiTopicsConsumerImpl);
+ }
+
@Test(timeOut = testTimeout)
public void testGetConsumersAndGetTopics() throws Exception {
String key = "TopicsConsumerGet";
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
index 351025d426a..35f772028f1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -154,10 +154,10 @@ public class ConsumerBuilderImpl<T> implements ConsumerBuilder<T> {
if (conf.isRetryEnable() && conf.getTopicNames().size() > 0) {
TopicName topicFirst = TopicName.get(conf.getTopicNames().iterator().next());
//Issue 9327: do compatibility check in case of the default retry and dead letter topic name changed
- String oldRetryLetterTopic = topicFirst.getNamespace() + "/" + conf.getSubscriptionName()
- + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
- String oldDeadLetterTopic = topicFirst.getNamespace() + "/" + conf.getSubscriptionName()
- + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX;
+ String oldRetryLetterTopic = TopicName.get(topicFirst.getDomain().value(), topicFirst.getNamespaceObject(),
+ conf.getSubscriptionName() + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX).toString();
+ String oldDeadLetterTopic = TopicName.get(topicFirst.getDomain().value(), topicFirst.getNamespaceObject(),
+ conf.getSubscriptionName() + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX).toString();
DeadLetterPolicy deadLetterPolicy = conf.getDeadLetterPolicy();
if (deadLetterPolicy == null || StringUtils.isBlank(deadLetterPolicy.getRetryLetterTopic())
|| StringUtils.isBlank(deadLetterPolicy.getDeadLetterTopic())) {
(pulsar) 05/05: [fix] Fix issues with Pulsar Alpine docker image stability: remove glibc-compat (#23762)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d3d8e507980302064a9183a9df851398ded1e52d
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Dec 21 00:41:37 2024 -0800
[fix] Fix issues with Pulsar Alpine docker image stability: remove glibc-compat (#23762)
(cherry picked from commit 906d10e547abf8097dfd16e5564df3573246f6fc)
---
docker/glibc-package/Dockerfile | 80 ------------
docker/glibc-package/scripts/APKBUILD | 53 --------
docker/glibc-package/scripts/glibc-bin.trigger | 21 ----
docker/glibc-package/scripts/ld.so.conf | 23 ----
docker/kinesis-producer-alpine/Dockerfile | 90 +++++++++++++
.../README.md | 14 +--
docker/kinesis-producer-alpine/build-alpine.sh | 139 +++++++++++++++++++++
.../kinesis_producer_alpine.patch | 127 +++++++++++++++++++
docker/pulsar-all/Dockerfile | 38 +++++-
docker/pulsar-all/pom.xml | 5 +
docker/pulsar/Dockerfile | 11 +-
pom.xml | 3 +-
pulsar-io/kinesis/pom.xml | 2 +-
.../org/apache/pulsar/io/kinesis/KinesisSink.java | 8 +-
.../pulsar/io/kinesis/KinesisSinkConfig.java | 23 ++++
.../apache/pulsar/io/kinesis/KinesisSinkTest.java | 37 +++---
.../docker-images/latest-version-image/Dockerfile | 24 +---
tests/docker-images/latest-version-image/pom.xml | 1 -
.../integration/io/sinks/KinesisSinkTester.java | 13 +-
19 files changed, 469 insertions(+), 243 deletions(-)
diff --git a/docker/glibc-package/Dockerfile b/docker/glibc-package/Dockerfile
deleted file mode 100644
index 016e5c62236..00000000000
--- a/docker/glibc-package/Dockerfile
+++ /dev/null
@@ -1,80 +0,0 @@
-#
-# 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.
-#
-
-
-ARG GLIBC_VERSION=2.38
-ARG ALPINE_VERSION=3.20
-
-FROM ubuntu:22.04 as build
-ARG GLIBC_VERSION
-
-RUN apt-get -q update \
- && apt-get -qy install \
- bison \
- build-essential \
- gawk \
- gettext \
- openssl \
- python3 \
- texinfo \
- wget
-
-# Build GLibc
-RUN wget -qO- https://ftpmirror.gnu.org/libc/glibc-${GLIBC_VERSION}.tar.gz | tar zxf -
-RUN mkdir /glibc-build
-WORKDIR /glibc-build
-RUN /glibc-${GLIBC_VERSION}/configure \
- --prefix=/usr/glibc-compat \
- --libdir=/usr/glibc-compat/lib \
- --libexecdir=/usr/glibc-compat/lib \
- --enable-multi-arch \
- --enable-stack-protector=strong
-RUN make -j$(nproc)
-RUN make install
-RUN tar --dereference --hard-dereference -zcf /glibc-bin.tar.gz /usr/glibc-compat
-
-
-################################################
-## Build the APK package
-FROM alpine:$ALPINE_VERSION as apk
-ARG GLIBC_VERSION
-
-RUN apk add abuild sudo build-base
-
-RUN mkdir /build
-WORKDIR build
-
-COPY --from=build /glibc-bin.tar.gz /build
-
-COPY ./scripts /build
-
-RUN echo "pkgver=\"${GLIBC_VERSION}\"" >> /build/APKBUILD
-RUN echo "sha512sums=\"$(sha512sum glibc-bin.tar.gz ld.so.conf)\"" >> /build/APKBUILD
-
-RUN abuild-keygen -a -i -n
-RUN abuild -F -c -r
-
-################################################
-## Last stage - Only leaves the packages
-FROM busybox
-ARG GLIBC_VERSION
-
-RUN mkdir -p /root/packages
-COPY --from=apk /root/packages/*/glibc-${GLIBC_VERSION}-r0.apk /root/packages
-COPY --from=apk /root/packages/*/glibc-bin-${GLIBC_VERSION}-r0.apk /root/packages
diff --git a/docker/glibc-package/scripts/APKBUILD b/docker/glibc-package/scripts/APKBUILD
deleted file mode 100644
index 0545508f0a7..00000000000
--- a/docker/glibc-package/scripts/APKBUILD
+++ /dev/null
@@ -1,53 +0,0 @@
-#
-# 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.
-#
-
-pkgname="glibc"
-pkgrel="0"
-pkgdesc="GNU C Library compatibility layer"
-arch="all"
-url="https:/pulsar.apache.org"
-license="LGPL"
-options="!check"
-source="glibc-bin.tar.gz
-ld.so.conf"
-subpackages="${pkgname}-bin ${pkgname}-dev"
-triggers="glibc-bin.trigger=/lib:/usr/lib:/usr/glibc-compat/lib"
-depends="libuuid libgcc"
-
-package() {
- mkdir -p $pkgdir/lib $pkgdir/usr/glibc-compat/lib/locale $pkgdir/usr/glibc-compat/lib64 $pkgdir/etc $pkgdir/usr/glibc-compat/etc/
- cp -a $srcdir/usr $pkgdir
- cp $srcdir/ld.so.conf $pkgdir/usr/glibc-compat/etc/ld.so.conf
- cd $pkgdir/usr/glibc-compat
- rm -rf etc/rpc bin sbin lib/gconv lib/getconf lib/audit share var include
-
- FILENAME=$(ls $pkgdir/usr/glibc-compat/lib/ld-linux-*.so.*)
- LIBNAME=$(basename $FILENAME)
- ln -s /usr/glibc-compat/lib/$LIBNAME $pkgdir/lib/$LIBNAME
- ln -s /usr/glibc-compat/lib/$LIBNAME $pkgdir/usr/glibc-compat/lib64/$LIBNAME
- ln -s /usr/glibc-compat/etc/ld.so.cache $pkgdir/etc/ld.so.cache
-}
-
-bin() {
- depends="$pkgname libc6-compat"
- mkdir -p $subpkgdir/usr/glibc-compat
- cp -a $srcdir/usr/glibc-compat/bin $subpkgdir/usr/glibc-compat
- cp -a $srcdir/usr/glibc-compat/sbin $subpkgdir/usr/glibc-compat
-}
-
diff --git a/docker/glibc-package/scripts/glibc-bin.trigger b/docker/glibc-package/scripts/glibc-bin.trigger
deleted file mode 100755
index 5bae5d7ca2b..00000000000
--- a/docker/glibc-package/scripts/glibc-bin.trigger
+++ /dev/null
@@ -1,21 +0,0 @@
-#!/bin/sh
-#
-# 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.
-#
-
-/usr/glibc-compat/sbin/ldconfig
\ No newline at end of file
diff --git a/docker/glibc-package/scripts/ld.so.conf b/docker/glibc-package/scripts/ld.so.conf
deleted file mode 100644
index 6548b9300bb..00000000000
--- a/docker/glibc-package/scripts/ld.so.conf
+++ /dev/null
@@ -1,23 +0,0 @@
-#
-# 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.
-#
-
-/usr/local/lib
-/usr/glibc-compat/lib
-/usr/lib
-/lib
diff --git a/docker/kinesis-producer-alpine/Dockerfile b/docker/kinesis-producer-alpine/Dockerfile
new file mode 100644
index 00000000000..ffdf44f55d0
--- /dev/null
+++ b/docker/kinesis-producer-alpine/Dockerfile
@@ -0,0 +1,90 @@
+#
+# 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.
+#
+
+ARG ALPINE_VERSION=3.20
+
+# Builds an Alpine image with kinesis_producer compiled for Alpine Linux / musl
+
+# Build stage
+FROM alpine:$ALPINE_VERSION AS kinesis-producer-build
+ENV KINESIS_PRODUCER_LIB_VERSION=0.15.12
+
+# Install build dependencies
+RUN apk update && apk add --no-cache \
+ git \
+ binutils \
+ coreutils \
+ alpine-sdk \
+ util-linux \
+ cmake \
+ autoconf \
+ automake \
+ libtool \
+ curl \
+ bash \
+ tar \
+ libuuid \
+ linux-headers \
+ zlib \
+ zlib-dev \
+ perl \
+ wget \
+ boost-dev \
+ openssl-dev \
+ curl-dev \
+ build-base \
+ util-linux-dev \
+ g++ \
+ make \
+ upx
+
+ENV LANG=C.UTF-8
+
+RUN mkdir /build
+COPY kinesis_producer_alpine.patch /build/
+
+# Clone KPL and copy build script
+RUN cd /build && \
+ git clone --depth 1 --single-branch --branch v${KINESIS_PRODUCER_LIB_VERSION} https://github.com/awslabs/amazon-kinesis-producer && \
+ cd amazon-kinesis-producer && \
+ git apply ../kinesis_producer_alpine.patch
+
+# Copy and execute build script
+COPY build-alpine.sh /build/
+RUN chmod +x /build/build-alpine.sh
+RUN /build/build-alpine.sh
+
+# Final stage
+FROM alpine:$ALPINE_VERSION
+COPY --from=kinesis-producer-build /opt/amazon-kinesis-producer /opt/amazon-kinesis-producer
+RUN apk update && apk add --no-cache \
+ brotli-libs \
+ c-ares \
+ libcrypto3 \
+ libcurl \
+ libgcc \
+ libidn2 \
+ libpsl \
+ libssl3 \
+ libunistring \
+ nghttp2-libs \
+ zlib \
+ zstd-libs \
+ libuuid
+WORKDIR /opt/amazon-kinesis-producer/bin
diff --git a/docker/glibc-package/README.md b/docker/kinesis-producer-alpine/README.md
similarity index 54%
rename from docker/glibc-package/README.md
rename to docker/kinesis-producer-alpine/README.md
index ee1f643705a..4526f08c65e 100644
--- a/docker/glibc-package/README.md
+++ b/docker/kinesis-producer-alpine/README.md
@@ -19,21 +19,19 @@
-->
-# GLibc compatibility package
+# Alpine image with kinesis_producer compiled for Alpine Linux / musl
-This directory includes the Docker scripts to build an image with GLibc compiled for Alpine Linux.
+This directory includes the Docker scripts to build an image with `kinesis_producer` for Alpine Linux.
+`kinesis_producer` is a native executable that is required by [Amazon Kinesis Producer library (KPL)](https://github.com/awslabs/amazon-kinesis-producer) which is used by the Pulsar IO Kinesis Sink connector. The default `kinesis_producer` binary is compiled for glibc, and it does not work on Alpine Linux which uses musl.
-This is used to ensure plugins that are going to be used in the Pulsar image and that are depeding on GLibc, will
-still be working correctly in the Alpine Image. (eg: Netty Tc-Native and Kinesis Producer Library).
-
-This image only needs to be re-created when we want to upgrade to a newer version of GLibc.
+This image only needs to be re-created when we want to upgrade to a newer version of `kinesis_producer`.
# Steps
1. Change the version in the Dockerfile for this directory.
2. Rebuild the image and push it to Docker Hub:
```
-docker buildx build --platform=linux/amd64,linux/arm64 -t apachepulsar/glibc-base:2.38 . --push
+docker buildx build --platform=linux/amd64,linux/arm64 -t apachepulsar/pulsar-io-kinesis-sink-kinesis_producer:0.15.12 . --push
```
-The image tag is then used in `docker/pulsar/Dockerfile`.
+The image tag is then used in `docker/pulsar-all/Dockerfile`. The `kinesis_producer` binary is copied from the image to the `pulsar-all` image that is used by Pulsar Functions to run the Pulsar IO Kinesis Sink connector. The environment variable `PULSAR_IO_KINESIS_KPL_PATH` is set to `/opt/amazon-kinesis-producer/bin/kinesis_producer` and this is how the Kinesis Sink connector knows where to find the `kinesis_producer` binary.
\ No newline at end of file
diff --git a/docker/kinesis-producer-alpine/build-alpine.sh b/docker/kinesis-producer-alpine/build-alpine.sh
new file mode 100644
index 00000000000..23718450bbc
--- /dev/null
+++ b/docker/kinesis-producer-alpine/build-alpine.sh
@@ -0,0 +1,139 @@
+#!/bin/bash
+#
+# 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.
+#
+
+set -e
+set -x
+
+INSTALL_DIR=/build/third_party
+AWS_SDK_CPP_VERSION="1.11.420"
+PROTOBUF_VERSION="3.11.4"
+BOOST_VERSION="1.76.0"
+BOOST_VERSION_UNDERSCORED="${BOOST_VERSION//\./_}"
+
+# Create install directory
+mkdir -p $INSTALL_DIR
+
+# Setup environment variables
+export CC="gcc"
+export CXX="g++"
+export CXXFLAGS="-I$INSTALL_DIR/include -O3 -Wno-implicit-fallthrough -Wno-int-in-bool-context"
+export LDFLAGS="-L$INSTALL_DIR/lib"
+export LD_LIBRARY_PATH="$INSTALL_DIR/lib:$LD_LIBRARY_PATH"
+
+cd $INSTALL_DIR
+
+# Build protobuf
+if [ ! -d "protobuf-${PROTOBUF_VERSION}" ]; then
+ curl -LO https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOBUF_VERSION}/protobuf-all-${PROTOBUF_VERSION}.tar.gz
+ tar xf protobuf-all-${PROTOBUF_VERSION}.tar.gz
+ rm protobuf-all-${PROTOBUF_VERSION}.tar.gz
+
+ cd protobuf-${PROTOBUF_VERSION}
+ ./configure --prefix=${INSTALL_DIR} \
+ --disable-shared \
+ CFLAGS="-fPIC" \
+ CXXFLAGS="-fPIC ${CXXFLAGS}" \
+ --with-pic
+ make -j4
+ make install
+ cd ..
+fi
+
+# Build Boost
+if [ ! -d "boost_${BOOST_VERSION_UNDERSCORED}" ]; then
+ curl -LO https://boostorg.jfrog.io/artifactory/main/release/${BOOST_VERSION}/source/boost_${BOOST_VERSION_UNDERSCORED}.tar.gz
+ tar xf boost_${BOOST_VERSION_UNDERSCORED}.tar.gz
+ rm boost_${BOOST_VERSION_UNDERSCORED}.tar.gz
+
+ cd boost_${BOOST_VERSION_UNDERSCORED}
+
+ BOOST_LIBS="regex,thread,log,system,random,filesystem,chrono,atomic,date_time,program_options,test"
+
+ ./bootstrap.sh --with-libraries=$BOOST_LIBS --with-toolset=gcc
+
+ ./b2 \
+ -j4 \
+ variant=release \
+ link=static \
+ threading=multi \
+ runtime-link=static \
+ --prefix=${INSTALL_DIR} \
+ cxxflags="-fPIC ${CXXFLAGS}" \
+ install
+
+ cd ..
+fi
+
+# Download and build AWS SDK
+if [ ! -d "aws-sdk-cpp" ]; then
+ git clone --depth 1 --branch ${AWS_SDK_CPP_VERSION} https://github.com/awslabs/aws-sdk-cpp.git aws-sdk-cpp
+ pushd aws-sdk-cpp
+ git config submodule.fetchJobs 8
+ git submodule update --init --depth 1 --recursive
+ popd
+
+ rm -rf aws-sdk-cpp-build
+ mkdir aws-sdk-cpp-build
+ cd aws-sdk-cpp-build
+
+ cmake \
+ -DBUILD_ONLY="kinesis;monitoring;sts" \
+ -DCMAKE_BUILD_TYPE=RelWithDebInfo \
+ -DSTATIC_LINKING=1 \
+ -DCMAKE_PREFIX_PATH="$INSTALL_DIR" \
+ -DCMAKE_C_COMPILER="$CC" \
+ -DCMAKE_CXX_COMPILER="$CXX" \
+ -DCMAKE_CXX_FLAGS="$CXXFLAGS" \
+ -DCMAKE_INSTALL_PREFIX="$INSTALL_DIR" \
+ -DCMAKE_FIND_FRAMEWORK=LAST \
+ -DENABLE_TESTING="OFF" \
+ ../aws-sdk-cpp
+ make -j4
+ make install
+ cd ..
+fi
+
+# Build the native kinesis producer
+cd /build/amazon-kinesis-producer
+ln -fs ../third_party
+cmake -DCMAKE_PREFIX_PATH="$INSTALL_DIR" -DCMAKE_BUILD_TYPE=RelWithDebInfo .
+make -j4
+
+FINAL_DIR=/opt/amazon-kinesis-producer
+# copy the binary
+mkdir -p $FINAL_DIR/bin
+cp kinesis_producer $FINAL_DIR/bin/kinesis_producer.original
+
+# capture version information
+git describe --long --tags > $FINAL_DIR/bin/.version
+git rev-parse HEAD > $FINAL_DIR/bin/.revision
+uname -a > $FINAL_DIR/bin/.system_info
+cat /etc/os-release > $FINAL_DIR/bin/.os_info
+date > $FINAL_DIR/bin/.build_time
+
+# copy tests
+mkdir -p $FINAL_DIR/tests
+cp tests $FINAL_DIR/tests/
+cp test_driver $FINAL_DIR/tests/
+
+# Strip and compress the binary
+cd $FINAL_DIR/bin
+strip -o kinesis_producer.stripped kinesis_producer.original
+upx --best -o kinesis_producer kinesis_producer.stripped
\ No newline at end of file
diff --git a/docker/kinesis-producer-alpine/kinesis_producer_alpine.patch b/docker/kinesis-producer-alpine/kinesis_producer_alpine.patch
new file mode 100644
index 00000000000..d1ddd6a8550
--- /dev/null
+++ b/docker/kinesis-producer-alpine/kinesis_producer_alpine.patch
@@ -0,0 +1,127 @@
+From 96ba2eb7145363586529e6c770dcc0920bf04ac2 Mon Sep 17 00:00:00 2001
+From: Lari Hotari <lh...@users.noreply.github.com>
+Date: Wed, 18 Dec 2024 17:16:02 +0200
+Subject: [PATCH] Adapt build for Alpine, fix issue with NULL_BACKTRACE support
+
+- also use dynamic linking to some libraries (zlib, openssl, libz, libcurl, libcrypto)
+ to reduce binary size
+---
+ CMakeLists.txt | 20 +++++++++++---------
+ aws/utils/backtrace/bsd_backtrace.cc | 2 +-
+ aws/utils/backtrace/gcc_backtrace.cc | 3 +--
+ aws/utils/backtrace/null_backtrace.cc | 5 ++---
+ aws/utils/signal_handler.cc | 1 -
+ 5 files changed, 15 insertions(+), 16 deletions(-)
+
+diff --git a/CMakeLists.txt b/CMakeLists.txt
+index 2dd7084..2ba47e6 100644
+--- a/CMakeLists.txt
++++ b/CMakeLists.txt
+@@ -213,22 +213,24 @@ set(STATIC_LIBS
+ boost_chrono)
+
+ find_package(Threads)
+-find_package(ZLIB)
++find_package(ZLIB REQUIRED)
+ find_package(AWSSDK REQUIRED COMPONENTS kinesis monitoring sts)
++find_package(OpenSSL REQUIRED)
++find_package(CURL REQUIRED)
+
+-add_library(LibCrypto STATIC IMPORTED)
+-set_property(TARGET LibCrypto PROPERTY IMPORTED_LOCATION ${THIRD_PARTY_LIB_DIR}/libcrypto.a)
++add_library(LibCrypto SHARED IMPORTED)
++set_property(TARGET LibCrypto PROPERTY IMPORTED_LOCATION ${OPENSSL_CRYPTO_LIBRARY})
+ set_property(TARGET LibCrypto PROPERTY IMPORTED_LINK_INTERFACE_LIBRARIES ${LIBDL_LIBRARIES})
+
+-add_library(LibSsl STATIC IMPORTED)
+-set_property(TARGET LibSsl PROPERTY IMPORTED_LOCATION ${THIRD_PARTY_LIB_DIR}/libssl.a)
++add_library(LibSsl SHARED IMPORTED)
++set_property(TARGET LibSsl PROPERTY IMPORTED_LOCATION ${OPENSSL_SSL_LIBRARY})
+ set_property(TARGET LibSsl PROPERTY IMPORTED_LINK_INTERFACE_LIBRARIES LibCrypto)
+
+-add_library(LibZ STATIC IMPORTED)
+-set_property(TARGET LibZ PROPERTY IMPORTED_LOCATION ${THIRD_PARTY_LIB_DIR}/libz.a)
++add_library(LibZ SHARED IMPORTED)
++set_property(TARGET LibZ PROPERTY IMPORTED_LOCATION ${ZLIB_LIBRARIES})
+
+-add_library(LibCurl STATIC IMPORTED)
+-set_property(TARGET LibCurl PROPERTY IMPORTED_LOCATION ${THIRD_PARTY_LIB_DIR}/libcurl.a)
++add_library(LibCurl SHARED IMPORTED)
++set_property(TARGET LibCurl PROPERTY IMPORTED_LOCATION ${CURL_LIBRARIES})
+ set_property(TARGET LibCurl PROPERTY IMPORTED_LINK_INTERFACE_LIBRARIES ${CMAKE_THREAD_LIBS_INIT} ${LIBRT_LIBRARIES} ${LIBDL_LIBRARIES} LibSsl LibZ)
+
+ add_library(LibProto STATIC IMPORTED)
+diff --git a/aws/utils/backtrace/bsd_backtrace.cc b/aws/utils/backtrace/bsd_backtrace.cc
+index fb5dbe3..cd6c5fe 100644
+--- a/aws/utils/backtrace/bsd_backtrace.cc
++++ b/aws/utils/backtrace/bsd_backtrace.cc
+@@ -15,10 +15,10 @@
+
+ #include "backtrace.h"
+ #include <aws/utils/writer_methods.h>
+-#include <execinfo.h>
+ #include <unistd.h>
+
+ #ifdef BSD_BACKTRACE
++#include <execinfo.h>
+ namespace aws {
+ namespace utils {
+ namespace backtrace {
+diff --git a/aws/utils/backtrace/gcc_backtrace.cc b/aws/utils/backtrace/gcc_backtrace.cc
+index 446ede9..32a866d 100644
+--- a/aws/utils/backtrace/gcc_backtrace.cc
++++ b/aws/utils/backtrace/gcc_backtrace.cc
+@@ -15,7 +15,6 @@
+
+ #include "backtrace.h"
+ #include <aws/utils/writer_methods.h>
+-#include <execinfo.h>
+ #include <unistd.h>
+ #include <aws/utils/logging.h>
+ #include <aws/utils/writer_methods.h>
+@@ -23,7 +22,7 @@
+ #include <cstdio>
+
+ #ifdef LIB_BACKTRACE
+-
++#include <execinfo.h>
+ #include <backtrace.h>
+
+ namespace {
+diff --git a/aws/utils/backtrace/null_backtrace.cc b/aws/utils/backtrace/null_backtrace.cc
+index 69d57f9..d443eae 100644
+--- a/aws/utils/backtrace/null_backtrace.cc
++++ b/aws/utils/backtrace/null_backtrace.cc
+@@ -15,10 +15,9 @@
+
+ #include "backtrace.h"
+ #include <aws/utils/writer_methods.h>
+-#include <execinfo.h>
+ #include <unistd.h>
+
+-#ifdef NULL_STACKTRACE
++#ifdef NULL_BACKTRACE
+
+ namespace aws {
+ namespace utils {
+@@ -36,4 +35,4 @@ void stack_trace_for_signal(int skip, bool /*signaled*/) {
+ }
+ }
+
+-#endif // NULL_STACKTRACE
++#endif // NULL_BACKTRACE
+diff --git a/aws/utils/signal_handler.cc b/aws/utils/signal_handler.cc
+index b58ab0e..f483c77 100644
+--- a/aws/utils/signal_handler.cc
++++ b/aws/utils/signal_handler.cc
+@@ -19,7 +19,6 @@
+ #include "backtrace/backtrace.h"
+
+ #include <signal.h>
+-#include <execinfo.h>
+ #include <unistd.h>
+ #include <cstring>
+ #include <cstdint>
+--
+2.47.1
+
diff --git a/docker/pulsar-all/Dockerfile b/docker/pulsar-all/Dockerfile
index 81ad74b6500..59ab86e4556 100644
--- a/docker/pulsar-all/Dockerfile
+++ b/docker/pulsar-all/Dockerfile
@@ -17,16 +17,42 @@
# under the License.
#
+# global arguments (only to be used in FROM clauses)
ARG PULSAR_IMAGE
-FROM busybox as pulsar-all
+ARG PULSAR_IO_KINESIS_KPL_IMAGE
-ARG PULSAR_IO_DIR
-ARG PULSAR_OFFLOADER_TARBALL
+FROM busybox AS pulsar-extensions
+ARG PULSAR_IO_DIR
ADD ${PULSAR_IO_DIR} /connectors
+ARG PULSAR_OFFLOADER_TARBALL
ADD ${PULSAR_OFFLOADER_TARBALL} /
RUN mv /apache-pulsar-offloaders-*/offloaders /offloaders
-FROM $PULSAR_IMAGE
-COPY --from=pulsar-all /connectors /pulsar/connectors
-COPY --from=pulsar-all /offloaders /pulsar/offloaders
+FROM ${PULSAR_IO_KINESIS_KPL_IMAGE} AS pulsar-io-kinesis-sink-kinesis_producer
+
+FROM ${PULSAR_IMAGE}
+COPY --from=pulsar-extensions /connectors /pulsar/connectors
+COPY --from=pulsar-extensions /offloaders /pulsar/offloaders
+# Copy the kinesis_producer native executable compiled for Alpine musl to the pulsar-all image
+# This is required to support the Pulsar IO Kinesis sink connector
+COPY --from=pulsar-io-kinesis-sink-kinesis_producer /opt/amazon-kinesis-producer/bin/kinesis_producer /opt/amazon-kinesis-producer/bin/.os_info /opt/amazon-kinesis-producer/bin/.build_time /opt/amazon-kinesis-producer/bin/.revision /opt/amazon-kinesis-producer/bin/.system_info /opt/amazon-kinesis-producer/bin/.version /opt/amazon-kinesis-producer/bin/
+# Set the environment variable to point to the kinesis_producer native executable
+ENV PULSAR_IO_KINESIS_KPL_PATH=/opt/amazon-kinesis-producer/bin/kinesis_producer
+# Install the required dependencies for the kinesis_producer native executable
+USER 0
+RUN apk update && apk add --no-cache \
+ brotli-libs \
+ c-ares \
+ libcrypto3 \
+ libcurl \
+ libgcc \
+ libidn2 \
+ libpsl \
+ libssl3 \
+ libunistring \
+ nghttp2-libs \
+ zlib \
+ zstd-libs \
+ libuuid
+USER 10000
\ No newline at end of file
diff --git a/docker/pulsar-all/pom.xml b/docker/pulsar-all/pom.xml
index 67fada6a49e..999bcf93a7a 100644
--- a/docker/pulsar-all/pom.xml
+++ b/docker/pulsar-all/pom.xml
@@ -80,6 +80,10 @@
<profile>
<id>docker</id>
<!-- include the docker image only when docker profile is active -->
+ <properties>
+ <!-- reference to the image built from docker/kinesis-producer-alpine/Dockerfile -->
+ <PULSAR_IO_KINESIS_KPL_IMAGE>apachepulsar/pulsar-io-kinesis-sink-kinesis_producer:0.15.12</PULSAR_IO_KINESIS_KPL_IMAGE>
+ </properties>
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
@@ -161,6 +165,7 @@
<PULSAR_IO_DIR>target/apache-pulsar-io-connectors-${project.version}-bin</PULSAR_IO_DIR>
<PULSAR_OFFLOADER_TARBALL>target/pulsar-offloader-distribution-${project.version}-bin.tar.gz</PULSAR_OFFLOADER_TARBALL>
<PULSAR_IMAGE>${docker.organization}/${docker.image}:${project.version}-${git.commit.id.abbrev}</PULSAR_IMAGE>
+ <PULSAR_IO_KINESIS_KPL_IMAGE>${PULSAR_IO_KINESIS_KPL_IMAGE}</PULSAR_IO_KINESIS_KPL_IMAGE>
</args>
<buildx>
<platforms>
diff --git a/docker/pulsar/Dockerfile b/docker/pulsar/Dockerfile
index c8fc3a4d13c..6f1f41755c9 100644
--- a/docker/pulsar/Dockerfile
+++ b/docker/pulsar/Dockerfile
@@ -75,7 +75,7 @@ ARG SNAPPY_VERSION
RUN apk add git alpine-sdk util-linux cmake autoconf automake libtool openjdk17 maven curl bash tar
ENV JAVA_HOME=/usr
RUN curl -Ls https://github.com/xerial/snappy-java/archive/refs/tags/v$SNAPPY_VERSION.tar.gz | tar zxf - && cd snappy-java-$SNAPPY_VERSION && make clean-native native
-FROM apachepulsar/glibc-base:2.38 as glibc
+
## Create final stage from Alpine image
## and add OpenJDK and Python dependencies (for Pulsar functions)
@@ -90,6 +90,9 @@ RUN apk add --no-cache \
py3-grpcio \
py3-yaml \
gcompat \
+ libgcc \
+ libstdc++ \
+ libuuid \
ca-certificates \
procps \
curl \
@@ -124,10 +127,6 @@ fastavro>=1.9.2\n\
RUN pip3 install --break-system-packages --no-cache-dir --only-binary grpcio -r /requirements.txt
RUN rm /requirements.txt
-# Install GLibc compatibility library
-COPY --from=glibc /root/packages /root/packages
-RUN apk add --allow-untrusted --force-overwrite /root/packages/glibc-*.apk
-
COPY --from=jvm /opt/jvm /opt/jvm
ENV JAVA_HOME=/opt/jvm
@@ -144,6 +143,8 @@ WORKDIR /pulsar
ENV PATH=$PATH:$JAVA_HOME/bin:/pulsar/bin
# Use musl libc library for RocksDB
ENV ROCKSDB_MUSL_LIBC=true
+# Preload gcompat library for glibc compatibility with Netty native libraries
+ENV LD_PRELOAD=/lib/libgcompat.so.0
# The UID must be non-zero. Otherwise, it is arbitrary. No logic should rely on its specific value.
ARG DEFAULT_USERNAME=pulsar
diff --git a/pom.xml b/pom.xml
index dad306c6f1c..476a32798e6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -166,7 +166,7 @@ flexible messaging model and an intuitive client API.</description>
<reflections.version>0.10.2</reflections.version>
<swagger.version>1.6.2</swagger.version>
<puppycrawl.checkstyle.version>10.14.2</puppycrawl.checkstyle.version>
- <docker-maven.version>0.45.0</docker-maven.version>
+ <docker-maven.version>0.45.1</docker-maven.version>
<docker.verbose>true</docker.verbose>
<typetools.version>0.5.0</typetools.version>
<byte-buddy.version>1.14.12</byte-buddy.version>
@@ -2098,6 +2098,7 @@ flexible messaging model and an intuitive client API.</description>
<exclude>**/*.so</exclude>
<exclude>**/*.so.*</exclude>
<exclude>**/*.dylib</exclude>
+ <exclude>**/*.patch</exclude>
<exclude>src/test/resources/*.txt</exclude>
</excludes>
</configuration>
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
index c2310d541c5..745e81d5d70 100644
--- a/pulsar-io/kinesis/pom.xml
+++ b/pulsar-io/kinesis/pom.xml
@@ -32,7 +32,7 @@
<properties>
<amazon-kinesis-client.version>2.2.8</amazon-kinesis-client.version>
- <amazon-kinesis-producer.version>0.14.13</amazon-kinesis-producer.version>
+ <amazon-kinesis-producer.version>0.15.12</amazon-kinesis-producer.version>
<json-flattener.version>0.13.0</json-flattener.version>
<flatbuffers-java.version>1.9.0</flatbuffers-java.version>
<jaxb-api.version>2.3.0</jaxb-api.version>
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
index d8e4e4bab85..1db63c90b27 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSink.java
@@ -88,7 +88,6 @@ import org.slf4j.LoggerFactory;
configClass = KinesisSinkConfig.class
)
public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObject> {
-
private static final Logger LOG = LoggerFactory.getLogger(KinesisSink.class);
private KinesisProducer kinesisProducer;
@@ -166,6 +165,12 @@ public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObj
if (kinesisSinkConfig.getAwsEndpointPort() != null) {
kinesisConfig.setKinesisPort(kinesisSinkConfig.getAwsEndpointPort());
}
+ if (kinesisSinkConfig.getAwsStsEndpoint() != null) {
+ kinesisConfig.setStsEndpoint(kinesisSinkConfig.getAwsStsEndpoint());
+ }
+ if (kinesisSinkConfig.getAwsStsPort() != null) {
+ kinesisConfig.setStsPort(kinesisSinkConfig.getAwsStsPort());
+ }
kinesisConfig.setRegion(kinesisSinkConfig.getAwsRegion());
kinesisConfig.setThreadingModel(ThreadingModel.POOLED);
kinesisConfig.setThreadPoolSize(4);
@@ -179,6 +184,7 @@ public class KinesisSink extends AbstractAwsConnector implements Sink<GenericObj
kinesisSinkConfig.getAwsCredentialPluginParam())
.getCredentialProvider();
kinesisConfig.setCredentialsProvider(credentialsProvider);
+ kinesisConfig.setNativeExecutable(StringUtils.trimToEmpty(kinesisSinkConfig.getNativeExecutable()));
this.streamName = kinesisSinkConfig.getAwsKinesisStreamName();
this.kinesisProducer = new KinesisProducer(kinesisConfig);
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
index f81fd32134b..a3b87d4886a 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/KinesisSinkConfig.java
@@ -104,6 +104,16 @@ public class KinesisSinkConfig extends BaseKinesisConfig implements Serializable
help = "The maximum delay(in milliseconds) between retries.")
private long retryMaxDelayInMillis = 60000;
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "Path to the native Amazon Kinesis Producer Library (KPL) binary.\n"
+ + "Only use this setting if you want to use a custom build of the native code.\n"
+ + "This setting can also be set with the environment variable `PULSAR_IO_KINESIS_KPL_PATH`.\n"
+ + "If not set, the Kinesis sink will use the built-in native executable."
+ )
+ private String nativeExecutable = System.getenv("PULSAR_IO_KINESIS_KPL_PATH");
+
public static KinesisSinkConfig load(Map<String, Object> config, SinkContext sinkContext) {
KinesisSinkConfig kinesisSinkConfig = IOConfigUtils.loadWithSecrets(config, KinesisSinkConfig.class, sinkContext);
checkArgument(isNotBlank(kinesisSinkConfig.getAwsRegion())
@@ -149,4 +159,17 @@ public class KinesisSinkConfig extends BaseKinesisConfig implements Serializable
FULL_MESSAGE_IN_JSON_EXPAND_VALUE
}
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "Custom AWS STS endpoint"
+ )
+ private String awsStsEndpoint = "";
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "Custom AWS STS port to connect to"
+ )
+ private Integer awsStsPort;
}
diff --git a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java
index 8a5d5be0ca4..c682776a14d 100644
--- a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java
+++ b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/KinesisSinkTest.java
@@ -18,6 +18,16 @@
*/
package org.apache.pulsar.io.kinesis;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import lombok.SneakyThrows;
import org.apache.pulsar.client.api.Message;
@@ -42,24 +52,14 @@ import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
import software.amazon.awssdk.services.kinesis.model.ListShardsRequest;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
-import java.net.URI;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
-
public class KinesisSinkTest {
public static final String STREAM_NAME = "my-stream-1";
- public static LocalStackContainer LOCALSTACK_CONTAINER = new LocalStackContainer(DockerImageName.parse("localstack/localstack:1.0.4"))
- .withServices(LocalStackContainer.Service.KINESIS);
+ public static LocalStackContainer LOCALSTACK_CONTAINER =
+ new LocalStackContainer(DockerImageName.parse("localstack/localstack:4.0.3"))
+ .withServices(LocalStackContainer.Service.KINESIS, LocalStackContainer.Service.STS)
+ .withEnv("KINESIS_PROVIDER", "kinesalite");
@BeforeClass(alwaysRun = true)
public void beforeClass() throws Exception {
@@ -123,10 +123,13 @@ public class KinesisSinkTest {
}
private Map<String, Object> createConfig() {
- final URI endpointOverride = LOCALSTACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS);
+ final URI kinesisEndpointOverride = LOCALSTACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.KINESIS);
Map<String, Object> map = new HashMap<>();
- map.put("awsEndpoint", endpointOverride.getHost());
- map.put("awsEndpointPort", endpointOverride.getPort());
+ map.put("awsEndpoint", kinesisEndpointOverride.getHost());
+ map.put("awsEndpointPort", kinesisEndpointOverride.getPort());
+ final URI stsEndpointOverride = LOCALSTACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.STS);
+ map.put("awsStsEndpoint", stsEndpointOverride.getHost());
+ map.put("awsStsPort", stsEndpointOverride.getPort());
map.put("skipCertificateValidation", true);
map.put("awsKinesisStreamName", STREAM_NAME);
map.put("awsRegion", "us-east-1");
diff --git a/tests/docker-images/latest-version-image/Dockerfile b/tests/docker-images/latest-version-image/Dockerfile
index 0645dd2e78a..3efc4817214 100644
--- a/tests/docker-images/latest-version-image/Dockerfile
+++ b/tests/docker-images/latest-version-image/Dockerfile
@@ -17,10 +17,9 @@
# under the License.
#
-# build go lang examples first in a separate layer
ARG PULSAR_ALL_IMAGE
-ARG PULSAR_IMAGE
+# build go lang examples first in a separate layer
FROM golang:1.21-alpine as pulsar-function-go
COPY target/pulsar-function-go/ /go/src/github.com/apache/pulsar/pulsar-function-go
@@ -28,13 +27,10 @@ RUN cd /go/src/github.com/apache/pulsar/pulsar-function-go && go install ./...
RUN cd /go/src/github.com/apache/pulsar/pulsar-function-go/pf && go install
RUN cd /go/src/github.com/apache/pulsar/pulsar-function-go/examples && go install ./...
-# Reference pulsar-all to copy connectors from there
-FROM $PULSAR_ALL_IMAGE as pulsar-all
-
########################################
###### Main image build
########################################
-FROM $PULSAR_IMAGE
+FROM $PULSAR_ALL_IMAGE
# Switch to run as the root user to simplify building container and then running
# supervisord. Each of the pulsar components are spawned by supervisord and their
@@ -78,20 +74,6 @@ COPY target/certificate-authority /pulsar/certificate-authority/
# copy broker plugins
COPY target/plugins/ /pulsar/examples/
-# Include all offloaders
-COPY --from=pulsar-all /pulsar/offloaders /pulsar/offloaders
-
-# Include only the connectors needed by integration tests
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-cassandra-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-debezium-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-elastic-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-hdfs*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-jdbc-postgres-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-kafka-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-rabbitmq-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-batch-data-generator-*.nar /pulsar/connectors/
-COPY --from=pulsar-all /pulsar/connectors/pulsar-io-kinesis-*.nar /pulsar/connectors/
-
# download Oracle JDBC driver for Oracle Debezium Connector tests
RUN mkdir -p META-INF/bundled-dependencies
RUN cd META-INF/bundled-dependencies && curl -sSL https://search.maven.org/remotecontent?filepath=com/oracle/ojdbc/ojdbc8/19.3.0.0/ojdbc8-19.3.0.0.jar -o ojdbc8-19.3.0.0.jar
@@ -110,4 +92,4 @@ RUN jar uf connectors/pulsar-io-debezium-oracle-*.nar META-INF/bundled-dependenc
RUN mkdir -p pulsar
RUN chmod g+rwx pulsar
-CMD bash
+CMD bash
\ No newline at end of file
diff --git a/tests/docker-images/latest-version-image/pom.xml b/tests/docker-images/latest-version-image/pom.xml
index c1f627f9614..cdf4ac9e206 100644
--- a/tests/docker-images/latest-version-image/pom.xml
+++ b/tests/docker-images/latest-version-image/pom.xml
@@ -156,7 +156,6 @@
<build>
<contextDir>${project.basedir}</contextDir>
<args>
- <PULSAR_IMAGE>${docker.organization}/${docker.image}:${project.version}-${git.commit.id.abbrev}</PULSAR_IMAGE>
<PULSAR_ALL_IMAGE>${docker.organization}/${docker.image}-all:${project.version}-${git.commit.id.abbrev}</PULSAR_ALL_IMAGE>
</args>
<tags>
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java
index 6ca9c7b1083..83cb0088cd7 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/KinesisSinkTester.java
@@ -79,6 +79,11 @@ public class KinesisSinkTester extends SinkTester<LocalStackContainer> {
sinkConfig.put("awsKinesisStreamName", STREAM_NAME);
sinkConfig.put("awsRegion", "us-east-1");
sinkConfig.put("awsCredentialPluginParam", "{\"accessKey\":\"access\",\"secretKey\":\"secret\"}");
+ sinkConfig.put("awsEndpoint", NAME);
+ sinkConfig.put("awsEndpointPort", LOCALSTACK_SERVICE_PORT);
+ sinkConfig.put("awsStsEndpoint", NAME);
+ sinkConfig.put("awsStsPort", LOCALSTACK_SERVICE_PORT);
+ sinkConfig.put("skipCertificateValidation", true);
if (withSchema) {
sinkConfig.put("messageFormat", "FULL_MESSAGE_IN_JSON_EXPAND_VALUE");
}
@@ -100,9 +105,6 @@ public class KinesisSinkTester extends SinkTester<LocalStackContainer> {
public void prepareSink() throws Exception {
final LocalStackContainer localStackContainer = getServiceContainer();
final URI endpointOverride = localStackContainer.getEndpointOverride(LocalStackContainer.Service.KINESIS);
- sinkConfig.put("awsEndpoint", NAME);
- sinkConfig.put("awsEndpointPort", LOCALSTACK_SERVICE_PORT);
- sinkConfig.put("skipCertificateValidation", true);
client = KinesisAsyncClient.builder().credentialsProvider(() -> AwsBasicCredentials.create(
"access",
"secret"))
@@ -128,8 +130,9 @@ public class KinesisSinkTester extends SinkTester<LocalStackContainer> {
@Override
protected LocalStackContainer createSinkService(PulsarCluster cluster) {
- return new LocalStackContainer(DockerImageName.parse("localstack/localstack:1.0.4"))
- .withServices(LocalStackContainer.Service.KINESIS);
+ return new LocalStackContainer(DockerImageName.parse("localstack/localstack:4.0.3"))
+ .withServices(LocalStackContainer.Service.KINESIS, LocalStackContainer.Service.STS)
+ .withEnv("KINESIS_PROVIDER", "kinesalite");
}
@Override
(pulsar) 03/05: [fix][client] Fix reader message filtering issue during blue-green cluster switch (#23693)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8700d1c2d7e6f40e95055c74bf2913c0a8aab2c7
Author: hrzzzz <64...@users.noreply.github.com>
AuthorDate: Fri Dec 20 20:05:09 2024 +0800
[fix][client] Fix reader message filtering issue during blue-green cluster switch (#23693)
Co-authored-by: ruihongzhou <ru...@tencent.com>
(cherry picked from commit 34c2f30d7838a1d50484985ee8bcfb1d573c50ed)
---
.../broker/service/ClusterMigrationTest.java | 111 +++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 7 ++
2 files changed, 118 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
index e6a7d049366..167c154c1fd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
@@ -45,6 +45,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
@@ -911,6 +912,116 @@ public class ClusterMigrationTest {
client2.close();
}
+ public void testMigrationWithReader() throws Exception {
+ final String topicName = BrokerTestUtil
+ .newUniqueName("persistent://" + namespace + "/migrationTopic");
+
+ @Cleanup
+ PulsarClient client1 = PulsarClient.builder()
+ .serviceUrl(url1.toString())
+ .statsInterval(0, TimeUnit.SECONDS)
+ .build();
+ // cluster-1 producer/reader
+ Producer<byte[]> producer1 = client1.newProducer()
+ .topic(topicName)
+ .enableBatching(false)
+ .producerName("cluster1-1")
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+ Reader<byte[]> reader1 =client1.newReader()
+ .topic(topicName)
+ .startMessageId(MessageId.earliest)
+ .subscriptionRolePrefix("s1")
+ .create();
+
+ AbstractTopic topic1 = (AbstractTopic) pulsar1.getBrokerService().getTopic(topicName, false).getNow(null).get();
+ retryStrategically((test) -> !topic1.getProducers().isEmpty(), 5, 500);
+ retryStrategically((test) -> !topic1.getSubscriptions().isEmpty(), 5, 500);
+ assertFalse(topic1.getProducers().isEmpty());
+ assertFalse(topic1.getSubscriptions().isEmpty());
+
+ // build backlog
+ reader1.close();
+ int n = 8;
+ for (int i = 0; i < n; i++) {
+ producer1.send("test1".getBytes());
+ }
+
+ @Cleanup
+ PulsarClient client2 = PulsarClient.builder()
+ .serviceUrl(url2.toString())
+ .statsInterval(0, TimeUnit.SECONDS)
+ .build();
+ // cluster-2 producer
+ Producer<byte[]> producer2 = client2.newProducer()
+ .topic(topicName)
+ .enableBatching(false)
+ .producerName("cluster2-1")
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+
+ AbstractTopic topic2 = (AbstractTopic) pulsar2.getBrokerService().getTopic(topicName, false).getNow(null).get();
+ assertFalse(topic2.getProducers().isEmpty());
+ assertTrue(topic2.getSubscriptions().isEmpty());
+
+ // migrate topic to cluster-2
+ ClusterUrl migratedUrl = new ClusterUrl(pulsar2.getWebServiceAddress(), pulsar2.getWebServiceAddressTls(),
+ pulsar2.getBrokerServiceUrl(), null);
+ admin1.clusters().updateClusterMigration("r1", true, migratedUrl);
+ assertEquals(admin1.clusters().getClusterMigration("r1").getMigratedClusterUrl(), migratedUrl);
+ retryStrategically((test) -> {
+ try {
+ topic1.checkClusterMigration().get();
+ return true;
+ } catch (Exception e) {
+ // ok
+ }
+ return false;
+ }, 10, 500);
+ topic1.checkClusterMigration().get();
+
+ sleep(1000);
+ producer1.sendAsync("test1".getBytes());
+
+ // producer is disconnected from cluster-1
+ retryStrategically((test) -> topic1.getProducers().isEmpty(), 10, 500);
+ assertTrue(topic1.getProducers().isEmpty());
+
+ // producer is connected with cluster-2
+ retryStrategically((test) -> topic2.getProducers().size() == 2, 10, 500);
+ assertEquals(topic2.getProducers().size(), 2);
+
+ // try to consume backlog messages from cluster-1
+ reader1 = client1.newReader()
+ .topic(topicName)
+ .startMessageId(MessageId.earliest)
+ .subscriptionRolePrefix("s1")
+ .create();
+ for (int i = 0; i < n; i++) {
+ Message<byte[]> msg = reader1.readNext();
+ assertEquals(msg.getData(), "test1".getBytes());
+ }
+
+ // after consuming all messages, reader should have disconnected from cluster-1 and reconnect with cluster-2
+ retryStrategically((test) -> !topic2.getSubscriptions().isEmpty(), 10, 500);
+ assertFalse(topic2.getSubscriptions().isEmpty());
+ assertTrue(topic1.getSubscriptions().isEmpty());
+
+ n = 4;
+ // publish messages to cluster-2 and consume them
+ for (int i = 0; i < n; i++) {
+ producer1.send("test2".getBytes());
+ }
+
+ for (int i = 0; i < n; i++) {
+ assertEquals(reader1.readNext(2, TimeUnit.SECONDS).getData(), "test2".getBytes());
+ }
+
+ client1.close();
+ client2.close();
+ }
+
+
@Test(dataProvider = "NamespaceMigrationTopicSubscriptionTypes")
public void testNamespaceMigrationWithReplicationBacklog(SubscriptionType subType, boolean isClusterMigrate, boolean isNamespaceMigrate) throws Exception {
log.info("--- Starting ReplicatorTest::testNamespaceMigrationWithReplicationBacklog ---");
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index e01c6d4643b..4d1b51e34db 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -37,6 +37,7 @@ import io.netty.util.concurrent.FastThreadLocal;
import io.opentelemetry.api.common.Attributes;
import java.io.IOException;
import java.net.URI;
+import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -3107,6 +3108,12 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
&& Commands.peerSupportsAckReceipt(cnx.getRemoteEndpointProtocolVersion());
}
+ @Override
+ protected void setRedirectedClusterURI(String serviceUrl, String serviceUrlTls) throws URISyntaxException {
+ super.setRedirectedClusterURI(serviceUrl, serviceUrlTls);
+ acknowledgmentsGroupingTracker.flushAndClean();
+ }
+
private static final Logger log = LoggerFactory.getLogger(ConsumerImpl.class);
@VisibleForTesting
(pulsar) 02/05: [fix][broker] Fix bug causing loss of migrated information when setting other localPolicies in namespace (#23764)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0f15a3667e1566b7f0b646061cdef19a6fd3775a
Author: hrzzzz <64...@users.noreply.github.com>
AuthorDate: Fri Dec 20 20:04:24 2024 +0800
[fix][broker] Fix bug causing loss of migrated information when setting other localPolicies in namespace (#23764)
Co-authored-by: ruihongzhou <ru...@tencent.com>
(cherry picked from commit bbe2cabc3ec0375607cb12665cab0b4745dbd36e)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 20 ++++++++++------
.../pulsar/common/naming/NamespaceBundles.java | 3 ++-
.../apache/pulsar/broker/admin/NamespacesTest.java | 27 ++++++++++++++++++++++
.../pulsar/common/policies/data/LocalPolicies.java | 11 ++++++++-
4 files changed, 52 insertions(+), 9 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index d80e2487b4f..ca4c685b280 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -960,7 +960,8 @@ public abstract class NamespacesBase extends AdminResource {
LocalPolicies localPolicies = oldPolicies.map(
policies -> new LocalPolicies(policies.bundles,
bookieAffinityGroup,
- policies.namespaceAntiAffinityGroup))
+ policies.namespaceAntiAffinityGroup,
+ policies.migrated))
.orElseGet(() -> new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles()),
bookieAffinityGroup,
null));
@@ -1779,7 +1780,8 @@ public abstract class NamespacesBase extends AdminResource {
getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, (lp)->
lp.map(policies -> new LocalPolicies(policies.bundles,
policies.bookieAffinityGroup,
- antiAffinityGroup))
+ antiAffinityGroup,
+ policies.migrated))
.orElseGet(() -> new LocalPolicies(defaultBundle(),
null, antiAffinityGroup))
);
@@ -1816,7 +1818,8 @@ public abstract class NamespacesBase extends AdminResource {
getLocalPolicies().setLocalPolicies(namespaceName, (policies)->
new LocalPolicies(policies.bundles,
policies.bookieAffinityGroup,
- null));
+ null,
+ policies.migrated));
log.info("[{}] Successfully removed anti-affinity group for a namespace={}", clientAppId(), namespaceName);
} catch (Exception e) {
log.error("[{}] Failed to remove anti-affinity group for namespace {}", clientAppId(), namespaceName, e);
@@ -2765,10 +2768,13 @@ public abstract class NamespacesBase extends AdminResource {
protected void internalEnableMigration(boolean migrated) {
validateSuperUserAccess();
try {
- getLocalPolicies().setLocalPolicies(namespaceName, (policies) -> {
- policies.migrated = migrated;
- return policies;
- });
+ getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, oldPolicies -> oldPolicies.map(
+ policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup,
+ policies.namespaceAntiAffinityGroup,
+ migrated))
+ .orElseGet(() -> new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles()),
+ null, null, migrated)));
log.info("Successfully updated migration on namespace {}", namespaceName);
} catch (Exception e) {
log.error("Failed to update migration on namespace {}", namespaceName, e);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java
index fa7baeaa606..3ee365cdd45 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java
@@ -197,6 +197,7 @@ public class NamespaceBundles {
public LocalPolicies toLocalPolicies() {
return new LocalPolicies(this.getBundlesData(),
localPolicies.map(lp -> lp.getLeft().bookieAffinityGroup).orElse(null),
- localPolicies.map(lp -> lp.getLeft().namespaceAntiAffinityGroup).orElse(null));
+ localPolicies.map(lp -> lp.getLeft().namespaceAntiAffinityGroup).orElse(null),
+ localPolicies.map(lp -> lp.getLeft().migrated).orElse(false));
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
index f2948660952..18cc449d15d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -102,6 +102,7 @@ import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
+import org.apache.pulsar.common.policies.data.BookieAffinityGroupData;
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.DispatchRate;
@@ -2195,4 +2196,30 @@ public class NamespacesTest extends MockedPulsarServiceBaseTest {
admin.namespaces().deleteNamespace(namespace);
}
+
+ public void testMigratedInfoIsNotLostDuringOtherLocalPoliciesUpdate() throws Exception {
+ String namespace = BrokerTestUtil.newUniqueName(this.testTenant + "/namespace");
+ admin.namespaces().createNamespace(namespace, Set.of(testLocalCluster));
+
+ admin.namespaces().updateMigrationState(namespace, true);
+ assertTrue(admin.namespaces().getPolicies(namespace).migrated);
+
+ String bookieAffinityGroupPrimary = "group1";
+ admin.namespaces().setBookieAffinityGroup(namespace,
+ BookieAffinityGroupData.builder().bookkeeperAffinityGroupPrimary(bookieAffinityGroupPrimary).build());
+ assertEquals(admin.namespaces().getBookieAffinityGroup(namespace).getBookkeeperAffinityGroupPrimary(),
+ bookieAffinityGroupPrimary);
+ assertTrue(admin.namespaces().getPolicies(namespace).migrated);
+
+ String namespaceAntiAffinityGroup = "group2";
+ admin.namespaces().setNamespaceAntiAffinityGroup(namespace, namespaceAntiAffinityGroup);
+ assertEquals(admin.namespaces().getNamespaceAntiAffinityGroup(namespace), namespaceAntiAffinityGroup);
+ assertTrue(admin.namespaces().getPolicies(namespace).migrated);
+
+ admin.namespaces().deleteBookieAffinityGroup(namespace);
+ assertTrue(admin.namespaces().getPolicies(namespace).migrated);
+
+ admin.namespaces().deleteNamespaceAntiAffinityGroup(namespace);
+ assertTrue(admin.namespaces().getPolicies(namespace).migrated);
+ }
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/LocalPolicies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/LocalPolicies.java
index 3b17dbe067e..43f5130eb9f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/LocalPolicies.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/LocalPolicies.java
@@ -34,20 +34,29 @@ public class LocalPolicies {
public final BookieAffinityGroupData bookieAffinityGroup;
// namespace anti-affinity-group
public final String namespaceAntiAffinityGroup;
- public boolean migrated;
+ public final boolean migrated;
public LocalPolicies() {
bundles = defaultBundle();
bookieAffinityGroup = null;
namespaceAntiAffinityGroup = null;
+ migrated = false;
}
public LocalPolicies(BundlesData data,
BookieAffinityGroupData bookieAffinityGroup,
String namespaceAntiAffinityGroup) {
+ this(data, bookieAffinityGroup, namespaceAntiAffinityGroup, false);
+ }
+
+ public LocalPolicies(BundlesData data,
+ BookieAffinityGroupData bookieAffinityGroup,
+ String namespaceAntiAffinityGroup,
+ boolean migrated) {
bundles = data;
this.bookieAffinityGroup = bookieAffinityGroup;
this.namespaceAntiAffinityGroup = namespaceAntiAffinityGroup;
+ this.migrated = migrated;
}
}
\ No newline at end of file