You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/01/19 02:12:09 UTC
[pulsar] branch branch-2.7 updated (8b7c81e -> 4dcc738)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 8b7c81e [Functions-worker] Fix broker and functions-worker authentication compatibility (#9190)
new 109ff54 Fix locking for ConsumerImpl when creating deadLetterProducer. (#9166)
new 4dcc738 Skip the interceptor for MediaType.MULTIPART_FORM_DATA (#9217)
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:
.../pulsar/broker/web/PreInterceptFilter.java | 3 ++-
.../pulsar/broker/web/ResponseHandlerFilter.java | 4 +++-
.../org/apache/pulsar/client/impl/ConsumerImpl.java | 21 +++++++++++++--------
3 files changed, 18 insertions(+), 10 deletions(-)
[pulsar] 02/02: Skip the interceptor for
MediaType.MULTIPART_FORM_DATA (#9217)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4dcc7381136960edd69f54e6841388e6a028e429
Author: lipenghui <pe...@apache.org>
AuthorDate: Mon Jan 18 17:03:32 2021 +0800
Skip the interceptor for MediaType.MULTIPART_FORM_DATA (#9217)
Skip the interceptor for `MediaType.MULTIPART_FORM_DATA`
(cherry picked from commit 93b2239945fe189f1eaeb8724062143cb51d2d62)
---
.../main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java | 3 ++-
.../main/java/org/apache/pulsar/broker/web/ResponseHandlerFilter.java | 4 +++-
2 files changed, 5 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java
index 57f5c20..f206185 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PreInterceptFilter.java
@@ -54,7 +54,8 @@ public class PreInterceptFilter implements Filter {
servletRequest.getServletContext().getContextPath(),
servletRequest.getContentType());
}
- if (MediaType.MULTIPART_FORM_DATA.equalsIgnoreCase(servletRequest.getContentType())) {
+ if (MediaType.MULTIPART_FORM_DATA.equalsIgnoreCase(servletRequest.getContentType())
+ || MediaType.APPLICATION_OCTET_STREAM.equalsIgnoreCase(servletRequest.getContentType())) {
// skip multipart request at this moment
filterChain.doFilter(servletRequest, servletResponse);
return;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ResponseHandlerFilter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ResponseHandlerFilter.java
index 3739581..60d8a48 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ResponseHandlerFilter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ResponseHandlerFilter.java
@@ -28,6 +28,7 @@ import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response.Status;
import org.apache.pulsar.broker.PulsarService;
@@ -65,7 +66,8 @@ public class ResponseHandlerFilter implements Filter {
/* connection is already invalidated */
}
}
- if (interceptorEnabled) {
+ if (interceptorEnabled && !MediaType.MULTIPART_FORM_DATA.equalsIgnoreCase(request.getContentType())
+ && !MediaType.APPLICATION_OCTET_STREAM.equalsIgnoreCase(request.getContentType())) {
interceptor.onWebserviceResponse(request, response);
}
}
[pulsar] 01/02: Fix locking for ConsumerImpl when creating
deadLetterProducer. (#9166)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 109ff54b86929f562830e666aa13b41ecad65292
Author: Marvin Cai <zx...@streamnative.io>
AuthorDate: Sun Jan 17 20:59:10 2021 -0800
Fix locking for ConsumerImpl when creating deadLetterProducer. (#9166)
Fixes #9162
### Motivation
When we check if we need to create producer for DLQ, we used double-checked locking, https://github.com/apache/pulsar/blame/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L701-L712
however we should do the second check after we acquired the lock, and the field we're checking should be a volatile field.
### Modifications
Fix the double-checked locking and make the field volatile.
(cherry picked from commit b019c4b75348f2e4ff6a9b6c3367b2b951429fad)
---
.../org/apache/pulsar/client/impl/ConsumerImpl.java | 21 +++++++++++++--------
1 file changed, 13 insertions(+), 8 deletions(-)
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 0473e23..eb7d7d2 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
@@ -169,7 +169,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
private final DeadLetterPolicy deadLetterPolicy;
- private Producer<T> deadLetterProducer;
+ private volatile Producer<T> deadLetterProducer;
private volatile Producer<T> retryLetterProducer;
private final ReadWriteLock createProducerLock = new ReentrantReadWriteLock();
@@ -708,8 +708,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
processPossibleToDLQ((MessageIdImpl)messageId);
if (deadLetterProducer == null) {
try {
+ createProducerLock.writeLock().lock();
if (deadLetterProducer == null) {
- createProducerLock.writeLock().lock();
deadLetterProducer = client.newProducer(schema)
.topic(this.deadLetterPolicy
.getDeadLetterTopic())
@@ -718,9 +718,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
} catch (Exception e) {
log.error("Create dead letter producer exception with topic: {}", deadLetterPolicy.getDeadLetterTopic(), e);
- } finally {
+ } finally {
createProducerLock.writeLock().unlock();
- }
+ }
}
if (deadLetterProducer != null) {
propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_REAL_TOPIC, originTopicNameStr);
@@ -1840,12 +1840,17 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (deadLetterMessages != null) {
if (deadLetterProducer == null) {
try {
- deadLetterProducer = client.newProducer(schema)
- .topic(this.deadLetterPolicy.getDeadLetterTopic())
- .blockIfQueueFull(false)
- .create();
+ createProducerLock.writeLock().lock();
+ if (deadLetterProducer == null) {
+ deadLetterProducer = client.newProducer(schema)
+ .topic(this.deadLetterPolicy.getDeadLetterTopic())
+ .blockIfQueueFull(false)
+ .create();
+ }
} catch (Exception e) {
log.error("Create dead letter producer exception with topic: {}", deadLetterPolicy.getDeadLetterTopic(), e);
+ } finally {
+ createProducerLock.writeLock().unlock();
}
}
if (deadLetterProducer != null) {