Timer task is disposed due to restartTimerOnMaxSize leading to data loss

Steps to reproduce:

Submit data fast enough,
If timer is running while buffer is full ,timer task will be disposed.

We can see it in :

influxdb-client-java/client/src/main/java/com/influxdb/client/internal/flowable/FlowableBufferTimedFlushable.java

Lines 156 to 196 in d641b11

    
           public void onNext(T t) { 
        
               U b; 
        
               synchronized (this) { 
        
                   b = buffer; 
        
                   if (b == null) { 
        
                       return; 
        
                   } 
        
                   b.add(t); 
        
                   if (b.size() < maxSize) { 
        
                       return; 
        
                   } 
        
                   buffer = null; 
        
                   producerIndex++; 
        
               } 
        
               if (restartTimerOnMaxSize) { 
        
                   timer.dispose(); 
        
               } 
        
               fastPathOrderedEmitMax(b, false, this); 
        
               try { 
        
                   b = Objects.requireNonNull(bufferSupplier.get(), "The supplied buffer is null"); 
        
               } catch (Throwable e) { 
        
                   Exceptions.throwIfFatal(e); 
        
                   cancel(); 
        
                   downstream.onError(e); 
        
                   return; 
        
               } 
        
               synchronized (this) { 
        
                   buffer = b; 
        
                   consumerIndex++; 
        
               } 
        
               if (restartTimerOnMaxSize) { 
        
                   timer = w.schedulePeriodically(this, timespan, timespan, unit); 
        
               } 
        
           }

This bug will lead to #793. I think setting restartTimerOnMaxSize to false will be a good choice. Or we can add a flag when task is running and do not dispose the timer task.

Specifications:

Client Version: 6.7.0
InfluxDB Version: 2.6.1
JDK Version: 11
Platform: linux

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Timer task is disposed due to `restartTimerOnMaxSize` leading to data loss #806

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	public void onNext(T t) {
	U b;
	synchronized (this) {
	b = buffer;
	if (b == null) {
	return;
	}

	b.add(t);

	if (b.size() < maxSize) {
	return;
	}

	buffer = null;
	producerIndex++;
	}

	if (restartTimerOnMaxSize) {
	timer.dispose();
	}

	fastPathOrderedEmitMax(b, false, this);

	try {
	b = Objects.requireNonNull(bufferSupplier.get(), "The supplied buffer is null");
	} catch (Throwable e) {
	Exceptions.throwIfFatal(e);
	cancel();
	downstream.onError(e);
	return;
	}

	synchronized (this) {
	buffer = b;
	consumerIndex++;
	}
	if (restartTimerOnMaxSize) {
	timer = w.schedulePeriodically(this, timespan, timespan, unit);
	}
	}

Search code, repositories, users, issues, pull requests...

Timer task is disposed due to restartTimerOnMaxSize leading to data loss #806

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Timer task is disposed due to `restartTimerOnMaxSize` leading to data loss #806