Skip to content

Commit a6ecee0

Browse files
raju-saravananraju-saravanan
committed
ISSUE 568 : Synchronize Kerberos ticket renewal for non-renewable ticket
1 parent 5b7ecfa commit a6ecee0

File tree

2 files changed

+88
-22
lines changed

2 files changed

+88
-22
lines changed

‎common-auth/src/main/java/com/hortonworks/registries/auth/KerberosLogin.java

+40-6
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
/**
22
* Copyright 2017 Hortonworks.
3-
*
3+
* <p>
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
66
* You may obtain a copy of the License at
7-
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
7+
* <p>
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
* <p>
1010
* Unless required by applicable law or agreed to in writing, software
1111
* distributed under the License is distributed on an "AS IS" BASIS,
1212
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -30,6 +30,8 @@
3030
import java.util.Map;
3131
import java.util.Random;
3232
import java.util.Set;
33+
import java.util.concurrent.TimeUnit;
34+
import java.util.concurrent.locks.Lock;
3335

3436
/**
3537
* This class is responsible for logging in to Kerberos and refreshing credentials for
@@ -62,6 +64,9 @@ public class KerberosLogin extends AbstractLogin {
6264
private long minTimeBeforeRelogin = 1 * 60 * 1000;
6365
private String kinitCmd = "/usr/bin/kinit";
6466

67+
private Lock kerberosTGTRenewalLock;
68+
private long kerberosTGTRenewalTimeoutMS;
69+
6570
/**
6671
* Method to configure this instance with specific properties
6772
* @param loginContextName
@@ -85,6 +90,15 @@ public void configure(Map<String, ?> configs, final String loginContextName) {
8590
}
8691
}
8792

93+
public KerberosLogin() {
94+
95+
}
96+
97+
public KerberosLogin(Lock kerberosTGTRenewalLock, long kerberosTGTRenewalTimeoutMS) {
98+
this.kerberosTGTRenewalLock = kerberosTGTRenewalLock;
99+
this.kerberosTGTRenewalTimeoutMS = kerberosTGTRenewalTimeoutMS;
100+
}
101+
88102
/**
89103
* Method called once initially to login. It also starts the thread used
90104
* to periodically re-login to the Kerberos Authentication Server.
@@ -147,7 +161,7 @@ public void close() {
147161
}
148162
}
149163

150-
private void spawnReloginThread () {
164+
private void spawnReloginThread() {
151165
// Refresh the Ticket Granting Ticket (TGT) periodically. How often to refresh is determined by the
152166
// TGT's existing expiry date and the configured minTimeBeforeRelogin. For testing and development,
153167
// you can decrease the interval of expiration of tickets (for example, to 3 minutes) by running:
@@ -208,7 +222,7 @@ public void run() {
208222
}
209223
}
210224
try {
211-
reLogin();
225+
synchronizeReLogin();
212226
} catch (LoginException le) {
213227
log.error("Failed to refresh TGT: refresh thread exiting now.", le);
214228
return;
@@ -246,6 +260,26 @@ private KerberosTicket getTGT() {
246260
return null;
247261
}
248262

263+
private void synchronizeReLogin() throws LoginException {
264+
if (kerberosTGTRenewalLock != null) {
265+
try {
266+
if (kerberosTGTRenewalLock.tryLock(kerberosTGTRenewalTimeoutMS, TimeUnit.MILLISECONDS)) {
267+
try {
268+
reLogin();
269+
} finally {
270+
kerberosTGTRenewalLock.unlock();
271+
}
272+
} else {
273+
throw new LoginException("Timed out while waiting to acquire a lock for renewing Kerberos TGT");
274+
}
275+
} catch (InterruptedException e) {
276+
throw new LoginException("Error while acquiring lock for renewing Kerberos TGT : " + e.getMessage());
277+
}
278+
} else {
279+
reLogin();
280+
}
281+
}
282+
249283
/**
250284
* Re-login a principal. This method assumes that {@link #login()} has happened already.
251285
* @throws javax.security.auth.login.LoginException on a failure

‎schema-registry/client/src/main/java/com/hortonworks/registries/schemaregistry/client/SchemaRegistryClient.java

+48-16
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import com.hortonworks.registries.schemaregistry.errors.SchemaBranchAlreadyExistsException;
4949
import com.hortonworks.registries.schemaregistry.errors.SchemaBranchNotFoundException;
5050
import com.hortonworks.registries.schemaregistry.errors.SchemaNotFoundException;
51+
import com.hortonworks.registries.schemaregistry.exceptions.RegistryRetryableException;
5152
import com.hortonworks.registries.schemaregistry.serde.SerDesException;
5253
import com.hortonworks.registries.schemaregistry.serde.SnapshotDeserializer;
5354
import com.hortonworks.registries.schemaregistry.serde.SnapshotSerializer;
@@ -106,6 +107,9 @@
106107
import java.util.concurrent.ConcurrentHashMap;
107108
import java.util.concurrent.ExecutionException;
108109
import java.util.concurrent.TimeUnit;
110+
import java.util.concurrent.locks.Lock;
111+
import java.util.concurrent.locks.ReadWriteLock;
112+
import java.util.concurrent.locks.ReentrantReadWriteLock;
109113

110114
import static com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient.Configuration.DEFAULT_CONNECTION_TIMEOUT;
111115
import static com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient.Configuration.DEFAULT_READ_TIMEOUT;
@@ -157,11 +161,15 @@ public class SchemaRegistryClient implements ISchemaRegistryClient {
157161
private static final Set<Class<?>> SERIALIZER_INTERFACE_CLASSES = Sets.<Class<?>>newHashSet(SnapshotSerializer.class, PullSerializer.class);
158162
private static final String SEARCH_FIELDS = SCHEMA_REGISTRY_PATH + "/search/schemas/fields";
159163
private static Subject subject;
164+
private static ReadWriteLock kerberosSynchronizationLock = null;
165+
private static final long KERBEROS_SYNCHRONIZATION_TIMEOUT_MS = 180000;
160166

161167
static {
162168
String jaasConfigFile = System.getProperty("java.security.auth.login.config");
163169
if (jaasConfigFile != null && !jaasConfigFile.trim().isEmpty()) {
164-
KerberosLogin kerberosLogin = new KerberosLogin();
170+
kerberosSynchronizationLock = new ReentrantReadWriteLock(true);
171+
Lock kerberosTGTRenewalLock = kerberosSynchronizationLock.writeLock();
172+
KerberosLogin kerberosLogin = new KerberosLogin(kerberosTGTRenewalLock, KERBEROS_SYNCHRONIZATION_TIMEOUT_MS);
165173
kerberosLogin.configure(new HashMap<>(), REGISTY_CLIENT_JAAS_SECTION);
166174
try {
167175
subject = kerberosLogin.login().getSubject();
@@ -447,7 +455,7 @@ public void deleteSchema(String schemaName) throws SchemaNotFoundException {
447455
}
448456

449457
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path(String.format("%s", schemaName));
450-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
458+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
451459
@Override
452460
public Response run() {
453461
return target.request(MediaType.APPLICATION_JSON_TYPE).delete(Response.class);
@@ -516,7 +524,7 @@ public SchemaIdVersion uploadSchemaVersion(final String schemaBranchName,
516524
.bodyPart(streamDataBodyPart);
517525

518526
Entity<MultiPart> multiPartEntity = Entity.entity(multipartEntity, MediaType.MULTIPART_FORM_DATA);
519-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
527+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
520528
@Override
521529
public Response run() {
522530
return target.request().post(multiPartEntity, Response.class);
@@ -575,7 +583,7 @@ public void deleteSchemaVersion(SchemaVersionKey schemaVersionKey) throws Schema
575583

576584
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path(String.format("%s/versions/%s", schemaVersionKey
577585
.getSchemaName(), schemaVersionKey.getVersion()));
578-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
586+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
579587
@Override
580588
public Response run() {
581589
return target.request(MediaType.APPLICATION_JSON_TYPE).delete(Response.class);
@@ -606,7 +614,7 @@ private SchemaIdVersion doAddSchemaVersion(String schemaBranchName, String schem
606614

607615
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path(schemaName).path("/versions").queryParam("branch", schemaBranchName)
608616
.queryParam("disableCanonicalCheck", disableCanonicalCheck);
609-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
617+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
610618
@Override
611619
public Response run() {
612620
return target.request(MediaType.APPLICATION_JSON_TYPE).post(Entity.json(schemaVersion), Response.class);
@@ -759,7 +767,7 @@ public void startSchemaVersionReview(Long schemaVersionId) throws SchemaNotFound
759767
@Override
760768
public SchemaVersionMergeResult mergeSchemaVersion(Long schemaVersionId, boolean disableCanonicalCheck) throws SchemaNotFoundException, IncompatibleSchemaException {
761769
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path(schemaVersionId + "/merge").queryParam("disableCanonicalCheck", disableCanonicalCheck);
762-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
770+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
763771
@Override
764772
public Response run() {
765773
return target.request().post(null);
@@ -795,7 +803,7 @@ public SchemaVersionLifecycleStateMachineInfo getSchemaVersionLifecycleStateMach
795803
@Override
796804
public SchemaBranch createSchemaBranch(Long schemaVersionId, SchemaBranch schemaBranch) throws SchemaBranchAlreadyExistsException, SchemaNotFoundException {
797805
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path("versionsById/" + schemaVersionId + "/branch");
798-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
806+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
799807
@Override
800808
public Response run() {
801809
return target.request(MediaType.APPLICATION_JSON_TYPE).post(Entity.json(schemaBranch), Response.class);
@@ -819,7 +827,7 @@ public Response run() {
819827
@Override
820828
public Collection<SchemaBranch> getSchemaBranches(String schemaName) throws SchemaNotFoundException {
821829
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path(encode(schemaName) + "/branches");
822-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
830+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
823831
@Override
824832
public Response run() {
825833
return target.request().get();
@@ -839,7 +847,7 @@ public Response run() {
839847
@Override
840848
public void deleteSchemaBranch(Long schemaBranchId) throws SchemaBranchNotFoundException, InvalidSchemaBranchDeletionException {
841849
WebTarget target = currentSchemaRegistryTargets().schemasTarget.path("branch/" + schemaBranchId);
842-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
850+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
843851
@Override
844852
public Response run() {
845853
return target.request().delete();
@@ -868,7 +876,7 @@ private boolean transitionSchemaVersionState(Long schemaVersionId,
868876
byte[] transitionDetails) throws SchemaNotFoundException, SchemaLifecycleException {
869877

870878
WebTarget webTarget = currentSchemaRegistryTargets().schemaVersionsTarget.path(schemaVersionId + "/state/" + operationOrTargetState);
871-
Response response = Subject.doAs(subject, new PrivilegedAction<Response>() {
879+
Response response = synchronizeDoAction(new PrivilegedAction<Response>() {
872880
@Override
873881
public Response run() {
874882
return webTarget.request().post(Entity.text(transitionDetails));
@@ -919,7 +927,7 @@ public CompatibilityResult checkCompatibility(String schemaName, String toSchema
919927
public CompatibilityResult checkCompatibility(String schemaBranchName, String schemaName,
920928
String toSchemaText) throws SchemaNotFoundException {
921929
WebTarget webTarget = currentSchemaRegistryTargets().schemasTarget.path(encode(schemaName) + "/compatibility").queryParam("branch", schemaBranchName);
922-
String response = Subject.doAs(subject, new PrivilegedAction<String>() {
930+
String response = synchronizeDoAction(new PrivilegedAction<String>() {
923931
@Override
924932
public String run() {
925933
return webTarget.request().post(Entity.text(toSchemaText), String.class);
@@ -953,7 +961,7 @@ public String uploadFile(InputStream inputStream) {
953961
MultiPart multiPart = new MultiPart();
954962
BodyPart filePart = new StreamDataBodyPart("file", inputStream, "file");
955963
multiPart.bodyPart(filePart);
956-
return Subject.doAs(subject, new PrivilegedAction<String>() {
964+
return synchronizeDoAction(new PrivilegedAction<String>() {
957965
@Override
958966
public String run() {
959967
return currentSchemaRegistryTargets().filesTarget.request()
@@ -964,7 +972,7 @@ public String run() {
964972

965973
@Override
966974
public InputStream downloadFile(String fileId) {
967-
return Subject.doAs(subject, new PrivilegedAction<InputStream>() {
975+
return synchronizeDoAction(new PrivilegedAction<InputStream>() {
968976
@Override
969977
public InputStream run() {
970978
return currentSchemaRegistryTargets().filesTarget.path("download/" + encode(fileId))
@@ -1086,7 +1094,7 @@ private <T> T createInstance(SerDesInfo serDesInfo, boolean isSerializer) {
10861094
}
10871095

10881096
private <T> List<T> getEntities(WebTarget target, Class<T> clazz) {
1089-
String response = Subject.doAs(subject, new PrivilegedAction<String>() {
1097+
String response = synchronizeDoAction(new PrivilegedAction<String>() {
10901098
@Override
10911099
public String run() {
10921100
return target.request(MediaType.APPLICATION_JSON_TYPE).get(String.class);
@@ -1111,7 +1119,7 @@ private <T> List<T> parseResponseAsEntities(String response, Class<T> clazz) {
11111119
}
11121120

11131121
private <T> T postEntity(WebTarget target, Object json, Class<T> responseType) {
1114-
String response = Subject.doAs(subject, new PrivilegedAction<String>() {
1122+
String response = synchronizeDoAction(new PrivilegedAction<String>() {
11151123
@Override
11161124
public String run() {
11171125
return target.request(MediaType.APPLICATION_JSON_TYPE).post(Entity.json(json), String.class);
@@ -1130,7 +1138,7 @@ private <T> T readEntity(String response, Class<T> clazz) {
11301138
}
11311139

11321140
private <T> T getEntity(WebTarget target, Class<T> clazz) {
1133-
String response = Subject.doAs(subject, new PrivilegedAction<String>() {
1141+
String response = synchronizeDoAction(new PrivilegedAction<String>() {
11341142
@Override
11351143
public String run() {
11361144
return target.request(MediaType.APPLICATION_JSON_TYPE).get(String.class);
@@ -1140,6 +1148,30 @@ public String run() {
11401148
return readEntity(response, clazz);
11411149
}
11421150

1151+
private <T> T synchronizeDoAction(PrivilegedAction<T> action) {
1152+
if (kerberosSynchronizationLock == null) {
1153+
return Subject.doAs(subject, action);
1154+
} else {
1155+
Lock schemaRegistryLock = kerberosSynchronizationLock.readLock();
1156+
try {
1157+
if (schemaRegistryLock.tryLock(KERBEROS_SYNCHRONIZATION_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
1158+
T result;
1159+
try {
1160+
result = Subject.doAs(subject, action);
1161+
} finally {
1162+
schemaRegistryLock.unlock();
1163+
}
1164+
return result;
1165+
} else {
1166+
throw new RegistryRetryableException("Timed out while schema registry client was waiting for Kerberos TGT renewal");
1167+
}
1168+
} catch (InterruptedException e) {
1169+
throw new RegistryRetryableException("Error while schema registry client was waiting for Kerberos TGT renewal", e);
1170+
}
1171+
}
1172+
}
1173+
1174+
11431175
public static final class Configuration {
11441176
// we may want to remove schema.registry prefix from configuration properties as these are all properties
11451177
// given by client.

0 commit comments

Comments
 (0)