fix deadlock

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/timer/TimerTask.java

@@ -21,7 +21,7 @@
 public abstract class TimerTask implements Runnable {
 
     protected final long delayMs;
-    private TimerTaskEntry timerTaskEntry = null;
+    private volatile TimerTaskEntry timerTaskEntry = null;
 
     protected TimerTask(long delayMs) {
         this.delayMs = delayMs;
@@ -34,16 +34,19 @@ public synchronized void cancel() {
         }
     }
 
-    synchronized void setTimerTaskEntry(TimerTaskEntry entry) {
-        // if this timerTask is already held by an existing timer task entry,
-        // we will remove such an entry first.
-        if (null != timerTaskEntry && timerTaskEntry != entry) {
-            timerTaskEntry.remove();
+    void setTimerTaskEntry(TimerTaskEntry entry) {
+        synchronized (this) {
+            // if this timerTask is already held by an existing timer task entry,
+            // we will remove such an entry first.
+            if (null != timerTaskEntry && timerTaskEntry != entry) {
+                timerTaskEntry.remove();
+            }
+            timerTaskEntry = entry;
         }
-        timerTaskEntry = entry;
+
     }
 
-    synchronized TimerTaskEntry getTimerTaskEntry() {
+    TimerTaskEntry getTimerTaskEntry() {
         return timerTaskEntry;
     }