mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-12-19 13:37:49 +08:00
dc784bf217
* listing files to convert to parquet * write parquet files * save logs into parquet files * pass by value * compact logs into parquet format * can skip existing files * refactor * refactor * fix compilation * when no partition found * refactor * add untested parquet file read * rename package * refactor * rename files * remove unused * add merged log read func * parquet wants to know the file size * rewind by time * pass in stop ts * add stop ts * adjust log * minor * adjust log * skip .parquet files when reading message logs * skip non message files * Update subscriber_record.go * send messages * skip message data with only ts * skip non log files * update parquet-go package * ensure a valid record type * add new field to a record type * Update read_parquet_to_log.go * fix parquet file name generation * separating reading parquet and logs * add key field * add skipped logs * use in memory cache * refactor * refactor * refactor * refactor, and change compact log * refactor * rename * refactor * fix format * prefix v to version directory
46 lines
1.2 KiB
Go
46 lines
1.2 KiB
Go
package broker
|
|
|
|
import (
|
|
"fmt"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
|
|
"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
func (b *MessageQueueBroker) genLogFlushFunc(t topic.Topic, p topic.Partition) log_buffer.LogFlushFuncType {
|
|
partitionDir := topic.PartitionDir(t, p)
|
|
|
|
return func(logBuffer *log_buffer.LogBuffer, startTime, stopTime time.Time, buf []byte) {
|
|
if len(buf) == 0 {
|
|
return
|
|
}
|
|
|
|
startTime, stopTime = startTime.UTC(), stopTime.UTC()
|
|
|
|
targetFile := fmt.Sprintf("%s/%s", partitionDir, startTime.Format(topic.TIME_FORMAT))
|
|
|
|
// TODO append block with more metadata
|
|
|
|
for {
|
|
if err := b.appendToFile(targetFile, buf); err != nil {
|
|
glog.V(0).Infof("metadata log write failed %s: %v", targetFile, err)
|
|
time.Sleep(737 * time.Millisecond)
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
|
|
atomic.StoreInt64(&logBuffer.LastFlushTsNs, stopTime.UnixNano())
|
|
|
|
b.accessLock.Lock()
|
|
defer b.accessLock.Unlock()
|
|
if localPartition := b.localTopicManager.GetLocalPartition(t, p); localPartition != nil {
|
|
localPartition.NotifyLogFlushed(logBuffer.LastFlushTsNs)
|
|
}
|
|
|
|
glog.V(0).Infof("flushing at %d to %s size %d", logBuffer.LastFlushTsNs, targetFile, len(buf))
|
|
}
|
|
}
|