[pulsar] branch master updated: catch bad version in persistentopic, make its behavior same as namespace (#3260)

[sijie]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 1fa81ca  catch bad version in persistentopic, make its behavior same 
as namespace (#3260)
1fa81ca is described below

commit 1fa81ca54f688938c83a0be3c98bebc7589be668
Author: Samuel <foreversun...@yahoo.com>
AuthorDate: Sat Dec 29 02:15:41 2018 +0800

    catch bad version in persistentopic, make its behavior same as namespace 
(#3260)
    
    ### Motivation
    
    Fixes #2952
    
    make topic's behavior same as 
[Namespaces.java](https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java)
---
 .../broker/admin/impl/PersistentTopicsBase.java    | 22 ++++++++++++++++++++--
 1 file changed, 20 insertions(+), 2 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index de1af66..aea5a16 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -297,7 +297,12 @@ public class PersistentTopicsBase extends AdminResource {
             log.warn("[{}] Failed to grant permissions on topic {}: Namespace 
does not exist", clientAppId(),
                     topicUri);
             throw new RestException(Status.NOT_FOUND, "Namespace does not 
exist");
-        } catch (Exception e) {
+        } catch (KeeperException.BadVersionException e) {
+            log.warn("[{}] Failed to grant permissions on topic {}: concurrent 
modification", clientAppId(),
+                    topicUri);
+            throw new RestException(Status.CONFLICT, "Concurrent 
modification");
+        }
+        catch (Exception e) {
             log.error("[{}] Failed to grant permissions for topic {}", 
clientAppId(), topicUri, e);
             throw new RestException(e);
         }
@@ -330,7 +335,12 @@ public class PersistentTopicsBase extends AdminResource {
             log.warn("[{}] Failed to revoke permissions on topic {}: Namespace 
does not exist", clientAppId(),
                     topicUri);
             throw new RestException(Status.NOT_FOUND, "Namespace does not 
exist");
-        } catch (Exception e) {
+        } catch (KeeperException.BadVersionException e) {
+            log.warn("[{}] Failed to revoke permissions on topic {}: 
concurrent modification", clientAppId(),
+                topicUri);
+            throw new RestException(Status.CONFLICT, "Concurrent 
modification");
+        }
+        catch (Exception e) {
             log.error("[{}] Failed to revoke permissions for topic {}", 
clientAppId(), topicUri, e);
             throw new RestException(e);
         }
@@ -376,6 +386,10 @@ public class PersistentTopicsBase extends AdminResource {
         } catch (KeeperException.NodeExistsException e) {
             log.warn("[{}] Failed to create already existing partitioned topic 
{}", clientAppId(), topicName);
             throw new RestException(Status.CONFLICT, "Partitioned topic 
already exists");
+        } catch (KeeperException.BadVersionException e) {
+                log.warn("[{}] Failed to create partitioned topic {}: 
concurrent modification", clientAppId(),
+                        topicName);
+                throw new RestException(Status.CONFLICT, "Concurrent 
modification");
         } catch (Exception e) {
             log.error("[{}] Failed to create partitioned topic {}", 
clientAppId(), topicName, e);
             throw new RestException(e);
@@ -480,6 +494,10 @@ public class PersistentTopicsBase extends AdminResource {
             log.info("[{}] Deleted partitioned topic {}", clientAppId(), 
topicName);
         } catch (KeeperException.NoNodeException nne) {
             throw new RestException(Status.NOT_FOUND, "Partitioned topic does 
not exist");
+        } catch (KeeperException.BadVersionException e) {
+            log.warn("[{}] Failed to delete partitioned topic {}: concurrent 
modification", clientAppId(),
+                    topicName);
+            throw new RestException(Status.CONFLICT, "Concurrent 
modification");
         } catch (Exception e) {
             log.error("[{}] Failed to delete partitioned topic {}", 
clientAppId(), topicName, e);
             throw new RestException(e);