1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
/*
* ============LICENSE_START==========================================
* org.onap.music
* ===================================================================
* Copyright (c) 2017 AT&T Intellectual Property
* ===================================================================
* Licensed 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.
*
* ============LICENSE_END=============================================
* ====================================================================
*/
package org.onap.music.lockingservice;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.onap.music.eelf.logging.EELFLoggerDelegate;
import org.onap.music.eelf.logging.format.AppMessages;
import org.onap.music.eelf.logging.format.ErrorSeverity;
import org.onap.music.eelf.logging.format.ErrorTypes;
import org.onap.music.lockingservice.ZooKeeperOperation;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A base class for protocol implementations which provides a number of higher level helper methods
* for working with ZooKeeper along with retrying synchronous operations if the connection to
* ZooKeeper closes such as {@link #retryOperation(ZooKeeperOperation)}
*
*/
class ProtocolSupport {
private EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(ProtocolSupport.class);
protected ZooKeeper zookeeper;
private AtomicBoolean closed = new AtomicBoolean(false);
private long retryDelay = 500L;
private int retryCount = 10;
private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
/**
* Closes this strategy and releases any ZooKeeper resources; but keeps the ZooKeeper instance
* open
*/
public void close() {
if (closed.compareAndSet(false, true)) {
doClose();
}
}
/**
* return zookeeper client instance
*
* @return zookeeper client instance
*/
public ZooKeeper getZookeeper() {
return zookeeper;
}
/**
* return the acl its using
*
* @return the acl.
*/
public List<ACL> getAcl() {
return acl;
}
/**
* set the acl
*
* @param acl the acl to set to
*/
public void setAcl(List<ACL> acl) {
this.acl = acl;
}
/**
* get the retry delay in milliseconds
*
* @return the retry delay
*/
public long getRetryDelay() {
return retryDelay;
}
/**
* Sets the time waited between retry delays
*
* @param retryDelay the retry delay
*/
public void setRetryDelay(long retryDelay) {
this.retryDelay = retryDelay;
}
/**
* Allow derived classes to perform some custom closing operations to release resources
*/
protected void doClose() {
throw new UnsupportedOperationException();
}
/**
* Perform the given operation, retrying if the connection fails
*
* @return object. it needs to be cast to the callee's expected return type.
* @param operation FILL IN
* @throws KeeperException FILL IN
* @throws InterruptedException FILL IN
*/
protected Object retryOperation(ZooKeeperOperation operation)
throws KeeperException, InterruptedException {
KeeperException exception = null;
for (int i = 0; i < retryCount; i++) {
try {
return operation.execute();
} catch (KeeperException.SessionExpiredException e) {
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.SESSIONEXPIRED+" for: " + zookeeper + " so reconnecting due to: " + e, ErrorSeverity.ERROR, ErrorTypes.SESSIONEXPIRED);
throw e;
} catch (KeeperException.ConnectionLossException e) {
if (exception == null) {
exception = e;
}
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.CONNCECTIVITYERROR, ErrorSeverity.ERROR, ErrorTypes.SESSIONEXPIRED);
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),"Attempt " + i + " failed with connection loss so attempting to reconnect: " + e);
retryDelay(i);
}
}
throw exception;
}
/**
* Ensures that the given path exists with no data, the current ACL and no flags
*
* @param path the lock path
*/
protected void ensurePathExists(String path) {
ensureExists(path, null, acl, CreateMode.PERSISTENT);
}
/**
* Ensures that the given path exists with the given data, ACL and flags
*
* @param path the lock path
* @param data the data
* @param acl list of ACLs applying to the path
* @param flags create mode flags
*/
protected void ensureExists(final String path, final byte[] data, final List<ACL> acl,
final CreateMode flags) {
try {
retryOperation(new ZooKeeperOperation() {
public boolean execute() throws KeeperException, InterruptedException {
Stat stat = zookeeper.exists(path, false);
if (stat != null) {
return true;
}
zookeeper.create(path, data, acl, flags);
return true;
}
});
} catch (KeeperException e) {
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.KEEPERERROR, ErrorSeverity.ERROR, ErrorTypes.LOCKINGERROR);
} catch (InterruptedException e) {
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.EXECUTIONINTERRUPTED, ErrorSeverity.ERROR, ErrorTypes.LOCKINGERROR);
}
}
/**
* Returns true if this protocol has been closed
*
* @return true if this protocol is closed
*/
protected boolean isClosed() {
return closed.get();
}
/**
* Performs a retry delay if this is not the first attempt
*
* @param attemptCount the number of the attempts performed so far
*/
protected void retryDelay(int attemptCount) {
if (attemptCount > 0) {
try {
Thread.sleep(attemptCount * retryDelay);
} catch (InterruptedException e) {
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),AppMessages.EXECUTIONINTERRUPTED, ErrorSeverity.ERROR, ErrorTypes.GENERALSERVICEERROR);
logger.error(EELFLoggerDelegate.errorLogger, e.getMessage(),"Thread failed to sleep: " + e);
Thread.currentThread().interrupt();
}
}
}
}
|
